<commit_before># frozen_string_literal: true

require 'weakref'
require 'base64'
require 'securerandom'

module MessageBus
  # Like a hash, just does its best to stay in sync across the farm.
  # On boot all instances are blank, but they populate as various processes
  # fill it up.
  class DistributedCache
    DEFAULT_SITE_ID = 'default'

    class Manager
      CHANNEL_NAME ||= '/distributed_hash'.freeze

      attr_accessor :app_version

      def initialize(message_bus = nil)
        @subscribers = []
        @subscribed = false
        @lock = Mutex.new
        @message_bus = message_bus || MessageBus
      end

      def subscribers
        @subscribers
      end

      def process_message(message)
        i = @subscribers.length - 1

        payload = message.data

        while i >= 0
          begin
            current = @subscribers[i]

            next if payload["origin"] == current.identity
            next if current.key != payload["hash_key"]

            next if @app_version && payload["app_version"] != @app_version

            hash = current.hash(message.site_id || DEFAULT_SITE_ID)

            case payload["op"]
            when "set" then hash[payload["key"]] = payload["marshalled"] ? Marshal.load(Base64.decode64(payload["value"])) : payload["value"]
            when "delete" then hash.delete(payload["key"])
            when "clear"  then hash.clear
            end
          rescue WeakRef::RefError
            @subscribers.delete_at(i)
          ensure
            i -= 1
          end
        end
      end

      def ensure_subscribe!
        return if @subscribed

        @lock.synchronize do
          return if @subscribed

          @message_bus.subscribe(CHANNEL_NAME) do |message|
            @lock.synchronize do
              process_message(message)
            end
          end
          @subscribed = true
        end
      end

      def publish(hash, message)
        message[:origin] = hash.identity
        message[:hash_key] = hash.key
        message[:app_version] = @app_version if @app_version
        @message_bus.publish(CHANNEL_NAME, message, user_ids: [-1])
      end

      def set(hash, key, value)
        # special support for set
        marshal = (Set === value || Hash === value || Array === value)
        value = Base64.encode64(Marshal.dump(value)) if marshal
        publish(hash, op: :set, key: key, value: value, marshalled: marshal)
      end

      def delete(hash, key)
        publish(hash, op: :delete, key: key)
      end

      def clear(hash)
        publish(hash, op: :clear)
      end

      def register(hash)
        @lock.synchronize do
          @subscribers << WeakRef.new(hash)
        end
      end<commit_msg>Log when DistributedCache encounters an error when publishing.<commit_after># frozen_string_literal: true

require 'weakref'
require 'base64'
require 'securerandom'

module MessageBus
  # Like a hash, just does its best to stay in sync across the farm.
  # On boot all instances are blank, but they populate as various processes
  # fill it up.
  class DistributedCache
    DEFAULT_SITE_ID = 'default'

    class Manager
      CHANNEL_NAME ||= '/distributed_hash'.freeze

      attr_accessor :app_version

      def initialize(message_bus = nil)
        @subscribers = []
        @subscribed = false
        @lock = Mutex.new
        @message_bus = message_bus || MessageBus
      end

      def subscribers
        @subscribers
      end

      def process_message(message)
        i = @subscribers.length - 1

        payload = message.data

        while i >= 0
          begin
            current = @subscribers[i]

            next if payload["origin"] == current.identity
            next if current.key != payload["hash_key"]

            next if @app_version && payload["app_version"] != @app_version

            hash = current.hash(message.site_id || DEFAULT_SITE_ID)

            case payload["op"]
            when "set" then hash[payload["key"]] = payload["marshalled"] ? Marshal.load(Base64.decode64(payload["value"])) : payload["value"]
            when "delete" then hash.delete(payload["key"])
            when "clear"  then hash.clear
            end
          rescue WeakRef::RefError
            @subscribers.delete_at(i)
          ensure
            i -= 1
          end
        end
      end

      def ensure_subscribe!
        return if @subscribed

        @lock.synchronize do
          return if @subscribed

          @message_bus.subscribe(CHANNEL_NAME) do |message|
            @lock.synchronize do
              process_message(message)
            end
          end
          @subscribed = true
        end
      end

      def publish(hash, message)
        message[:origin] = hash.identity
        message[:hash_key] = hash.key
        message[:app_version] = @app_version if @app_version

        begin
          @message_bus.publish(CHANNEL_NAME, message, user_ids: [-1])
        rescue => e
          @message_bus.logger.warn("DistributedCache failed to publish: #{e.message}\n#{e.backtrace.join("\n")}")
          raise
        end
      end

      def set(hash, key, value)
        # special support for set
        marshal = (Set === value || Hash === value || Array === value)
        value = Base64.encode64(Marshal.dump(value)) if marshal
        publish(hash, op: :set, key: key, value: value, marshalled: marshal)
      end

      def delete(hash, key)
        publish(hash, op: :delete, key: key)
      end

      def clear(hash)
        publish(hash, op: :clear)
      end

      def register(hash)
        @lock.synchronize do
          @subscribers << WeakRef.new(hash)
        end
      end
{'fraction_non_alphanumeric': 0.09518754423213022, 'fraction_numerical': 0.004953998584571833, 'mean_word_length': 2.0101171458998937, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2120772', 'n_tokens_mistral': 1631, 'n_tokens_neox': 1554, 'n_words': 440}