Check projector active state using a specialized event publisher

This simplifies the projector logic as all projector state tracking is
now only checked by the `ActiveProjectorsEventPublisher`. Replay now
uses the normal `EventPublisher` so the logic to determine if an event
is allowed to be processed by a projector can be simplified.

Author: erikrozendaal

CHANGELOG.md

@@ -11,16 +11,25 @@
 
 - Also in preperation of rolling upgrades the enabled projectors are
   now optionally tracked in the `projector_states` table. Use the
-  `enable_projector_states` configuration flag to enable this feature.
-
-  Only projectors that are active or replaying will process
-  events. When an unknown projector is active all event processing
-  will be blocked. This allows deploying a new version while the old
-  version of the code is still running and processing events. Only
-  after executing `Sequent#activate_current_configuration!` will the
-  new code start processing events, while the old code is blocked. At
-  this moment traffic can be redirected from the old version to the
-  new version.
+  `enable_projector_states` configuration flag and
+  `ActiveProjectorsEventPublisher` as the `event_publisher` enable
+  this feature.
+
+  Only projectors that are active will process events during normal
+  operation. Events that are published while a newer projector is
+  activating or active will fail with a
+  `DifferentProjectorVersionIsActiveError` error. Event replay uses
+  the normal `EventPublisher` which does not check for projector
+  state, since replaying will write to temporary tables until replay
+  is finalized
+
+  When an unknown projector is active all event processing will be
+  blocked. This allows deploying a new version while the old version
+  of the code is still running and processing events. Only after
+  executing `Sequent#activate_current_configuration!` will the new
+  code start processing events, while the old code is blocked. At this
+  moment traffic can be redirected from the old version to the new
+  version.
 
   Activating the current configuration is part of the
   `migrate:offline` rake task, so the current deployment process (run

lib/sequent/core/active_projectors_event_publisher.rb

@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+module Sequent
+  module Core
+    class UnknownActiveProjectorError < ProjectorMigrationError; end
+    class DifferentProjectorVersionIsActiveError < ProjectorMigrationError; end
+
+    #
+    # Subtype of EventPublisher that only dispatches events to Projectors that are marked active in
+    # the projector_states table. Also fails if there are unknown active projectors. This allows
+    # upgrading to new code (with new projectors) without having to shutdown the old code first, as
+    # the old code will start failing as soon as the new code's Sequent configuration is activated
+    # (using `Sequent#activate_current_configuration!`).
+    #
+    class ActiveProjectorsEventPublisher < EventPublisher
+      def publish_events(events)
+        return if configuration.disable_event_handlers
+
+        ensure_no_unknown_active_projectors!
+
+        super
+      end
+
+      private
+
+      def handle_message(handler, event)
+        super if active?(handler)
+      end
+
+      def ensure_no_unknown_active_projectors!
+        expected_version = Sequent.migrations_class&.version
+        return if expected_version.nil?
+
+        registered_projectors = Migratable.projectors.to_set(&:name)
+        active_projectors = Projectors.projector_states
+          .values
+          .select { |s| s.active_version == expected_version }
+          .to_set(&:name)
+        unknown_active_projectors = active_projectors - registered_projectors
+        if unknown_active_projectors.present?
+          fail UnknownActiveProjectorError,
+               "cannot publish event when unknown projectors are active #{unknown_active_projectors}"
+        end
+      end
+
+      def active?(handler)
+        return true unless handler.is_a?(Projector)
+
+        version = Sequent.migrations_class&.version
+        return true if version.nil?
+
+        # Projector states are not enable so all projectors are considered active
+        return true unless Sequent.configuration.enable_projector_states
+
+        projector_state = Projectors.projector_states[handler.class.name]
+        return false if projector_state.nil?
+
+        return true if projector_state.activating_version.nil? && projector_state.active_version == version
+
+        # A different projector version is active, so we cannot write
+        # new events since they will not be properly propagated.
+        fail DifferentProjectorVersionIsActiveError,
+             "projector #{handler.class} version #{version} does not match state #{projector_state}"
+      end
+    end
+  end
+end

lib/sequent/core/core.rb

@@ -19,5 +19,6 @@
 require_relative 'workflow'
 require_relative 'random_uuid_generator'
 require_relative 'event_publisher'
+require_relative 'active_projectors_event_publisher'
 require_relative 'aggregate_snapshotter'
 require_relative 'projectors'

lib/sequent/core/event_publisher.rb

@@ -3,9 +3,6 @@
 module Sequent
   module Core
     class ProjectorMigrationError < RuntimeError; end
-    class UnknownActiveProjectorError < ProjectorMigrationError; end
-    class ReplayingProjectorMismatchError < ProjectorMigrationError; end
-    class DifferentProjectorVersionIsActiveError < ProjectorMigrationError; end
 
     #
     # EventPublisher ensures that, for every thread, events will be published
@@ -35,15 +32,6 @@ def message
       def publish_events(events)
         return if configuration.disable_event_handlers
 
-        ensure_no_unknown_active_projectors!
-
-        events.each { |event| events_queue.push(event) }
-        process_events
-      end
-
-      def replay_events(events)
-        ensure_only_replaying_projectors_subscribed!
-
         events.each { |event| events_queue.push(event) }
         process_events
       end
@@ -68,49 +56,20 @@ def process_event(event)
         Sequent.logger.debug("[EventPublisher] Publishing event #{event.class}") if Sequent.logger.debug?
 
         configuration.event_handlers.each do |handler|
-          handler.handle_message event
+          handle_message(handler, event)
         rescue ProjectorMigrationError
           raise
         rescue StandardError
           raise PublishEventError.new(handler.class, event)
         end
       end
 
-      def configuration
-        Sequent.configuration
+      def handle_message(handler, event)
+        handler.handle_message(event)
       end
 
-      def ensure_no_unknown_active_projectors!
-        return unless Sequent.configuration.enable_projector_states
-
-        expected_version = Sequent.migrations_class&.version
-        return if expected_version.nil?
-
-        registered_projectors = Migratable.projectors.to_set(&:name)
-        active_projectors = Projectors.projector_states
-          .values
-          .select { |s| s.active_version == expected_version }
-          .to_set(&:name)
-        unknown_active_projectors = active_projectors - registered_projectors
-        if unknown_active_projectors.present?
-          fail UnknownActiveProjectorError,
-               "cannot publish event when unknown projectors are active #{unknown_active_projectors}"
-        end
-      end
-
-      def ensure_only_replaying_projectors_subscribed!
-        return unless Sequent.configuration.enable_projector_states
-        return unless Sequent.migrations_class
-
-        registered_projectors = Migratable.projectors.to_set(&:name)
-        replaying_projectors = Projectors.projector_states
-          .values
-          .select { |state| state.replaying? || state.activating? }
-          .to_set(&:name)
-        if registered_projectors != replaying_projectors
-          fail ReplayingProjectorMismatchError,
-               "cannot replay event when different projectors are replaying #{replaying_projectors}"
-        end
+      def configuration
+        Sequent.configuration
       end
     end
   end

lib/sequent/core/event_store.rb

@@ -172,15 +172,19 @@ def events_exists?(aggregate_id)
       #
       # @param get_events lambda that returns the events cursor
       # @param on_progress lambda that gets called on substantial progress
-      def replay_events_from_cursor(get_events:, block_size: 2000,
-                                    on_progress: PRINT_PROGRESS)
+      def replay_events_from_cursor(
+        get_events:,
+        event_publisher: Sequent.configuration.event_publisher,
+        block_size: 2000,
+        on_progress: PRINT_PROGRESS
+      )
         progress = 0
         cursor = get_events.call
         ids_replayed = []
         cursor.each_row_batch(block_size:).each do |records|
           events = records.map(&method(:deserialize_event))
           Sequent.configuration.transaction_provider.transactional do
-            Sequent.configuration.event_publisher.replay_events(events)
+            event_publisher.publish_events(events)
           end
           records.each do |record|
             progress += 1

lib/sequent/core/projector.rb

@@ -112,10 +112,6 @@ def self.replay_persistor
         nil
       end
 
-      def dispatch_message(...)
-        super if active?
-      end
-
       def_delegators :@persistor, :execute_sql, :commit
 
       def update_record(record_class, *rest, &block)
@@ -198,32 +194,6 @@ def ensure_valid!
           EOS
         end
       end
-
-      def active?
-        version = Sequent.migrations_class&.version
-        return true if version.nil?
-
-        # Projector states are not enable so all projectors are considered active
-        return true unless Sequent.configuration.enable_projector_states
-
-        projector_state = Projectors.projector_states[self.class.name]
-        return false if projector_state.nil?
-
-        if projector_state.activating_version.present?
-          # Current projector version is activating, so we write to the temporary replay table.
-          return true if projector_state.activating_version == version
-        else
-          return true if projector_state.active_version == version
-
-          # Replaying the current version, so run this projector (it will write to a temporary table).
-          return true if projector_state.replaying_version == version
-        end
-
-        # A different projector version is active or activating, so we cannot write new events since they will
-        # not be properly propagated.
-        fail DifferentProjectorVersionIsActiveError,
-             "projector #{self.class} version #{version} does not match state #{projector_state}"
-      end
     end
   end
 end

lib/sequent/migrations/view_schema.rb

@@ -395,8 +395,9 @@ def replay_events(
         &on_progress
       )
         Sequent.configuration.event_store.replay_events_from_cursor(
-          block_size: 1000,
           get_events:,
+          event_publisher: Sequent::Core::EventPublisher.new,
+          block_size: 1000,
           on_progress:,
         )