Add support for running multiple orchestrators at once with sidekiq

examples/remote_executor.rb

@@ -62,8 +62,9 @@ def run_server
       run(world)
     end
 
-    def initialize_sidekiq_orchestrator
+    def initialize_sidekiq_orchestrator(id)
       ExampleHelper.create_world do |config|
+        config.id = id if id
         config.persistence_adapter = persistence_adapter
         config.connector           = connector
         config.executor            = ::Dynflow::Executors::Sidekiq::Core
@@ -164,8 +165,9 @@ def run_client(count)
   # assuming the remote executor was required as part of initialization
   # of the ActiveJob worker
   queues = Sidekiq.configure_server { |c| c.options[:queues] }
-  world = if queues.include?("dynflow_orchestrator")
-            RemoteExecutorExample.initialize_sidekiq_orchestrator
+  world = if queues.include?("dynflow_orchestrator") || (orchestrator_queue = queues.find { |q| q.start_with?('dynflow_orchestrator:') })
+            orchestrator_queue ||= 'dynflow_orchestrator'
+            RemoteExecutorExample.initialize_sidekiq_orchestrator(orchestrator_queue.split(':')[1])
           elsif (queues - ['dynflow_orchestrator']).any?
             RemoteExecutorExample.initialize_sidekiq_worker
           end

lib/dynflow/config.rb

@@ -79,6 +79,10 @@ def queues
       @queues ||= QueuesConfig.new
     end
 
+    config_attr :id, String do |world, config|
+      SecureRandom.uuid
+    end
+
     config_attr :logger_adapter, LoggerAdapters::Abstract do
       LoggerAdapters::Simple.new
     end

lib/dynflow/executors/abstract/core.rb

@@ -89,6 +89,9 @@ def execution_status(execution_plan_id = nil)
           {}
         end
 
+        def prune_orphaned_queues
+        end
+
         def heartbeat
           @logger.debug('Executor heartbeat')
           record = @world.coordinator.find_records(:id => @world.id,

lib/dynflow/executors/parallel.rb

@@ -61,6 +61,10 @@ def halt(execution_plan_id)
         @core.tell([:halt, execution_plan_id])
       end
 
+      def prune_orphaned_queues
+        @core.tell(:prune_orphaned_queues)
+      end
+
       def initialized
         @core_initialized
       end

lib/dynflow/executors/sidekiq/core.rb

@@ -6,6 +6,7 @@
 require 'dynflow/executors/sidekiq/worker_jobs'
 require 'dynflow/executors/sidekiq/redis_locking'
 
+require 'sidekiq/api'
 require 'sidekiq-reliable-fetch'
 Sidekiq.configure_server do |config|
   # Use semi-reliable fetch
@@ -28,20 +29,22 @@ class Core < Abstract::Core
         def initialize(world, *_args)
           @world = world
           @logger = world.logger
-          wait_for_orchestrator_lock
+          @subqueue = ::Sidekiq.configure_server { |c| c[:queues].find { |q| q.start_with? 'dynflow_orchestrator:' } }
+          @reply_queue = @subqueue || 'dynflow_orchestrator'
+          wait_for_orchestrator_lock unless @subqueue
           super
           schedule_update_telemetry
-          begin_startup!
+          begin_startup! unless @subqueue
         end
 
         def heartbeat
           super
-          reacquire_orchestrator_lock
+          reacquire_orchestrator_lock unless @subqueue
         end
 
         def start_termination(*args)
           super
-          release_orchestrator_lock
+          release_orchestrator_lock unless @subqueue
           finish_termination
         end
 
@@ -50,9 +53,20 @@ def execution_status(execution_plan_id = nil)
           {}
         end
 
+        def prune_orphaned_queues
+          active_world_ids = @world.coordinator.find_worlds(true).map(&:id)
+          ::Sidekiq::Queue.all.each do |queue|
+            next unless queue.name.start_with?('dynflow_orchestrator:')
+            world_id = queue.name.split(':', 2)[1]
+            next if active_world_ids.include?(world_id)
+            logger.info("Removing orphaned orchestrator queue #{queue.name}")
+            queue.clear
+          end
+        end
+
         def feed_pool(work_items)
           work_items.each do |new_work|
-            WorkerJobs::PerformWork.set(queue: suggest_queue(new_work)).perform_async(new_work)
+            WorkerJobs::PerformWork.set(queue: suggest_queue(new_work)).perform_async(new_work, @reply_queue)
           end
         end
 
@@ -68,6 +82,8 @@ def update_telemetry
         end
 
         def work_finished(work, delayed_events = nil)
+          return super if @subqueue
+
           # If the work item is sent in reply to a request from the current orchestrator, proceed
           if work.sender_orchestrator_id == @world.id
             super

lib/dynflow/executors/sidekiq/orchestrator_jobs.rb

@@ -7,8 +7,6 @@ module OrchestratorJobs
         # handles resposnes about finished work form the workers
         # or some event to handle on orchestrator side
         class WorkerDone < InternalJobBase
-          sidekiq_options queue: :dynflow_orchestrator
-
           # @param request_envelope [Dispatcher::Request] - request to handle on orchestrator side
           #   usually to start new execution or to pass some event
           def perform(work_item, delayed_events = nil)
@@ -23,8 +21,6 @@ def perform(work_item, delayed_events = nil)
 
         # handles setting up an event on orchestrator
         class PlanEvent < InternalJobBase
-          sidekiq_options queue: :dynflow_orchestrator
-
           # @param event_envelope [Dispatcher::Event] - request to handle on orchestrator side
           #   usually to start new execution or to pass some event
           def perform(execution_plan_id, step_id, event, time)
@@ -33,8 +29,6 @@ def perform(execution_plan_id, step_id, event, time)
         end
 
         class HandlePersistenceError < InternalJobBase
-          sidekiq_options queue: :dynflow_orchestrator
-
           # @param request_envelope [Dispatcher::Request] - request to handle on orchestrator side
           #   usually to start new execution or to pass some event
           def perform(error, work_item)

lib/dynflow/executors/sidekiq/worker_jobs.rb

@@ -5,18 +5,18 @@ module Executors
     module Sidekiq
       module WorkerJobs
         class PerformWork < InternalJobBase
-          def perform(work_item)
+          def perform(work_item, respond_to = 'dynflow_orchestrator')
             with_telemetry(work_item) do
               Executors.run_user_code do
                 work_item.world = Dynflow.process_world
                 work_item.execute
               end
             end
           rescue Errors::PersistenceError => e
-            OrchestratorJobs::HandlePersistenceError.perform_async(e, work_item)
+            OrchestratorJobs::HandlePersistenceError.set(queue: respond_to).perform_async(e, work_item)
           ensure
             step = work_item.step if work_item.is_a?(Director::StepWorkItem)
-            OrchestratorJobs::WorkerDone.perform_async(work_item, step && step.delayed_events)
+            OrchestratorJobs::WorkerDone.set(queue: respond_to).perform_async(work_item, step && step.delayed_events)
           end
 
           private

lib/dynflow/world.rb

@@ -23,7 +23,7 @@ def initialize(config)
       Dynflow::Telemetry.set_adapter @config.telemetry_adapter
       Dynflow::Telemetry.register_metrics!
 
-      @id                     = SecureRandom.uuid
+      @id                     = @config.id
       @logger_adapter         = @config.logger_adapter
       @clock                  = spawn_and_wait(Clock, 'clock', logger)
       @config.validate

lib/dynflow/world/invalidation.rb

@@ -142,6 +142,7 @@ def perform_validity_checks
         end
         pruned = connector.prune_undeliverable_envelopes(self)
         logger.error("Pruned #{pruned} undeliverable envelopes") unless pruned.zero?
+        executor.prune_orphaned_queues if executor
         world_invalidation_result.values.select { |result| result == :invalidated }.size
       end
 

test/bats/sidekiq-orchestrator.bats

@@ -176,3 +176,27 @@ teardown() {
   timeout 10 bundle exec ruby examples/remote_executor.rb client 1
   wait_for 1 1 grep -P 'dynflow: ExecutionPlan.*running >>.*stopped' "$(bg_output_file o2)"
 }
+
+@test "multi-orchestrator — jobs spread across two orchestrators using per-orchestrator queues" {
+  cd "$(get_project_root)"
+
+  # Start orchestrator 1 with subqueue dynflow_orchestrator:orch-a
+  uuid1=$(uuidgen)
+  run_background 'o1' bundle exec sidekiq -c 1 -r ./examples/remote_executor.rb -q dynflow_orchestrator:${uuid1}
+  wait_for 30 1 grep -q "World ${uuid1} started..." "$(bg_output_file o1)"
+
+  # Start orchestrator 2 with subqueue dynflow_orchestrator:orch-b
+  uuid2=$(uuidgen)
+  run_background 'o2' bundle exec sidekiq -c 1 -r ./examples/remote_executor.rb -q dynflow_orchestrator:${uuid2}
+  wait_for 30 1 grep -q "World ${uuid2} started..." "$(bg_output_file o2)"
+
+  # Start one worker
+  run_background 'w1' bundle exec sidekiq -r ./examples/remote_executor.rb -q default
+
+  # Trigger 10 jobs — with 2 orchestrators round-robin, each should get ~5
+  timeout 60 bundle exec ruby examples/remote_executor.rb client 10
+
+  # Assert both orchestrators handled at least one job
+  wait_for 5 1 grep -qP 'ExecutionPlan.*running >>.*stopped' "$(bg_output_file o1)"
+  wait_for 5 1 grep -qP 'ExecutionPlan.*running >>.*stopped' "$(bg_output_file o2)"
+}

test/sidekiq_multi_orchestrator_test.rb

@@ -0,0 +1,176 @@
+# frozen_string_literal: true
+
+require_relative 'test_helper'
+require 'mocha/minitest'
+
+require 'sidekiq'
+require 'sidekiq/api'
+require 'dynflow/executors/sidekiq/core'
+
+module Dynflow
+  module SidekiqMultiOrchestratorTest
+    describe Executors::Sidekiq::Core do
+      after do
+        ::Dynflow.instance_variable_set('@process_world', nil)
+      end
+
+      # @param subqueue [String, nil] the per-orchestrator queue (e.g. "dynflow_orchestrator:abc") this
+      #   "process" is configured to listen on, or nil to simulate the legacy shared-queue setup
+      def build_world(subqueue: nil, id: nil)
+        ::Sidekiq.stubs(:configure_server).returns(subqueue)
+        world = WorldFactory.create_world do |c|
+          c.executor = Executors::Sidekiq::Core
+          c.id = id if id
+        end
+        ::Dynflow.instance_variable_set('@process_world', world)
+        world
+      end
+
+      # digs out the actual Sidekiq::Core instance behind the actor reference, mirroring
+      # the get_director helper in test_helper.rb
+      def raw_core(world)
+        world.executor.instance_variable_get('@core').instance_variable_get('@core').context
+      end
+
+      describe 'reply queue selection' do
+        it 'falls back to the shared queue when no per-orchestrator queue is configured' do
+          Executors::Sidekiq::Core.any_instance.expects(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.expects(:begin_startup!)
+          world = build_world(subqueue: nil)
+          _(raw_core(world).instance_variable_get('@reply_queue')).must_equal 'dynflow_orchestrator'
+        end
+
+        it 'uses the per-orchestrator queue when one is configured' do
+          Executors::Sidekiq::Core.any_instance.expects(:wait_for_orchestrator_lock).never
+          Executors::Sidekiq::Core.any_instance.expects(:begin_startup!).never
+          world = build_world(subqueue: 'dynflow_orchestrator:some-world-id')
+          _(raw_core(world).instance_variable_get('@reply_queue')).must_equal 'dynflow_orchestrator:some-world-id'
+        end
+      end
+
+      describe 'heartbeat lock reacquisition' do
+        it 'reacquires the orchestrator lock when no subqueue is configured' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          Executors::Sidekiq::Core.any_instance.expects(:reacquire_orchestrator_lock)
+          world = build_world(subqueue: nil)
+          world.executor.core.ask!(:heartbeat)
+        end
+
+        it 'does not touch the lock when a subqueue is configured' do
+          Executors::Sidekiq::Core.any_instance.expects(:reacquire_orchestrator_lock).never
+          world = build_world(subqueue: 'dynflow_orchestrator:some-world-id')
+          world.executor.core.ask!(:heartbeat)
+        end
+      end
+
+      describe 'termination lock release' do
+        it 'releases the orchestrator lock when no subqueue is configured' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          Executors::Sidekiq::Core.any_instance.expects(:release_orchestrator_lock)
+          world = build_world(subqueue: nil)
+          world.terminate.wait(5)
+        end
+
+        it 'does not touch the lock when a subqueue is configured' do
+          Executors::Sidekiq::Core.any_instance.expects(:release_orchestrator_lock).never
+          world = build_world(subqueue: 'dynflow_orchestrator:some-world-id')
+          world.terminate.wait(5)
+        end
+      end
+
+      describe '#feed_pool' do
+        it 'tells workers to reply on the shared queue by default' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          work_item = stub('work_item', :queue => :default)
+          proxy = mock('perform_async_proxy')
+          proxy.expects(:perform_async).with(work_item, 'dynflow_orchestrator')
+          Executors::Sidekiq::WorkerJobs::PerformWork.expects(:set).with(:queue => :default).returns(proxy)
+          raw_core(world).feed_pool([work_item])
+        end
+
+        it 'tells workers to reply on the orchestrator-specific queue when configured' do
+          reply_queue = 'dynflow_orchestrator:some-world-id'
+          world = build_world(subqueue: reply_queue)
+          work_item = stub('work_item', :queue => :default)
+          proxy = mock('perform_async_proxy')
+          proxy.expects(:perform_async).with(work_item, reply_queue)
+          Executors::Sidekiq::WorkerJobs::PerformWork.expects(:set).with(:queue => :default).returns(proxy)
+          raw_core(world).feed_pool([work_item])
+        end
+      end
+
+      describe '#work_finished' do
+        it 'processes any work unconditionally when a subqueue is configured' do
+          world = build_world(subqueue: 'dynflow_orchestrator:some-world-id')
+          work = stub('work', :sender_orchestrator_id => 'someone-else')
+          Executors::Abstract::Core.any_instance.expects(:work_finished).with(work, nil)
+          raw_core(world).work_finished(work)
+        end
+
+        it 'processes work sent by this orchestrator even without a subqueue' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          work = stub('work', :sender_orchestrator_id => world.id)
+          Executors::Abstract::Core.any_instance.expects(:work_finished).with(work, nil)
+          raw_core(world).work_finished(work)
+        end
+
+        it 'defers foreign work to handle_unknown_work_item when no subqueue is configured' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          core = raw_core(world)
+          work = stub('work', :sender_orchestrator_id => 'someone-else')
+          Executors::Abstract::Core.any_instance.expects(:work_finished).never
+          core.expects(:handle_unknown_work_item).with(work)
+          core.work_finished(work)
+        end
+      end
+
+      describe 'drain / startup-complete recovery flow' do
+        it 'begin_startup! enqueues a DrainMarker job for this world' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::WorkerJobs::DrainMarker.expects(:perform_async).with('fixed-test-world-id')
+          build_world(subqueue: nil, id: 'fixed-test-world-id')
+        end
+
+        it 'DrainMarker hands off to StartupComplete for the same world id' do
+          Executors::Sidekiq::OrchestratorJobs::StartupComplete.expects(:perform_async).with('some-world-id')
+          Executors::Sidekiq::WorkerJobs::DrainMarker.new.perform('some-world-id')
+        end
+
+        it 'StartupComplete tells the matching orchestrator core that startup completed' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          world.executor.core.expects(:tell).with([:startup_complete])
+          Executors::Sidekiq::OrchestratorJobs::StartupComplete.new.perform(world.id)
+        end
+
+        it 'StartupComplete discards notifications meant for a different world' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          world.executor.core.expects(:tell).never
+          Executors::Sidekiq::OrchestratorJobs::StartupComplete.new.perform('a-different-world-id')
+        end
+
+        it '#startup_complete runs validity checks and clears recovery mode' do
+          Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock)
+          Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!)
+          world = build_world(subqueue: nil)
+          core = raw_core(world)
+          core.instance_variable_set('@recovery', true)
+          world.expects(:perform_validity_checks)
+          core.startup_complete
+          _(core.instance_variable_get('@recovery')).must_equal false
+        end
+      end
+    end
+  end
+end