diff --git a/Gemfile b/Gemfile index 28099303..2c4cc461 100644 --- a/Gemfile +++ b/Gemfile @@ -21,6 +21,7 @@ gem "dry-validation", "~> 1.10" gem "globalid", "~> 1.0" group :development, :test do + gem "activejob", "~> 7.0" gem "opentelemetry-api" gem "opentelemetry-sdk" end diff --git a/Gemfile.lock b/Gemfile.lock index e98a8170..d39b4204 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -11,21 +11,24 @@ PATH GEM remote: https://rubygems.org/ specs: - activesupport (8.1.1) + activejob (7.2.3.1) + activesupport (= 7.2.3.1) + globalid (>= 0.3.6) + activesupport (7.2.3.1) base64 + benchmark (>= 0.3) bigdecimal concurrent-ruby (~> 1.0, >= 1.3.1) connection_pool (>= 2.2.5) drb i18n (>= 1.6, < 2) - json logger (>= 1.4.2) - minitest (>= 5.1) + minitest (>= 5.1, < 6) securerandom (>= 0.3) tzinfo (~> 2.0, >= 2.0.5) - uri (>= 0.13.1) ast (2.4.3) base64 (0.3.0) + benchmark (0.5.0) bigdecimal (3.2.3) concurrent-ruby (1.3.5) connection_pool (2.4.1) @@ -179,7 +182,6 @@ GEM unicode-display_width (3.2.0) unicode-emoji (~> 4.1) unicode-emoji (4.2.0) - uri (1.1.1) zeitwerk (2.7.3) PLATFORMS @@ -187,6 +189,7 @@ PLATFORMS ruby DEPENDENCIES + activejob (~> 7.0) debug (~> 1.8) dry-validation (~> 1.10) globalid (~> 1.0) diff --git a/lib/ruby_reactor.rb b/lib/ruby_reactor.rb index bc2f4404..c23cf4a6 100644 --- a/lib/ruby_reactor.rb +++ b/lib/ruby_reactor.rb @@ -26,6 +26,13 @@ # sidekiq is optional, async features won't be available end +# Load active_job if available (for the ActiveJob async adapter) +begin + require "active_job" +rescue LoadError + # active_job is optional, only needed when using the ActiveJob adapter +end + loader = Zeitwerk::Loader.for_gem loader.inflector.inflect("api" => "API", "rspec" => "RSpec") loader.setup @@ -356,7 +363,17 @@ def self.reactor_storage_name(reactor_class) def self.start_sweeper! return unless configuration.sweeper_enabled - SidekiqWorkers::SweeperWorker.schedule_next + sweeper_job_class.schedule_next + end + + # The sweeper job class living alongside the configured `async_router` + # (e.g. `Adapters::Sidekiq::Router` -> `Adapters::Sidekiq::SweeperWorker`), + # so the chain is kicked through whichever backend is configured instead of + # a hardcoded Sidekiq class. + def self.sweeper_job_class + router = configuration.async_router + namespace = Object.const_get(router.name.rpartition("::").first) + namespace.const_get(:SweeperWorker) end # Run both recovery sweepers exactly once and return their counts. The diff --git a/lib/ruby_reactor/adapters/active_job/compat.rb b/lib/ruby_reactor/adapters/active_job/compat.rb new file mode 100644 index 00000000..38d6d233 --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/compat.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +module RubyReactor + module Adapters + module ActiveJob + # `Sidekiq::Worker` gives every job class `.perform_async` / `.perform_in` + # for free; ActiveJob only has `.perform_later`. Extending this onto an + # ActiveJob class normalizes its enqueue API to the same two class + # methods, so `RubyReactor::Worker` and `RubyReactor::SweeperJob` can keep + # calling `self.class.perform_in(...)` unchanged regardless of backend. + # Both methods return the job id (a String), matching what Sidekiq's + # native `perform_async`/`perform_in` return. + module Compat + def perform_async(*args) + perform_later(*args).job_id + end + + def perform_in(delay, *args) + set(wait: delay).perform_later(*args).job_id + end + end + end + end +end diff --git a/lib/ruby_reactor/adapters/active_job/map_collector_worker.rb b/lib/ruby_reactor/adapters/active_job/map_collector_worker.rb new file mode 100644 index 00000000..470a783a --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/map_collector_worker.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "active_job" + +module RubyReactor + module Adapters + module ActiveJob + class MapCollectorWorker < ::ActiveJob::Base + extend Compat + + queue_as { RubyReactor.configuration.queue_name } + + def perform(arguments) + RubyReactor::Map::Collector.perform(arguments) + end + end + end + end +end diff --git a/lib/ruby_reactor/adapters/active_job/map_element_worker.rb b/lib/ruby_reactor/adapters/active_job/map_element_worker.rb new file mode 100644 index 00000000..fbdd0a46 --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/map_element_worker.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "active_job" + +module RubyReactor + module Adapters + module ActiveJob + class MapElementWorker < ::ActiveJob::Base + extend Compat + + queue_as { RubyReactor.configuration.queue_name } + + def perform(arguments) + RubyReactor::Map::ElementExecutor.perform(arguments) + end + end + end + end +end diff --git a/lib/ruby_reactor/adapters/active_job/router.rb b/lib/ruby_reactor/adapters/active_job/router.rb new file mode 100644 index 00000000..73d63daa --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/router.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +module RubyReactor + module Adapters + module ActiveJob + class Router + # Identity-only payload: the worker rehydrates the live context from storage + # by (context_id, reactor_class_name). The caller already holds context_id, so + # there is no blob to deserialize here. + def self.perform_async(context_id, reactor_class_name = nil, intermediate_results: {}) + job_id = RubyReactor::Adapters::ActiveJob::Worker.perform_async(context_id, reactor_class_name) + RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, + execution_id: context_id) + end + + def self.perform_in(delay, context_id, reactor_class_name = nil, intermediate_results: {}) + job_id = RubyReactor::Adapters::ActiveJob::Worker.perform_in(delay, context_id, reactor_class_name) + RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, + execution_id: context_id) + end + + # rubocop:disable Metrics/ParameterLists + def self.perform_map_element_async(map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, + strict_ordering:, parent_context_id:, parent_reactor_class_name:, + step_name:, batch_size: nil, serialized_context: nil, fail_fast: nil) + job_id = RubyReactor::Adapters::ActiveJob::MapElementWorker.perform_async( + { + "map_id" => map_id, + "element_id" => element_id, + "index" => index, + "serialized_inputs" => serialized_inputs, + "reactor_class_info" => reactor_class_info, + "strict_ordering" => strict_ordering, + "parent_context_id" => parent_context_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "batch_size" => batch_size, + "serialized_context" => serialized_context, + "fail_fast" => fail_fast + } + ) + RubyReactor::AsyncResult.new(job_id: job_id) + end + + def self.perform_map_element_in(delay, map_id:, element_id:, index:, serialized_inputs:, + reactor_class_info:, strict_ordering:, parent_context_id:, + parent_reactor_class_name:, step_name:, batch_size: nil, + serialized_context: nil, fail_fast: nil) + job_id = RubyReactor::Adapters::ActiveJob::MapElementWorker.perform_in( + delay, + { + "map_id" => map_id, + "element_id" => element_id, + "index" => index, + "serialized_inputs" => serialized_inputs, + "reactor_class_info" => reactor_class_info, + "strict_ordering" => strict_ordering, + "parent_context_id" => parent_context_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "batch_size" => batch_size, + "serialized_context" => serialized_context, + "fail_fast" => fail_fast + } + ) + # Return an AsyncResult so RetryManager#handle_async_retry recognises the + # element was successfully requeued and yields a RetryQueuedResult. + RubyReactor::AsyncResult.new(job_id: job_id) + end + # rubocop:enable Metrics/ParameterLists + + # rubocop:disable Metrics/ParameterLists + def self.perform_map_collection_async(parent_context_id:, map_id:, parent_reactor_class_name:, step_name:, + strict_ordering:, timeout:) + job_id = RubyReactor::Adapters::ActiveJob::MapCollectorWorker.perform_async( + { + "parent_context_id" => parent_context_id, + "map_id" => map_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "strict_ordering" => strict_ordering, + "timeout" => timeout + } + ) + RubyReactor::AsyncResult.new(job_id: job_id) + end + # rubocop:enable Metrics/ParameterLists + end + end + end +end diff --git a/lib/ruby_reactor/adapters/active_job/sweeper_worker.rb b/lib/ruby_reactor/adapters/active_job/sweeper_worker.rb new file mode 100644 index 00000000..cf83cbb7 --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/sweeper_worker.rb @@ -0,0 +1,16 @@ +# frozen_string_literal: true + +require "active_job" + +module RubyReactor + module Adapters + module ActiveJob + class SweeperWorker < ::ActiveJob::Base + extend Compat + include RubyReactor::SweeperJob + + queue_as { RubyReactor.configuration.queue_name } + end + end + end +end diff --git a/lib/ruby_reactor/adapters/active_job/worker.rb b/lib/ruby_reactor/adapters/active_job/worker.rb new file mode 100644 index 00000000..581f23a7 --- /dev/null +++ b/lib/ruby_reactor/adapters/active_job/worker.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +require "active_job" + +module RubyReactor + module Adapters + module ActiveJob + # ActiveJob worker for executing RubyReactor reactors asynchronously. + # All resume/snooze/escalate logic lives in `RubyReactor::Worker` — this + # class only wires it to ActiveJob. + # + # Infra-failure retries only: reactor-specific contention/config errors + # are already rescued inside `RubyReactor::Worker#perform` (snoozed or + # escalated to `failed`) before they'd ever reach this `retry_on`. + class Worker < ::ActiveJob::Base + extend Compat + include RubyReactor::Worker + + queue_as { RubyReactor.configuration.queue_name } + retry_on StandardError, attempts: RubyReactor.configuration.job_retry_count + end + end + end +end diff --git a/lib/ruby_reactor/adapters/sidekiq/map_collector_worker.rb b/lib/ruby_reactor/adapters/sidekiq/map_collector_worker.rb new file mode 100644 index 00000000..70ae63e6 --- /dev/null +++ b/lib/ruby_reactor/adapters/sidekiq/map_collector_worker.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +module RubyReactor + module Adapters + module Sidekiq + class MapCollectorWorker + include ::Sidekiq::Worker + + def perform(arguments) + RubyReactor::Map::Collector.perform(arguments) + end + end + end + end +end diff --git a/lib/ruby_reactor/adapters/sidekiq/map_element_worker.rb b/lib/ruby_reactor/adapters/sidekiq/map_element_worker.rb new file mode 100644 index 00000000..40d79109 --- /dev/null +++ b/lib/ruby_reactor/adapters/sidekiq/map_element_worker.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +module RubyReactor + module Adapters + module Sidekiq + class MapElementWorker + include ::Sidekiq::Worker + + def perform(arguments) + RubyReactor::Map::ElementExecutor.perform(arguments) + end + end + end + end +end diff --git a/lib/ruby_reactor/adapters/sidekiq/router.rb b/lib/ruby_reactor/adapters/sidekiq/router.rb new file mode 100644 index 00000000..a6d2e50d --- /dev/null +++ b/lib/ruby_reactor/adapters/sidekiq/router.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +module RubyReactor + module Adapters + module Sidekiq + class Router + # Identity-only payload: the worker rehydrates the live context from storage + # by (context_id, reactor_class_name). The caller already holds context_id, so + # there is no blob to deserialize here. + def self.perform_async(context_id, reactor_class_name = nil, intermediate_results: {}) + job_id = RubyReactor::Adapters::Sidekiq::Worker.perform_async(context_id, reactor_class_name) + RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, + execution_id: context_id) + end + + def self.perform_in(delay, context_id, reactor_class_name = nil, intermediate_results: {}) + job_id = RubyReactor::Adapters::Sidekiq::Worker.perform_in(delay, context_id, reactor_class_name) + RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, + execution_id: context_id) + end + + # rubocop:disable Metrics/ParameterLists + def self.perform_map_element_async(map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, + strict_ordering:, parent_context_id:, parent_reactor_class_name:, + step_name:, batch_size: nil, serialized_context: nil, fail_fast: nil) + job_id = RubyReactor::Adapters::Sidekiq::MapElementWorker.perform_async( + { + "map_id" => map_id, + "element_id" => element_id, + "index" => index, + "serialized_inputs" => serialized_inputs, + "reactor_class_info" => reactor_class_info, + "strict_ordering" => strict_ordering, + "parent_context_id" => parent_context_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "batch_size" => batch_size, + "serialized_context" => serialized_context, + "fail_fast" => fail_fast + } + ) + RubyReactor::AsyncResult.new(job_id: job_id) + end + + def self.perform_map_element_in(delay, map_id:, element_id:, index:, serialized_inputs:, + reactor_class_info:, strict_ordering:, parent_context_id:, + parent_reactor_class_name:, step_name:, batch_size: nil, + serialized_context: nil, fail_fast: nil) + job_id = RubyReactor::Adapters::Sidekiq::MapElementWorker.perform_in( + delay, + { + "map_id" => map_id, + "element_id" => element_id, + "index" => index, + "serialized_inputs" => serialized_inputs, + "reactor_class_info" => reactor_class_info, + "strict_ordering" => strict_ordering, + "parent_context_id" => parent_context_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "batch_size" => batch_size, + "serialized_context" => serialized_context, + "fail_fast" => fail_fast + } + ) + # Return an AsyncResult so RetryManager#handle_async_retry recognises the + # element was successfully requeued and yields a RetryQueuedResult. + RubyReactor::AsyncResult.new(job_id: job_id) + end + # rubocop:enable Metrics/ParameterLists + + # rubocop:disable Metrics/ParameterLists + def self.perform_map_collection_async(parent_context_id:, map_id:, parent_reactor_class_name:, step_name:, + strict_ordering:, timeout:) + job_id = RubyReactor::Adapters::Sidekiq::MapCollectorWorker.perform_async( + { + "parent_context_id" => parent_context_id, + "map_id" => map_id, + "parent_reactor_class_name" => parent_reactor_class_name, + "step_name" => step_name, + "strict_ordering" => strict_ordering, + "timeout" => timeout + } + ) + RubyReactor::AsyncResult.new(job_id: job_id) + end + # rubocop:enable Metrics/ParameterLists + end + end + end +end diff --git a/lib/ruby_reactor/adapters/sidekiq/sweeper_worker.rb b/lib/ruby_reactor/adapters/sidekiq/sweeper_worker.rb new file mode 100644 index 00000000..f4b0a060 --- /dev/null +++ b/lib/ruby_reactor/adapters/sidekiq/sweeper_worker.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "sidekiq" + +module RubyReactor + module Adapters + module Sidekiq + class SweeperWorker + include ::Sidekiq::Worker + include RubyReactor::SweeperJob + + # retry: false — the sweep is idempotent and self-rescheduling, so a failed + # tick must not pile up Sidekiq retries; the next tick (or a super_fetch + # recovery) re-runs it anyway. + sidekiq_options retry: false, queue: RubyReactor.configuration.queue_name + end + end + end +end diff --git a/lib/ruby_reactor/adapters/sidekiq/worker.rb b/lib/ruby_reactor/adapters/sidekiq/worker.rb new file mode 100644 index 00000000..27a4c2d8 --- /dev/null +++ b/lib/ruby_reactor/adapters/sidekiq/worker.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require "sidekiq" + +module RubyReactor + module Adapters + module Sidekiq + # Sidekiq worker for executing RubyReactor reactors asynchronously + # with non-blocking retry capabilities. All resume/snooze/escalate logic + # lives in `RubyReactor::Worker` — this class only wires it to Sidekiq. + class Worker + include ::Sidekiq::Worker + include RubyReactor::Worker + + # Enable Sidekiq retries for infrastructure failures only + sidekiq_options retry: RubyReactor.configuration.job_retry_count, dead: false, + queue: RubyReactor.configuration.queue_name + + sidekiq_retries_exhausted do |_, _exception| + # Handle infrastructure failures (network, Redis, etc.) + end + end + end + end +end diff --git a/lib/ruby_reactor/configuration.rb b/lib/ruby_reactor/configuration.rb index cf16ad8c..b2250811 100644 --- a/lib/ruby_reactor/configuration.rb +++ b/lib/ruby_reactor/configuration.rb @@ -7,13 +7,23 @@ module RubyReactor class Configuration include Singleton - attr_writer :sidekiq_queue, :sidekiq_retry_count, :logger, :async_router, + attr_writer :queue_name, :job_retry_count, :logger, :async_router, :lock_snooze_base_delay, :lock_snooze_jitter, :lock_snooze_max_attempts, :middlewares, :context_ttl, :context_lock_ttl, :checkpoint_min_interval, :sweeper_enabled, :sweeper_interval, :sweeper_limit + def queue_name + @queue_name ||= :default + end + + # Deprecated alias for `queue_name` — kept so existing Sidekiq-only configs + # don't break. def sidekiq_queue - @sidekiq_queue ||= :default + queue_name + end + + def sidekiq_queue=(value) + self.queue_name = value end # Retention TTL (seconds) for a stored reactor context. Storage is @@ -79,8 +89,18 @@ def context_lock_ttl @context_lock_ttl ||= 60 end + def job_retry_count + @job_retry_count ||= 3 + end + + # Deprecated alias for `job_retry_count` — kept so existing Sidekiq-only + # configs don't break. def sidekiq_retry_count - @sidekiq_retry_count ||= 3 + job_retry_count + end + + def sidekiq_retry_count=(value) + self.job_retry_count = value end # Base seconds the Sidekiq worker waits before re-checking a contended lock. @@ -104,7 +124,7 @@ def logger end def async_router - @async_router ||= RubyReactor::SidekiqAdapter + @async_router ||= RubyReactor::Adapters::Sidekiq::Router end def storage diff --git a/lib/ruby_reactor/map/element_executor.rb b/lib/ruby_reactor/map/element_executor.rb index 5b461b8d..61ea8dc3 100644 --- a/lib/ruby_reactor/map/element_executor.rb +++ b/lib/ruby_reactor/map/element_executor.rb @@ -51,7 +51,7 @@ def self.perform_element(arguments) # The element already runs inside its own background worker, so any async # steps (and async retries) must execute inline here rather than handing # off to a detached Worker that would escape map result/counter tracking. - # This mirrors SidekiqWorkers::Worker, which sets the same flag. + # This mirrors RubyReactor::Worker, which sets the same flag. context.inline_async_execution = true storage = RubyReactor.configuration.storage_adapter diff --git a/lib/ruby_reactor/rspec.rb b/lib/ruby_reactor/rspec.rb index a9d35a87..de4de277 100644 --- a/lib/ruby_reactor/rspec.rb +++ b/lib/ruby_reactor/rspec.rb @@ -3,6 +3,8 @@ require_relative "rspec/helpers" require_relative "rspec/matchers" require_relative "rspec/sidekiq_helpers" +require_relative "rspec/active_job_helpers" +require_relative "rspec/async_test_helpers" require_relative "rspec/storage_reset" require_relative "rspec/test_subject" @@ -44,6 +46,8 @@ def self.prepare_example! ::Sidekiq::Worker.clear_all end + ::ActiveJob::Base.queue_adapter.enqueued_jobs.clear if AsyncTestHelpers.active_job_testing? + adapter = ::RubyReactor.configuration.storage_adapter adapter.reset! if adapter.respond_to?(:reset!) diff --git a/lib/ruby_reactor/rspec/active_job_helpers.rb b/lib/ruby_reactor/rspec/active_job_helpers.rb new file mode 100644 index 00000000..9848e9dc --- /dev/null +++ b/lib/ruby_reactor/rspec/active_job_helpers.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +module RubyReactor + module RSpec + # ActiveJob counterpart to `SidekiqHelpers`, used internally by + # `AsyncTestHelpers` to drain the ActiveJob `:test` queue adapter. Only + # the static methods are needed (`AsyncTestHelpers` dispatches to them); + # `Rails::ActiveJob::TestHelper` already covers most spec-author-facing + # assertions, so this module is not auto-included into examples. + module ActiveJobHelpers + PendingJob = Struct.new(:job_class, :raw) do + def perform! + ::ActiveJob::Base.queue_adapter.enqueued_jobs.delete(raw) + job_class.new(*raw[:args]).perform_now + end + + def args + raw[:args] + end + end + + def self.test_adapter? + defined?(::ActiveJob::Base) && + ::ActiveJob::Base.queue_adapter.is_a?(::ActiveJob::QueueAdapters::TestAdapter) + end + + # Drain every queued job in the ActiveJob `:test` adapter until the + # queue is empty. Recursive — handles jobs that re-enqueue themselves + # (e.g. ordered_lock snoozes) and job chains that queue additional jobs. + def self.drain_async_jobs(max_iterations: 100) + return unless test_adapter? + + queue = ::ActiveJob::Base.queue_adapter.enqueued_jobs + + max_iterations.times do + break if queue.empty? + + queue.dup.each do |job| + queue.delete(job) + job[:job].new(*job[:args]).perform_now + end + end + end + + def self.pending_async_jobs + return [] unless test_adapter? + + ::ActiveJob::Base.queue_adapter.enqueued_jobs.map { |raw| PendingJob.new(raw[:job], raw) } + end + end + end +end diff --git a/lib/ruby_reactor/rspec/async_test_helpers.rb b/lib/ruby_reactor/rspec/async_test_helpers.rb new file mode 100644 index 00000000..74d0598c --- /dev/null +++ b/lib/ruby_reactor/rspec/async_test_helpers.rb @@ -0,0 +1,41 @@ +# frozen_string_literal: true + +module RubyReactor + module RSpec + # Single entry point `TestSubject` uses to detect and drain whichever + # async testing framework — Sidekiq::Testing fake mode or ActiveJob's + # `:test` queue adapter — is currently active, so the job-processing gate + # isn't hardcoded to one background processor. + module AsyncTestHelpers + def self.active? + sidekiq_testing? || active_job_testing? + end + + def self.sidekiq_testing? + defined?(::Sidekiq::Testing) && ::Sidekiq::Testing.fake? + end + + def self.active_job_testing? + ActiveJobHelpers.test_adapter? + end + + def self.drain_async_jobs(max_iterations: 100) + if sidekiq_testing? + SidekiqHelpers.drain_async_jobs(max_iterations: max_iterations) + elsif active_job_testing? + ActiveJobHelpers.drain_async_jobs(max_iterations: max_iterations) + end + end + + def self.pending_async_jobs + if sidekiq_testing? + SidekiqHelpers.pending_async_jobs + elsif active_job_testing? + ActiveJobHelpers.pending_async_jobs + else + [] + end + end + end + end +end diff --git a/lib/ruby_reactor/rspec/sidekiq_helpers.rb b/lib/ruby_reactor/rspec/sidekiq_helpers.rb index 2e51daf1..10b8621d 100644 --- a/lib/ruby_reactor/rspec/sidekiq_helpers.rb +++ b/lib/ruby_reactor/rspec/sidekiq_helpers.rb @@ -36,9 +36,9 @@ def args def self.worker_classes @worker_classes ||= [ - RubyReactor::SidekiqWorkers::Worker, - RubyReactor::SidekiqWorkers::MapElementWorker, - RubyReactor::SidekiqWorkers::MapCollectorWorker + RubyReactor::Adapters::Sidekiq::Worker, + RubyReactor::Adapters::Sidekiq::MapElementWorker, + RubyReactor::Adapters::Sidekiq::MapCollectorWorker ] end diff --git a/lib/ruby_reactor/rspec/test_subject.rb b/lib/ruby_reactor/rspec/test_subject.rb index 3acfcf5c..698e296e 100644 --- a/lib/ruby_reactor/rspec/test_subject.rb +++ b/lib/ruby_reactor/rspec/test_subject.rb @@ -191,9 +191,9 @@ def run end @run_result = nil - if @process_jobs && defined?(Sidekiq::Testing) - # Ensure SidekiqAdapter is used to capture jobs in fake mode - allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + if @process_jobs && AsyncTestHelpers.sidekiq_testing? + # Ensure the Sidekiq router is used to capture jobs in fake mode + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) # Avoid nesting error which happens in Sidekiq 7+ if a mode is already set begin @@ -203,6 +203,10 @@ def run rescue Sidekiq::Testing::TestModeAlreadySetError @run_result = execution_class.run(@inputs) end + elsif @process_jobs && AsyncTestHelpers.active_job_testing? + # Ensure the ActiveJob router is used to capture jobs in the :test adapter + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::ActiveJob::Router) + @run_result = execution_class.run(@inputs) else @run_result = execution_class.run(@inputs) end @@ -250,7 +254,7 @@ def result skipped_result(ctx) when "running" # Try to determine if it is truly running or if we just missed the completion - if @process_jobs && defined?(Sidekiq::Testing) + if @process_jobs && AsyncTestHelpers.active? # Force one more check process_pending_jobs # Reload status @@ -357,7 +361,7 @@ def resume(payload: {}, step: nil) @reactor_instance.continue(payload: payload, step_name: step_name) # Process any pending async jobs - process_pending_jobs if @process_jobs && defined?(Sidekiq::Testing) + process_pending_jobs if @process_jobs && AsyncTestHelpers.active? # Reload the reactor instance to get updated state @reactor_instance = @reactor_class.find(@reactor_instance.context.context_id) @@ -430,9 +434,9 @@ def ensure_executed! private def process_pending_jobs - return unless defined?(Sidekiq::Testing) + return unless AsyncTestHelpers.active? - SidekiqHelpers.drain_async_jobs + AsyncTestHelpers.drain_async_jobs @reactor_instance = @reactor_class.find(@reactor_instance.context.context_id) end diff --git a/lib/ruby_reactor/sidekiq_adapter.rb b/lib/ruby_reactor/sidekiq_adapter.rb deleted file mode 100644 index 86ed5f7e..00000000 --- a/lib/ruby_reactor/sidekiq_adapter.rb +++ /dev/null @@ -1,87 +0,0 @@ -# frozen_string_literal: true - -module RubyReactor - class SidekiqAdapter - # Identity-only payload: the worker rehydrates the live context from storage - # by (context_id, reactor_class_name). The caller already holds context_id, so - # there is no blob to deserialize here. - def self.perform_async(context_id, reactor_class_name = nil, intermediate_results: {}) - job_id = SidekiqWorkers::Worker.perform_async(context_id, reactor_class_name) - RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, - execution_id: context_id) - end - - def self.perform_in(delay, context_id, reactor_class_name = nil, intermediate_results: {}) - job_id = SidekiqWorkers::Worker.perform_in(delay, context_id, reactor_class_name) - RubyReactor::AsyncResult.new(job_id: job_id, intermediate_results: intermediate_results, - execution_id: context_id) - end - - # rubocop:disable Metrics/ParameterLists - def self.perform_map_element_async(map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, - strict_ordering:, parent_context_id:, parent_reactor_class_name:, step_name:, - batch_size: nil, serialized_context: nil, fail_fast: nil) - job_id = RubyReactor::SidekiqWorkers::MapElementWorker.perform_async( - { - "map_id" => map_id, - "element_id" => element_id, - "index" => index, - "serialized_inputs" => serialized_inputs, - "reactor_class_info" => reactor_class_info, - "strict_ordering" => strict_ordering, - "parent_context_id" => parent_context_id, - "parent_reactor_class_name" => parent_reactor_class_name, - "step_name" => step_name, - "batch_size" => batch_size, - "serialized_context" => serialized_context, - "fail_fast" => fail_fast - } - ) - RubyReactor::AsyncResult.new(job_id: job_id) - end - - def self.perform_map_element_in(delay, map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, - strict_ordering:, parent_context_id:, parent_reactor_class_name:, step_name:, - batch_size: nil, serialized_context: nil, fail_fast: nil) - job_id = RubyReactor::SidekiqWorkers::MapElementWorker.perform_in( - delay, - { - "map_id" => map_id, - "element_id" => element_id, - "index" => index, - "serialized_inputs" => serialized_inputs, - "reactor_class_info" => reactor_class_info, - "strict_ordering" => strict_ordering, - "parent_context_id" => parent_context_id, - "parent_reactor_class_name" => parent_reactor_class_name, - "step_name" => step_name, - "batch_size" => batch_size, - "serialized_context" => serialized_context, - "fail_fast" => fail_fast - } - ) - # Return an AsyncResult so RetryManager#handle_async_retry recognises the - # element was successfully requeued and yields a RetryQueuedResult. - RubyReactor::AsyncResult.new(job_id: job_id) - end - # rubocop:enable Metrics/ParameterLists - - # rubocop:disable Metrics/ParameterLists - def self.perform_map_collection_async(parent_context_id:, map_id:, parent_reactor_class_name:, step_name:, - strict_ordering:, timeout:) - job_id = RubyReactor::SidekiqWorkers::MapCollectorWorker.perform_async( - { - "parent_context_id" => parent_context_id, - "map_id" => map_id, - "parent_reactor_class_name" => parent_reactor_class_name, - "step_name" => step_name, - "strict_ordering" => strict_ordering, - "timeout" => timeout - } - ) - RubyReactor::AsyncResult.new(job_id: job_id) - end - - # rubocop:enable Metrics/ParameterLists - end -end diff --git a/lib/ruby_reactor/sidekiq_workers/map_collector_worker.rb b/lib/ruby_reactor/sidekiq_workers/map_collector_worker.rb deleted file mode 100644 index f54c7092..00000000 --- a/lib/ruby_reactor/sidekiq_workers/map_collector_worker.rb +++ /dev/null @@ -1,13 +0,0 @@ -# frozen_string_literal: true - -module RubyReactor - module SidekiqWorkers - class MapCollectorWorker - include ::Sidekiq::Worker - - def perform(arguments) - RubyReactor::Map::Collector.perform(arguments) - end - end - end -end diff --git a/lib/ruby_reactor/sidekiq_workers/map_element_worker.rb b/lib/ruby_reactor/sidekiq_workers/map_element_worker.rb deleted file mode 100644 index e100d2be..00000000 --- a/lib/ruby_reactor/sidekiq_workers/map_element_worker.rb +++ /dev/null @@ -1,13 +0,0 @@ -# frozen_string_literal: true - -module RubyReactor - module SidekiqWorkers - class MapElementWorker - include ::Sidekiq::Worker - - def perform(arguments) - RubyReactor::Map::ElementExecutor.perform(arguments) - end - end - end -end diff --git a/lib/ruby_reactor/sidekiq_workers/sweeper_worker.rb b/lib/ruby_reactor/sidekiq_workers/sweeper_worker.rb deleted file mode 100644 index e4964191..00000000 --- a/lib/ruby_reactor/sidekiq_workers/sweeper_worker.rb +++ /dev/null @@ -1,73 +0,0 @@ -# frozen_string_literal: true - -require "sidekiq" -require "securerandom" - -module RubyReactor - module SidekiqWorkers - # Self-rescheduling recovery tick. Each run sweeps both the top-level reactor - # sweeper and the map sweeper, then schedules the next tick — a perpetual - # chain the host kicks once via `RubyReactor.start_sweeper!`. - # - # super_fetch safety. Sidekiq Enterprise `super_fetch` reliably re-runs a job - # whose worker died mid-execution. For a self-rescheduling chain that is a - # hazard: a tick can crash AFTER enqueuing its successor but BEFORE acking, so - # super_fetch recovers the crashed tick *alongside* the successor it already - # scheduled — the chain forks and then doubles every interval. We therefore do - # NOT rely on "exactly one job exists". The next tick is claimed by a - # per-time-window lock: every duplicate computes the SAME target window and - # only one wins the claim, so recovered/duplicated ticks collapse back to a - # single chain. The claim lock is never released — it simply expires — so no - # delete can race two duplicates into both winning. - class SweeperWorker - include ::Sidekiq::Worker - - # retry: false — the sweep is idempotent and self-rescheduling, so a failed - # tick must not pile up Sidekiq retries; the next tick (or a super_fetch - # recovery) re-runs it anyway. - sidekiq_options retry: false, queue: RubyReactor.configuration.sidekiq_queue - - def perform - config = RubyReactor.configuration - return unless config.sweeper_enabled - - run_sweeps(config) - ensure - # Always chain forward (unless disabled), even after an error above, so a - # single bad sweep can't kill recovery. The window lock keeps this from - # forking under super_fetch. - self.class.schedule_next if RubyReactor.configuration.sweeper_enabled - end - - def run_sweeps(config) - RubyReactor::Sweeper.run_once(limit: config.sweeper_limit) - RubyReactor::Map::Sweeper.run_once(limit: config.sweeper_limit) - rescue StandardError => e - config.logger.error("RubyReactor::SweeperWorker sweep failed: #{e.class}: #{e.message}") - end - - # Enqueue the next tick for the upcoming time window, claiming that window - # so concurrent/duplicate/recovered ticks produce exactly one successor. - # Idempotent: also safe to call from `start_sweeper!` on every process boot. - def self.schedule_next - interval = RubyReactor.configuration.sweeper_interval - window = (Time.now.to_i / interval) + 1 - - lock = RubyReactor::Lock.new( - "sweeper:window:#{window}", - owner: SecureRandom.uuid, - ttl: interval * 2, # outlive the window; expires on its own (never released) - wait: 0, - auto_extend: false - ) - lock.acquire # raises AcquisitionError if this window is already claimed - - delay = (window * interval) - Time.now.to_i - perform_in([delay, 1].max) - rescue RubyReactor::Lock::AcquisitionError - # Another tick already scheduled this window — collapse the duplicate. - nil - end - end - end -end diff --git a/lib/ruby_reactor/sidekiq_workers/worker.rb b/lib/ruby_reactor/sidekiq_workers/worker.rb deleted file mode 100644 index 38352a2e..00000000 --- a/lib/ruby_reactor/sidekiq_workers/worker.rb +++ /dev/null @@ -1,222 +0,0 @@ -# frozen_string_literal: true - -require "sidekiq" - -module RubyReactor - module SidekiqWorkers - # Sidekiq worker for executing RubyReactor reactors asynchronously - # with non-blocking retry capabilities - class Worker - include ::Sidekiq::Worker - - # Enable Sidekiq retries for infrastructure failures only - sidekiq_options retry: RubyReactor.configuration.sidekiq_retry_count, dead: false, - queue: RubyReactor.configuration.sidekiq_queue - - sidekiq_retries_exhausted do |_, exception| - # Handle infrastructure failures (network, Redis, etc.) - end - - # Identity-only payload: storage is the source of truth. Rehydrate the live - # context from storage by id, then resume. A nil read means the context was - # swept, expired, or already terminal-and-collected — nothing to resume. - def perform(context_id, reactor_class_name = nil, snooze_count = 0) - # Normalize so a nil/omitted name resolves to the same storage key the - # enqueue path wrote (always via reactor_storage_name). Without this a - # nil here builds "reactor::context:" and misses the stored - # "reactor:AnonymousReactor:context:", silently no-op'ing. - reactor_class_name ||= RubyReactor.reactor_storage_name(nil) - data = RubyReactor.configuration.storage_adapter.retrieve_context(context_id, reactor_class_name) - return if data.nil? - - begin - context = ContextSerializer.deserialize_hash(data) - rescue RubyReactor::Error::DeserializationError, - RubyReactor::Error::SchemaVersionError => e - # Permanent failures — re-reading the same stored blob will keep - # failing. Mark the context as failed (best-effort) and return so - # Sidekiq does not burn its retry budget. - handle_deserialization_failure(context_id, reactor_class_name, e) - return - end - - # If reactor_class_name is provided, use it to get the reactor class - # This handles cases where the class can't be found via const_get - if reactor_class_name && context.reactor_class.nil? - begin - context.reactor_class = Object.const_get(reactor_class_name) - rescue NameError - # If not found, try to find it in the current namespace - # This is a fallback for test environments - context.reactor_class = reactor_class_name.constantize if reactor_class_name.respond_to?(:constantize) - end - end - - # Mark that we're executing inline to prevent nested async calls - context.inline_async_execution = true - - begin - # Resume execution from the failed step - executor = Executor.new(context.reactor_class, {}, context) - executor.resume_execution - # No explicit save here: resume_execution's ensure block already persists - # the final root state (`save_context unless skip_context_persist?`), and - # in the worker the executor's context IS the root, so an extra checkpoint! - # would just re-write the identical blob to the identical key. The - # skip_context_persist? guard (stale-batch redelivery of an already-terminal - # context) is likewise honored there. - - # Return the executor (which now has the result stored in it) - executor - rescue RubyReactor::Lock::AcquisitionError, - RubyReactor::Semaphore::AcquisitionError, - RubyReactor::RateLimit::ExceededError, - RubyReactor::OrderedLock::WaitError => e - # Snooze on expected concurrency, rate, or ordering contention. - # OrderedLock::WaitError carries a poison-pill-derived retry hint, - # consumed by compute_snooze_delay below. We avoid Sidekiq's native - # retry path so this doesn't burn the job's retry budget or appear - # as an error in dashboards. After the configured cap is reached we - # escalate by marking the reactor as failed. - handle_snooze(context_id, reactor_class_name, context, snooze_count, e) - rescue RubyReactor::RateLimitRegistry::UnknownLimitError => e - # Permanent configuration error — snoozing or retrying the same job - # will keep failing. Mark the context failed immediately. - escalate_snooze(context, snooze_count, e) - end - end - - private - - def handle_snooze(context_id, reactor_class_name, context, snooze_count, error) - config = RubyReactor.configuration - max = config.lock_snooze_max_attempts - - # OrderedLock::WaitError bypasses the snooze cap. The gate's - # poison_pill_timeout is the only meaningful upper bound on how long a - # nonce can legitimately wait; capping snoozes would either fail jobs - # prematurely or strand the nonce in `assigned_at` until poison_pill - # eventually advances past it. Snooze until the gate passes (or poison - # auto-advance moves the cursor past us). - # The per-context liveness lock (`async:`) is also uncapped: a - # duplicate of the *same* execution may wait arbitrarily long for the - # live original to finish (e.g. a sweeper re-enqueue racing a slow but - # alive worker). Capping it would fail a legitimately-waiting duplicate. - capped = !(error.is_a?(RubyReactor::OrderedLock::WaitError) || - error.is_a?(RubyReactor::Lock::ContextLockContention)) - - if capped && max != :infinity && snooze_count >= max - escalate_snooze(context, snooze_count, error) - return - end - - delay = compute_snooze_delay(config, error) - # Re-enqueue by id: the context is already persisted in storage, so the - # rescheduled job rehydrates fresh state (no stale blob). - self.class.perform_in(delay, context_id, reactor_class_name, snooze_count + 1) - end - - # Use the error's `retry_after_seconds` hint when available - # (RateLimit::ExceededError carries the time until the bucket rolls); - # otherwise fall back to the configured base + jitter for lock/semaphore - # contention which has no precise hint. - # - # OrderedLock::WaitError is deliberately excluded from the hint path: its - # `retry_after_seconds` is the poison-pill window (the upper bound before - # a *dead* blocker is force-advanced), NOT how long the *live* blocker - # will take — which is usually milliseconds. Snoozing for the full window - # would make every out-of-order nonce sleep up to poison_pill_timeout even - # though its blocker finishes immediately, collapsing throughput. Re-poll - # at the base delay instead; poison auto-advance still clears a genuinely - # dead blocker on a later gate. - def compute_snooze_delay(config, error) - jitter = config.lock_snooze_jitter.to_f - jitter_amount = jitter.positive? ? rand(0.0..jitter) : 0.0 - - if hinted_retry?(error) - [error.retry_after_seconds.to_f, 0.1].max + jitter_amount - else - config.lock_snooze_base_delay.to_f + jitter_amount - end - end - - def hinted_retry?(error) - return false if error.is_a?(RubyReactor::OrderedLock::WaitError) - - error.respond_to?(:retry_after_seconds) && error.retry_after_seconds - end - - def escalate_snooze(context, snooze_count, error) - RubyReactor.configuration.logger.warn( - "RubyReactor snooze limit reached after #{snooze_count} attempts " \ - "for context #{context.context_id}: #{error.message}" - ) - - context.status = :failed - context.failure_reason = { - message: error.message, - exception_class: error.class.name, - snooze_attempts: snooze_count - } - - serialized = ContextSerializer.serialize(context) - reactor_class_name = RubyReactor.reactor_storage_name(context.reactor_class) - RubyReactor.configuration.storage_adapter.store_context( - context.context_id, - serialized, - reactor_class_name - ) - - # Escalation is a terminal Failure that never reaches the Executor's - # ensure path, so advance the ordered-lock cursor here. Without this - # the nonce stays stranded in assigned_at (successors stall for the - # full poison_pill_timeout) and, worse, the strict-mode chain marker - # is never recorded — successors would RUN instead of being skipped. - info = Executor::OrderedLockSupport.info_from(context) - Executor::OrderedLockSupport.advance_with_retry(info, failed: true) if info - end - - def log_infrastructure_failure(msg, exception) - RubyReactor.configuration.logger.error("RubyReactor infrastructure failure: #{exception.message}") - RubyReactor.configuration.logger.error("Job details: #{msg.inspect}") - end - - # The id-only payload already carries context_id and reactor_class_name, so - # there is no blob to parse for metadata — just mark the stored context - # failed (best-effort) so the job stops retrying a permanently-broken blob. - def handle_deserialization_failure(context_id, reactor_class_name, error) - RubyReactor.configuration.logger.error( - "RubyReactor deserialization failure for context " \ - "#{context_id || "unknown"}: #{error.class.name}: #{error.message}" - ) - - return unless context_id && reactor_class_name - - payload = build_failed_context_payload(context_id, reactor_class_name, error) - RubyReactor.configuration.storage_adapter.store_context( - context_id, - payload, - reactor_class_name - ) - rescue StandardError => e - # Don't let a persistence failure mask the original deserialization error. - RubyReactor.configuration.logger.error( - "RubyReactor failed to persist deserialization failure: #{e.class.name}: #{e.message}" - ) - end - - def build_failed_context_payload(context_id, reactor_class_name, error) - JSON.generate( - "schema_version" => ContextSerializer::SCHEMA_VERSION, - "context_id" => context_id, - "reactor_class" => reactor_class_name, - "status" => "failed", - "failure_reason" => { - "message" => error.message, - "exception_class" => error.class.name - } - ) - end - end - end -end diff --git a/lib/ruby_reactor/sweeper_job.rb b/lib/ruby_reactor/sweeper_job.rb new file mode 100644 index 00000000..122d76a2 --- /dev/null +++ b/lib/ruby_reactor/sweeper_job.rb @@ -0,0 +1,70 @@ +# frozen_string_literal: true + +require "securerandom" + +module RubyReactor + # Self-rescheduling recovery tick, shared by every queueing backend's + # sweeper job class. Each run sweeps both the top-level reactor sweeper and + # the map sweeper, then schedules the next tick — a perpetual chain the + # host kicks once via `RubyReactor.start_sweeper!`. + # + # super_fetch safety. Sidekiq Enterprise `super_fetch` reliably re-runs a job + # whose worker died mid-execution. For a self-rescheduling chain that is a + # hazard: a tick can crash AFTER enqueuing its successor but BEFORE acking, so + # super_fetch recovers the crashed tick *alongside* the successor it already + # scheduled — the chain forks and then doubles every interval. We therefore do + # NOT rely on "exactly one job exists". The next tick is claimed by a + # per-time-window lock: every duplicate computes the SAME target window and + # only one wins the claim, so recovered/duplicated ticks collapse back to a + # single chain. The claim lock is never released — it simply expires — so no + # delete can race two duplicates into both winning. + module SweeperJob + def self.included(base) + base.extend(ClassMethods) + end + + def perform + config = RubyReactor.configuration + return unless config.sweeper_enabled + + run_sweeps(config) + ensure + # Always chain forward (unless disabled), even after an error above, so a + # single bad sweep can't kill recovery. The window lock keeps this from + # forking under super_fetch. + self.class.schedule_next if RubyReactor.configuration.sweeper_enabled + end + + def run_sweeps(config) + RubyReactor::Sweeper.run_once(limit: config.sweeper_limit) + RubyReactor::Map::Sweeper.run_once(limit: config.sweeper_limit) + rescue StandardError => e + config.logger.error("RubyReactor sweeper sweep failed: #{e.class}: #{e.message}") + end + + module ClassMethods + # Enqueue the next tick for the upcoming time window, claiming that window + # so concurrent/duplicate/recovered ticks produce exactly one successor. + # Idempotent: also safe to call from `start_sweeper!` on every process boot. + def schedule_next + interval = RubyReactor.configuration.sweeper_interval + window = (Time.now.to_i / interval) + 1 + + lock = RubyReactor::Lock.new( + "sweeper:window:#{window}", + owner: SecureRandom.uuid, + ttl: interval * 2, # outlive the window; expires on its own (never released) + wait: 0, + auto_extend: false + ) + lock.acquire # raises AcquisitionError if this window is already claimed + + delay = (window * interval) - Time.now.to_i + perform_in([delay, 1].max) + rescue RubyReactor::Lock::AcquisitionError + # Another tick already scheduled this window — collapse the duplicate. + nil + end + end + end +end diff --git a/lib/ruby_reactor/worker.rb b/lib/ruby_reactor/worker.rb new file mode 100644 index 00000000..1dc876ff --- /dev/null +++ b/lib/ruby_reactor/worker.rb @@ -0,0 +1,212 @@ +# frozen_string_literal: true + +module RubyReactor + # Framework-agnostic resume/snooze/escalate logic shared by every queueing + # backend's worker class. Each backend (`Adapters::Sidekiq::Worker`, + # `Adapters::ActiveJob::Worker`, ...) includes this and supplies its own + # `self.class.perform_in` (native on Sidekiq::Worker, via + # `Adapters::ActiveJob::Compat` on ActiveJob::Base) — nothing here references + # a specific backend. + module Worker + # Identity-only payload: storage is the source of truth. Rehydrate the live + # context from storage by id, then resume. A nil read means the context was + # swept, expired, or already terminal-and-collected — nothing to resume. + def perform(context_id, reactor_class_name = nil, snooze_count = 0) + # Normalize so a nil/omitted name resolves to the same storage key the + # enqueue path wrote (always via reactor_storage_name). Without this a + # nil here builds "reactor::context:" and misses the stored + # "reactor:AnonymousReactor:context:", silently no-op'ing. + reactor_class_name ||= RubyReactor.reactor_storage_name(nil) + data = RubyReactor.configuration.storage_adapter.retrieve_context(context_id, reactor_class_name) + return if data.nil? + + begin + context = ContextSerializer.deserialize_hash(data) + rescue RubyReactor::Error::DeserializationError, + RubyReactor::Error::SchemaVersionError => e + # Permanent failures — re-reading the same stored blob will keep + # failing. Mark the context as failed (best-effort) and return so + # the job does not burn its retry budget. + handle_deserialization_failure(context_id, reactor_class_name, e) + return + end + + # If reactor_class_name is provided, use it to get the reactor class + # This handles cases where the class can't be found via const_get + if reactor_class_name && context.reactor_class.nil? + begin + context.reactor_class = Object.const_get(reactor_class_name) + rescue NameError + # If not found, try to find it in the current namespace + # This is a fallback for test environments + context.reactor_class = reactor_class_name.constantize if reactor_class_name.respond_to?(:constantize) + end + end + + # Mark that we're executing inline to prevent nested async calls + context.inline_async_execution = true + + begin + # Resume execution from the failed step + executor = Executor.new(context.reactor_class, {}, context) + executor.resume_execution + # No explicit save here: resume_execution's ensure block already persists + # the final root state (`save_context unless skip_context_persist?`), and + # in the worker the executor's context IS the root, so an extra checkpoint! + # would just re-write the identical blob to the identical key. The + # skip_context_persist? guard (stale-batch redelivery of an already-terminal + # context) is likewise honored there. + + # Return the executor (which now has the result stored in it) + executor + rescue RubyReactor::Lock::AcquisitionError, + RubyReactor::Semaphore::AcquisitionError, + RubyReactor::RateLimit::ExceededError, + RubyReactor::OrderedLock::WaitError => e + # Snooze on expected concurrency, rate, or ordering contention. + # OrderedLock::WaitError carries a poison-pill-derived retry hint, + # consumed by compute_snooze_delay below. We avoid the framework's native + # retry path so this doesn't burn the job's retry budget or appear + # as an error in dashboards. After the configured cap is reached we + # escalate by marking the reactor as failed. + handle_snooze(context_id, reactor_class_name, context, snooze_count, e) + rescue RubyReactor::RateLimitRegistry::UnknownLimitError => e + # Permanent configuration error — snoozing or retrying the same job + # will keep failing. Mark the context failed immediately. + escalate_snooze(context, snooze_count, e) + end + end + + private + + def handle_snooze(context_id, reactor_class_name, context, snooze_count, error) + config = RubyReactor.configuration + max = config.lock_snooze_max_attempts + + # OrderedLock::WaitError bypasses the snooze cap. The gate's + # poison_pill_timeout is the only meaningful upper bound on how long a + # nonce can legitimately wait; capping snoozes would either fail jobs + # prematurely or strand the nonce in `assigned_at` until poison_pill + # eventually advances past it. Snooze until the gate passes (or poison + # auto-advance moves the cursor past us). + # The per-context liveness lock (`async:`) is also uncapped: a + # duplicate of the *same* execution may wait arbitrarily long for the + # live original to finish (e.g. a sweeper re-enqueue racing a slow but + # alive worker). Capping it would fail a legitimately-waiting duplicate. + capped = !(error.is_a?(RubyReactor::OrderedLock::WaitError) || + error.is_a?(RubyReactor::Lock::ContextLockContention)) + + if capped && max != :infinity && snooze_count >= max + escalate_snooze(context, snooze_count, error) + return + end + + delay = compute_snooze_delay(config, error) + # Re-enqueue by id: the context is already persisted in storage, so the + # rescheduled job rehydrates fresh state (no stale blob). + self.class.perform_in(delay, context_id, reactor_class_name, snooze_count + 1) + end + + # Use the error's `retry_after_seconds` hint when available + # (RateLimit::ExceededError carries the time until the bucket rolls); + # otherwise fall back to the configured base + jitter for lock/semaphore + # contention which has no precise hint. + # + # OrderedLock::WaitError is deliberately excluded from the hint path: its + # `retry_after_seconds` is the poison-pill window (the upper bound before + # a *dead* blocker is force-advanced), NOT how long the *live* blocker + # will take — which is usually milliseconds. Snoozing for the full window + # would make every out-of-order nonce sleep up to poison_pill_timeout even + # though its blocker finishes immediately, collapsing throughput. Re-poll + # at the base delay instead; poison auto-advance still clears a genuinely + # dead blocker on a later gate. + def compute_snooze_delay(config, error) + jitter = config.lock_snooze_jitter.to_f + jitter_amount = jitter.positive? ? rand(0.0..jitter) : 0.0 + + if hinted_retry?(error) + [error.retry_after_seconds.to_f, 0.1].max + jitter_amount + else + config.lock_snooze_base_delay.to_f + jitter_amount + end + end + + def hinted_retry?(error) + return false if error.is_a?(RubyReactor::OrderedLock::WaitError) + + error.respond_to?(:retry_after_seconds) && error.retry_after_seconds + end + + def escalate_snooze(context, snooze_count, error) + RubyReactor.configuration.logger.warn( + "RubyReactor snooze limit reached after #{snooze_count} attempts " \ + "for context #{context.context_id}: #{error.message}" + ) + + context.status = :failed + context.failure_reason = { + message: error.message, + exception_class: error.class.name, + snooze_attempts: snooze_count + } + + serialized = ContextSerializer.serialize(context) + reactor_class_name = RubyReactor.reactor_storage_name(context.reactor_class) + RubyReactor.configuration.storage_adapter.store_context( + context.context_id, + serialized, + reactor_class_name + ) + + # Escalation is a terminal Failure that never reaches the Executor's + # ensure path, so advance the ordered-lock cursor here. Without this + # the nonce stays stranded in assigned_at (successors stall for the + # full poison_pill_timeout) and, worse, the strict-mode chain marker + # is never recorded — successors would RUN instead of being skipped. + info = Executor::OrderedLockSupport.info_from(context) + Executor::OrderedLockSupport.advance_with_retry(info, failed: true) if info + end + + def log_infrastructure_failure(msg, exception) + RubyReactor.configuration.logger.error("RubyReactor infrastructure failure: #{exception.message}") + RubyReactor.configuration.logger.error("Job details: #{msg.inspect}") + end + + # The id-only payload already carries context_id and reactor_class_name, so + # there is no blob to parse for metadata — just mark the stored context + # failed (best-effort) so the job stops retrying a permanently-broken blob. + def handle_deserialization_failure(context_id, reactor_class_name, error) + RubyReactor.configuration.logger.error( + "RubyReactor deserialization failure for context " \ + "#{context_id || "unknown"}: #{error.class.name}: #{error.message}" + ) + + return unless context_id && reactor_class_name + + payload = build_failed_context_payload(context_id, reactor_class_name, error) + RubyReactor.configuration.storage_adapter.store_context( + context_id, + payload, + reactor_class_name + ) + rescue StandardError => e + # Don't let a persistence failure mask the original deserialization error. + RubyReactor.configuration.logger.error( + "RubyReactor failed to persist deserialization failure: #{e.class.name}: #{e.message}" + ) + end + + def build_failed_context_payload(context_id, reactor_class_name, error) + JSON.generate( + "schema_version" => ContextSerializer::SCHEMA_VERSION, + "context_id" => context_id, + "reactor_class" => reactor_class_name, + "status" => "failed", + "failure_reason" => { + "message" => error.message, + "exception_class" => error.class.name + } + ) + end + end +end diff --git a/spec/async_retry_integration_spec.rb b/spec/async_retry_integration_spec.rb index c9692a79..0ab33519 100644 --- a/spec/async_retry_integration_spec.rb +++ b/spec/async_retry_integration_spec.rb @@ -6,7 +6,7 @@ before do Sidekiq::Testing.fake! Sidekiq::Worker.clear_all - allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) end after do @@ -24,7 +24,7 @@ expect(result.failure?).to be false # Verify job was queued - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) end it "executes full reactor asynchronously" do @@ -32,10 +32,10 @@ reactor.run(user_id: 123, email: "test@example.com") # Process the queued job - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # Verify the job was processed (in real Sidekiq this would happen asynchronously) - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(0) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(0) end end @@ -47,7 +47,7 @@ expect(result).to be_a(RubyReactor::AsyncResult) # Verify job was queued for async step - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) end it "resumes execution from async step in worker" do @@ -55,10 +55,10 @@ reactor.run(user_id: 123, email: "test@example.com") # Process the queued job - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # Verify the job was processed - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(0) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(0) end end @@ -98,17 +98,17 @@ expect(result).to be_a(RubyReactor::AsyncResult) # Should have 1 job queued - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) # Process the job - this should execute the step, fail, and requeue for retry - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # In fake mode, drain processes all jobs, including scheduled ones # Since the step fails and retries, it should process multiple jobs # But ultimately, when max_attempts is reached, no more jobs should be queued # After all processing, there should be no jobs left - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(0) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(0) end end @@ -118,10 +118,10 @@ reactor.run(attempt_count: 1) # Should have 1 job initially - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) # Process jobs - should retry a few times - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # Verify retries happened (in fake mode, jobs are processed immediately) # In real Sidekiq, this would be scheduled with delays @@ -132,8 +132,8 @@ reactor = RetryLinearReactor.new reactor.run - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) - RubyReactor::SidekiqWorkers::Worker.drain + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) + RubyReactor::Adapters::Sidekiq::Worker.drain end end @@ -142,8 +142,8 @@ reactor = RetryFixedReactor.new reactor.run - expect(RubyReactor::SidekiqWorkers::Worker.jobs.size).to eq(1) - RubyReactor::SidekiqWorkers::Worker.drain + expect(RubyReactor::Adapters::Sidekiq::Worker.jobs.size).to eq(1) + RubyReactor::Adapters::Sidekiq::Worker.drain end end end diff --git a/spec/examples/data_pipeline_spec.rb b/spec/examples/data_pipeline_spec.rb index fc1e2fe9..24593bd4 100644 --- a/spec/examples/data_pipeline_spec.rb +++ b/spec/examples/data_pipeline_spec.rb @@ -269,7 +269,7 @@ class AsyncMapReactor < RubyReactor::Reactor end before do - allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) Sidekiq::Testing.fake! end @@ -284,7 +284,7 @@ class AsyncMapReactor < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) # With batch_size: 2, should queue 2 jobs initially - expect(RubyReactor::SidekiqWorkers::MapElementWorker.jobs.size).to eq(2) + expect(RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.size).to eq(2) end end diff --git a/spec/map/async_map_execution_spec.rb b/spec/map/async_map_execution_spec.rb index 08f0d4fd..786faf4a 100644 --- a/spec/map/async_map_execution_spec.rb +++ b/spec/map/async_map_execution_spec.rb @@ -5,8 +5,8 @@ RSpec.describe "Async Map Execution" do before do # Use real Redis from spec_helper configuration - # But we need to ensure SidekiqAdapter is used instead of WorkerMock for this test - allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + # But we need to ensure Adapters::Sidekiq::Router is used instead of WorkerMock for this test + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) Sidekiq::Testing.fake! end @@ -23,7 +23,7 @@ # Check Sidekiq jobs # With batch_size: 1, only 1 job should be queued initially - expect(RubyReactor::SidekiqWorkers::MapElementWorker.jobs.size).to eq(1) + expect(RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.size).to eq(1) end it "processes map elements and triggers collector" do @@ -35,13 +35,13 @@ context_id = reactor.context.context_id # Process jobs - RubyReactor::SidekiqWorkers::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain # Check if Collector was queued - expect(RubyReactor::SidekiqWorkers::MapCollectorWorker.jobs.size).to eq(2) + expect(RubyReactor::Adapters::Sidekiq::MapCollectorWorker.jobs.size).to eq(2) # Process Collector - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Verify result in Redis storage = RubyReactor.configuration.storage_adapter diff --git a/spec/map/map_async_retry_spec.rb b/spec/map/map_async_retry_spec.rb index 9fb622c4..ab91289d 100644 --- a/spec/map/map_async_retry_spec.rb +++ b/spec/map/map_async_retry_spec.rb @@ -54,8 +54,8 @@ class AsyncRetryMapReactorV2 < RubyReactor::Reactor # No batch_size now runs through the same per-element path as batch_size # maps (batch_size defaults to the full source size), so drain the element # and collector workers; retries requeue MapElementWorker jobs. - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Retrieve final result context_id = reactor.context.context_id @@ -75,8 +75,8 @@ class AsyncRetryMapReactorV2 < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain context_id = reactor.context.context_id storage = RubyReactor.configuration.storage_adapter @@ -146,8 +146,8 @@ class AsyncBatchRetryMapReactorV2 < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) # Drain workers (multiple times might be needed for batches/retries) - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Retrieve final result context_id = reactor.context.context_id @@ -173,8 +173,8 @@ class AsyncBatchRetryMapReactorV2 < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) # Drain workers - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Retrieve final result context_id = reactor.context.context_id @@ -257,8 +257,8 @@ class AsyncFailFastFalseRetryReactorV2 < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) # Drain workers - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Retrieve final result context_id = reactor.context.context_id @@ -282,8 +282,8 @@ class AsyncFailFastFalseRetryReactorV2 < RubyReactor::Reactor expect(result).to be_a(RubyReactor::AsyncResult) # Drain workers - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Retrieve final result context_id = reactor.context.context_id diff --git a/spec/map/map_batch_size_spec.rb b/spec/map/map_batch_size_spec.rb index b0ea4035..3c5b12a3 100644 --- a/spec/map/map_batch_size_spec.rb +++ b/spec/map/map_batch_size_spec.rb @@ -5,8 +5,8 @@ RSpec.describe "Map Batch Size Execution" do before do # Use real Redis from spec_helper configuration - # But we need to ensure SidekiqAdapter is used instead of WorkerMock for this test - allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + # But we need to ensure Adapters::Sidekiq::Router is used instead of WorkerMock for this test + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) Sidekiq::Testing.fake! end @@ -23,21 +23,21 @@ expect(result).to be_a(RubyReactor::AsyncResult) # Should only queue 2 jobs initially because batch_size is 2 - expect(RubyReactor::SidekiqWorkers::MapElementWorker.jobs.size).to eq(2) + expect(RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.size).to eq(2) # Process the first batch - initial_jobs = RubyReactor::SidekiqWorkers::MapElementWorker.jobs.dup - RubyReactor::SidekiqWorkers::MapElementWorker.clear + initial_jobs = RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.dup + RubyReactor::Adapters::Sidekiq::MapElementWorker.clear initial_jobs.each do |job| - RubyReactor::SidekiqWorkers::MapElementWorker.new.perform(*job["args"]) + RubyReactor::Adapters::Sidekiq::MapElementWorker.new.perform(*job["args"]) end # Should have queued 2 more jobs (indices 2 and 3) - expect(RubyReactor::SidekiqWorkers::MapElementWorker.jobs.size).to eq(2) + expect(RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.size).to eq(2) # Verify the indices - job_args = RubyReactor::SidekiqWorkers::MapElementWorker.jobs.map { |j| j["args"].first } + job_args = RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.map { |j| j["args"].first } indices = job_args.map { |a| a["index"] } expect(indices).to contain_exactly(2, 3) end diff --git a/spec/map/map_fail_fast_spec.rb b/spec/map/map_fail_fast_spec.rb index 8bf9f35b..f2a42b98 100644 --- a/spec/map/map_fail_fast_spec.rb +++ b/spec/map/map_fail_fast_spec.rb @@ -372,13 +372,13 @@ class AsyncFailFastReactor < RubyReactor::Reactor # 2. MapCollectorWorker waits for results # Main Worker - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # Process elements (Batches) - RubyReactor::SidekiqWorkers::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain # Collector - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Reload reactor from storage to check status stored_reactor = AsyncFailFastReactor.find(context.context_id) diff --git a/spec/map/map_recovery_spec.rb b/spec/map/map_recovery_spec.rb index 31ab52ca..097a41f1 100644 --- a/spec/map/map_recovery_spec.rb +++ b/spec/map/map_recovery_spec.rb @@ -7,15 +7,15 @@ # completes off the index-keyed results hash. RSpec.describe "Map recovery (M1)" do before do - allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + allow(RubyReactor.configuration).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) Sidekiq::Testing.fake! end after { Sidekiq::Testing.inline! } let(:storage) { RubyReactor.configuration.storage_adapter } - let(:element_worker) { RubyReactor::SidekiqWorkers::MapElementWorker } - let(:collector_worker) { RubyReactor::SidekiqWorkers::MapCollectorWorker } + let(:element_worker) { RubyReactor::Adapters::Sidekiq::MapElementWorker } + let(:collector_worker) { RubyReactor::Adapters::Sidekiq::MapCollectorWorker } it "re-dispatches every missing index when element jobs are lost, then completes" do reactor = MapTestReactors::AsyncMapReactor.new diff --git a/spec/ruby_reactor/adapters/active_job/compat_spec.rb b/spec/ruby_reactor/adapters/active_job/compat_spec.rb new file mode 100644 index 00000000..52b814ef --- /dev/null +++ b/spec/ruby_reactor/adapters/active_job/compat_spec.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +require "spec_helper" +require "active_job" +require "ruby_reactor/adapters/active_job/compat" + +RSpec.describe RubyReactor::Adapters::ActiveJob::Compat do + let(:job_class) do + Class.new(ActiveJob::Base) do + extend RubyReactor::Adapters::ActiveJob::Compat + + def perform(*); end + end + end + + around do |example| + original = ActiveJob::Base.queue_adapter + ActiveJob::Base.queue_adapter = :test + example.run + ensure + ActiveJob::Base.queue_adapter = original + end + + describe ".perform_async" do + it "enqueues immediately via perform_later and returns the job id" do + job_id = job_class.perform_async("a", "b") + + enqueued = job_class.queue_adapter.enqueued_jobs + expect(enqueued.size).to eq(1) + expect(enqueued.first[:args]).to eq(%w[a b]) + expect(job_id).to be_a(String) + end + end + + describe ".perform_in" do + it "enqueues with a wait delay via perform_later and returns the job id" do + job_id = job_class.perform_in(30, "a") + + enqueued = job_class.queue_adapter.enqueued_jobs + expect(enqueued.size).to eq(1) + expect(enqueued.first[:args]).to eq(["a"]) + expect(job_id).to be_a(String) + end + end +end diff --git a/spec/ruby_reactor/adapters/active_job/router_spec.rb b/spec/ruby_reactor/adapters/active_job/router_spec.rb new file mode 100644 index 00000000..13e5c90d --- /dev/null +++ b/spec/ruby_reactor/adapters/active_job/router_spec.rb @@ -0,0 +1,44 @@ +# frozen_string_literal: true + +require "spec_helper" +require "active_job" +require "ruby_reactor/adapters/active_job/router" + +# End-to-end smoke test that a full reactor run can go through the +# ActiveJob backend top to bottom: enqueue via `Router.perform_async`, +# drain via the ActiveJob `:test` queue adapter, resume via `Worker#perform`. +# Mirrors the equivalent Sidekiq coverage in spec/async_retry_integration_spec.rb. +RSpec.describe RubyReactor::Adapters::ActiveJob::Router do + around do |example| + original = ActiveJob::Base.queue_adapter + ActiveJob::Base.queue_adapter = :test + example.run + ensure + ActiveJob::Base.queue_adapter = original + end + + before do + allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(described_class) + RubyReactor::Adapters::ActiveJob::Worker.queue_adapter.enqueued_jobs.clear + end + + it "queues a job immediately and returns an AsyncResult" do + reactor = TestAsyncReactor.new + result = reactor.run(user_id: 123, email: "test@example.com") + + expect(result).to be_a(RubyReactor::AsyncResult) + expect(result.async?).to be true + expect(RubyReactor::Adapters::ActiveJob::Worker.queue_adapter.enqueued_jobs.size).to eq(1) + end + + it "executes the full reactor once the enqueued job is performed" do + reactor = TestAsyncReactor.new + result = reactor.run(user_id: 123, email: "test@example.com") + + enqueued = RubyReactor::Adapters::ActiveJob::Worker.queue_adapter.enqueued_jobs.first + RubyReactor::Adapters::ActiveJob::Worker.new(*enqueued[:args]).perform_now + + instance = TestAsyncReactor.find(result.execution_id) + expect(instance.context.status.to_s).to eq("completed") + end +end diff --git a/spec/ruby_reactor/adapters/active_job/sweeper_worker_spec.rb b/spec/ruby_reactor/adapters/active_job/sweeper_worker_spec.rb new file mode 100644 index 00000000..88bbb769 --- /dev/null +++ b/spec/ruby_reactor/adapters/active_job/sweeper_worker_spec.rb @@ -0,0 +1,68 @@ +# frozen_string_literal: true + +require "spec_helper" +require "active_job" +require "ruby_reactor/adapters/active_job/sweeper_worker" + +RSpec.describe RubyReactor::Adapters::ActiveJob::SweeperWorker do + around do |example| + original = ActiveJob::Base.queue_adapter + ActiveJob::Base.queue_adapter = :test + example.run + ensure + ActiveJob::Base.queue_adapter = original + end + + before do + described_class.queue_adapter.enqueued_jobs.clear + allow(RubyReactor::Sweeper).to receive(:run_once).and_return(0) + allow(RubyReactor::Map::Sweeper).to receive(:run_once).and_return(0) + end + + describe ".schedule_next" do + it "enqueues exactly one tick for the upcoming window" do + described_class.schedule_next + expect(described_class.queue_adapter.enqueued_jobs.size).to eq(1) + end + + it "is single-flight: a duplicate call for the same window enqueues nothing more" do + described_class.schedule_next + described_class.schedule_next + + expect(described_class.queue_adapter.enqueued_jobs.size).to eq(1) + end + end + + describe "#perform" do + it "runs both sweepers and schedules the next tick" do + described_class.new.perform + + expect(RubyReactor::Sweeper).to have_received(:run_once) + expect(RubyReactor::Map::Sweeper).to have_received(:run_once) + expect(described_class.queue_adapter.enqueued_jobs.size).to eq(1) + end + + it "still chains forward when a sweep raises (recovery must not die on one bad sweep)" do + allow(RubyReactor::Sweeper).to receive(:run_once).and_raise(StandardError, "boom") + + expect { described_class.new.perform }.not_to raise_error + expect(described_class.queue_adapter.enqueued_jobs.size).to eq(1) + end + + context "when the sweeper is disabled" do + around do |example| + original = RubyReactor.configuration.sweeper_enabled + RubyReactor.configuration.sweeper_enabled = false + example.run + RubyReactor.configuration.sweeper_enabled = original + end + + it "does not sweep and does not chain" do + described_class.new.perform + + expect(RubyReactor::Sweeper).not_to have_received(:run_once) + expect(described_class.queue_adapter.enqueued_jobs).to be_empty + end + end + end +end diff --git a/spec/ruby_reactor/adapters/active_job/worker_spec.rb b/spec/ruby_reactor/adapters/active_job/worker_spec.rb new file mode 100644 index 00000000..a20b3224 --- /dev/null +++ b/spec/ruby_reactor/adapters/active_job/worker_spec.rb @@ -0,0 +1,73 @@ +# frozen_string_literal: true + +require "spec_helper" +require "active_job" +require "ruby_reactor/adapters/active_job/worker" + +# The resume/snooze/escalate logic itself is exercised exhaustively against +# the shared `RubyReactor::Worker` mixin in +# spec/ruby_reactor/adapters/sidekiq/worker_spec.rb. This spec only covers +# the ActiveJob-specific wiring: Compat-based enqueue, and that #perform +# delegates into the shared mixin. +# rubocop:disable RSpec/MultipleMemoizedHelpers +RSpec.describe RubyReactor::Adapters::ActiveJob::Worker do + around do |example| + original = ActiveJob::Base.queue_adapter + ActiveJob::Base.queue_adapter = :test + example.run + ensure + ActiveJob::Base.queue_adapter = original + end + + it "extends Compat for perform_async/perform_in" do + expect(described_class.singleton_class.included_modules).to include(RubyReactor::Adapters::ActiveJob::Compat) + end + + it "includes the shared RubyReactor::Worker mixin" do + expect(described_class.included_modules).to include(RubyReactor::Worker) + end + + describe "#perform" do + let(:context_id) { "test-execution-id" } + let(:reactor_class_name) { "TestReactor" } + let(:reactor_class) { Class.new { def self.name = "TestReactor" } } + let(:context) do + instance_double( + RubyReactor::Context, + context_id: context_id, + reactor_class: reactor_class, + inline_async_execution: true + ) + end + let(:executor) { instance_double(RubyReactor::Executor) } + let(:adapter) { instance_double(RubyReactor::Storage::RedisAdapter) } + let(:stored_data) { { "context_id" => context_id, "schema_version" => "1.0" } } + + before do + allow(RubyReactor.configuration).to receive(:storage_adapter).and_return(adapter) + allow(adapter).to receive(:retrieve_context).with(context_id, reactor_class_name).and_return(stored_data) + allow(RubyReactor::ContextSerializer).to receive(:deserialize_hash).and_return(context) + allow(RubyReactor::Executor).to receive(:new).and_return(executor) + allow(context).to receive(:inline_async_execution=) + end + + it "rehydrates by id and resumes the reactor" do + allow(executor).to receive(:resume_execution) + described_class.new.perform(context_id, reactor_class_name) + expect(executor).to have_received(:resume_execution) + end + + it "reschedules via the ActiveJob set(wait:).perform_later path on lock contention" do + allow(executor).to receive(:resume_execution).and_raise(RubyReactor::Lock::AcquisitionError) + RubyReactor.configuration.lock_snooze_base_delay = 5 + RubyReactor.configuration.lock_snooze_jitter = 0 + + described_class.new.perform(context_id, reactor_class_name) + + enqueued = described_class.queue_adapter.enqueued_jobs + expect(enqueued.size).to eq(1) + expect(enqueued.first[:args]).to eq([context_id, reactor_class_name, 1]) + end + end +end +# rubocop:enable RSpec/MultipleMemoizedHelpers diff --git a/spec/ruby_reactor/sidekiq_workers/sweeper_worker_spec.rb b/spec/ruby_reactor/adapters/sidekiq/sweeper_worker_spec.rb similarity index 94% rename from spec/ruby_reactor/sidekiq_workers/sweeper_worker_spec.rb rename to spec/ruby_reactor/adapters/sidekiq/sweeper_worker_spec.rb index e0539636..d35687a2 100644 --- a/spec/ruby_reactor/sidekiq_workers/sweeper_worker_spec.rb +++ b/spec/ruby_reactor/adapters/sidekiq/sweeper_worker_spec.rb @@ -2,9 +2,9 @@ require "spec_helper" require "sidekiq/testing" -require "ruby_reactor/sidekiq_workers/sweeper_worker" +require "ruby_reactor/adapters/sidekiq/sweeper_worker" -RSpec.describe RubyReactor::SidekiqWorkers::SweeperWorker do +RSpec.describe RubyReactor::Adapters::Sidekiq::SweeperWorker do before do described_class.clear # The sweep itself is exercised by the Sweeper specs; here we only care about diff --git a/spec/ruby_reactor/sidekiq_workers/worker_spec.rb b/spec/ruby_reactor/adapters/sidekiq/worker_spec.rb similarity index 99% rename from spec/ruby_reactor/sidekiq_workers/worker_spec.rb rename to spec/ruby_reactor/adapters/sidekiq/worker_spec.rb index 178206ef..f95f417b 100644 --- a/spec/ruby_reactor/sidekiq_workers/worker_spec.rb +++ b/spec/ruby_reactor/adapters/sidekiq/worker_spec.rb @@ -2,13 +2,13 @@ require "spec_helper" require "sidekiq/testing" -require "ruby_reactor/sidekiq_workers/worker" +require "ruby_reactor/adapters/sidekiq/worker" # The worker fans many collaborators (storage, serializer, executor, context, # router) into one unit, so the suite needs more memoized helpers than the cop's # default; the alternative (re-stubbing in every example) is worse. # rubocop:disable RSpec/MultipleMemoizedHelpers -RSpec.describe RubyReactor::SidekiqWorkers::Worker do +RSpec.describe RubyReactor::Adapters::Sidekiq::Worker do subject(:worker) { described_class.new } let(:reactor_class) { Class.new { def self.name = "TestReactor" } } diff --git a/spec/ruby_reactor/async_notification_interrupt_spec.rb b/spec/ruby_reactor/async_notification_interrupt_spec.rb index 56113f3c..e8676d6a 100644 --- a/spec/ruby_reactor/async_notification_interrupt_spec.rb +++ b/spec/ruby_reactor/async_notification_interrupt_spec.rb @@ -28,7 +28,7 @@ execution = reactor_class.run - # SidekiqAdapter now returns InterruptResult when inline! is active + # Adapters::Sidekiq::Router now returns InterruptResult when inline! is active expect(execution).to be_a(RubyReactor::InterruptResult) execution_id = execution.execution_id @@ -50,7 +50,7 @@ payload: { status: "approved" } ) - # SidekiqAdapter now returns Success when inline! is active + # Adapters::Sidekiq::Router now returns Success when inline! is active expect(result).to be_a(RubyReactor::Success) # Inline execution should have completed 'process_approval' @@ -71,8 +71,8 @@ context "with manual async execution" do before do - # Use real SidekiqAdapter to test actual Sidekiq queuing - allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + # Use real Adapters::Sidekiq::Router to test actual Sidekiq queuing + allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) end it "returns AsyncResult initially, then pauses at interrupt after worker runs" do @@ -89,7 +89,7 @@ # 2. Drain Sidekiq jobs to run send_notifications # This runs send_notifications and pauses at manager_approval - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(reactor_class.trace).to eq(%i[retrieve_context send_notifications]) @@ -107,7 +107,7 @@ expect(reactor_class.trace).to eq(%i[retrieve_context send_notifications]) # 4. Drain again to run process_approval - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(reactor_class.trace).to eq(%i[retrieve_context send_notifications process_approval]) diff --git a/spec/ruby_reactor/integration/locking_spec.rb b/spec/ruby_reactor/integration/locking_spec.rb index 7a5b63f6..6bee1750 100644 --- a/spec/ruby_reactor/integration/locking_spec.rb +++ b/spec/ruby_reactor/integration/locking_spec.rb @@ -91,7 +91,7 @@ RubyReactor.configuration.lock_snooze_jitter = 0 # Stub SidekiqWorker to capture rescheduling (snooze counter starts at 1) - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) # Persist the context, then drive the worker by id (identity-only payload). context = RubyReactor::Context.new({ user_id: 1 }, SimpleLockReactor) @@ -99,10 +99,10 @@ context.context_id, RubyReactor::ContextSerializer.serialize(context), "SimpleLockReactor" ) - worker = RubyReactor::SidekiqWorkers::Worker.new + worker = RubyReactor::Adapters::Sidekiq::Worker.new worker.perform(context.context_id, "SimpleLockReactor") - expect(RubyReactor::SidekiqWorkers::Worker).to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).to have_received(:perform_in) .with(5.0, instance_of(String), "SimpleLockReactor", 1) end end @@ -533,7 +533,7 @@ def capture_rate_limit_error # (it never calls Executor#execute), so the period and rate-limit gates must # be enforced there too — but only on first runs, never on genuine resumes. describe "Async first-run gating (worker resume path)" do - let(:worker) { RubyReactor::SidekiqWorkers::Worker.new } + let(:worker) { RubyReactor::Adapters::Sidekiq::Worker.new } # Identity-only payload: the worker rehydrates from storage by id, so a fresh # async first-run must be persisted before the worker is driven. @@ -564,7 +564,7 @@ def persist(context, reactor_class) end it "skips a fresh worker pass when the bucket is already marked" do - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) PeriodicReactor.run(org_id: 71) PeriodicCounters.reset @@ -572,7 +572,7 @@ def persist(context, reactor_class) expect(executor.result).to be_a(RubyReactor::Skipped) expect(PeriodicCounters.runs).to eq(0) - expect(RubyReactor::SidekiqWorkers::Worker).not_to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).not_to have_received(:perform_in) end it "does not re-check the gate on a genuine resume (a paused run must not skip itself)" do @@ -597,19 +597,19 @@ def persist(context, reactor_class) end it "snoozes a fresh worker pass when the window is full, using the retry_after hint" do - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) 3.times { RateLimitedReactor.run(account_id: 81) } RateLimitCounters.reset worker.perform(store_fresh_context(RateLimitedReactor, { account_id: 81 }), "RateLimitedReactor") expect(RateLimitCounters.runs).to eq(0) - expect(RubyReactor::SidekiqWorkers::Worker).to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).to have_received(:perform_in) .with(a_value >= 0.1, instance_of(String), "RateLimitedReactor", 1) end it "does not re-check the limit on a genuine resume (a paused run must not throttle itself)" do - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) 3.times { RateLimitedReactor.run(account_id: 82) } # window now full RateLimitCounters.reset @@ -619,11 +619,11 @@ def persist(context, reactor_class) worker.perform(context.context_id, "RateLimitedReactor") expect(RateLimitCounters.runs).to eq(1) - expect(RubyReactor::SidekiqWorkers::Worker).not_to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).not_to have_received(:perform_in) end it "marks the context failed immediately on an unknown named limit (no snooze, no Sidekiq retry)" do - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) RubyReactor.configuration.instance_variable_set(:@rate_limits, RubyReactor::RateLimitRegistry.new) context = RubyReactor::Context.new({}, NamedRateLimitedReactor) @@ -633,7 +633,7 @@ def persist(context, reactor_class) reloaded = NamedRateLimitedReactor.find(context.context_id) expect(reloaded.context.status.to_s).to eq("failed") - expect(RubyReactor::SidekiqWorkers::Worker).not_to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).not_to have_received(:perform_in) end end end diff --git a/spec/ruby_reactor/integration/ordered_lock_spec.rb b/spec/ruby_reactor/integration/ordered_lock_spec.rb index ef311490..10222fd7 100644 --- a/spec/ruby_reactor/integration/ordered_lock_spec.rb +++ b/spec/ruby_reactor/integration/ordered_lock_spec.rb @@ -49,15 +49,15 @@ # snooze must be queued. Note: WaitError bypasses lock_snooze_max_attempts, # so we drive perform manually rather than using `drain` (which would # loop until poison_pill_timeout elapses). - third = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.jobs.clear + third = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in).and_call_original - RubyReactor::SidekiqWorkers::Worker.new.perform(*third["args"]) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in).and_call_original + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*third["args"]) expect(OrderedLockCounters.runs).to be_empty expect(adapter.ordered_lock_peek("orders:9")[:last_completed]).to eq(0) - expect(RubyReactor::SidekiqWorkers::Worker).to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).to have_received(:perform_in) end it "snoozes a waiting nonce at the base delay, not the poison-pill window" do @@ -70,14 +70,14 @@ OrderedReactor.run(account_id: 11, thing: :first) # nonce 1 OrderedReactor.run(account_id: 11, thing: :second) # nonce 2 - second = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.jobs.clear + second = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear delays = [] - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) do |delay, *_| + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) do |delay, *_| delays << delay end - RubyReactor::SidekiqWorkers::Worker.new.perform(*second["args"]) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*second["args"]) expect(delays).to contain_exactly(0.01) end @@ -90,7 +90,7 @@ # The fake queue preserves enqueue order. Drain in order: RubyReactor.configuration.lock_snooze_base_delay = 0.01 RubyReactor.configuration.lock_snooze_jitter = 0 - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(OrderedLockCounters.runs).to eq([1, 2, 3]) end @@ -100,8 +100,8 @@ RubyReactor.configuration.lock_snooze_jitter = 0 OrderedReactor.run(account_id: 21, thing: :only) # nonce 1, lone batch - job = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.drain + job = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.drain expect(OrderedLockCounters.runs).to eq([:only]) expect(adapter.ordered_lock_peek("orders:21")).to be_ordered_lock_drained @@ -111,7 +111,7 @@ # reached :completed, so this must be skipped — NOT re-run — and must not # clobber the terminal record. context_id = job["args"].first # identity-only payload: args.first IS the id - RubyReactor::SidekiqWorkers::Worker.new.perform(*job["args"]) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*job["args"]) expect(OrderedLockCounters.runs).to eq([:only]) # step did not run again expect(OrderedReactor.find(context_id).context.status.to_s).to eq("completed") @@ -142,7 +142,7 @@ StrictOrderedReactor.run(account_id: 1, thing: :two) StrictOrderedReactor.run(account_id: 1, thing: :three) - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(OrderedLockCounters.runs).to eq([:one]) expect(adapter.ordered_lock_peek("strict:1")).to be_ordered_lock_drained @@ -152,9 +152,9 @@ StrictOrderedReactor.run(account_id: 2, thing: :one, fail: true) StrictOrderedReactor.run(account_id: 2, thing: :two) - first_job = RubyReactor::SidekiqWorkers::Worker.jobs.first - RubyReactor::SidekiqWorkers::Worker.jobs.shift - RubyReactor::SidekiqWorkers::Worker.new.perform(*first_job["args"]) + first_job = RubyReactor::Adapters::Sidekiq::Worker.jobs.first + RubyReactor::Adapters::Sidekiq::Worker.jobs.shift + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*first_job["args"]) expect(adapter.ordered_lock_peek("strict:2")[:first_failed]).to eq(1) end @@ -162,7 +162,7 @@ it "drains the marker after a full batch completes (strict skips advance the cursor)" do StrictOrderedReactor.run(account_id: 3, thing: :one, fail: true) StrictOrderedReactor.run(account_id: 3, thing: :two) - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain # Sequence drained; a fresh assign should start at 1 (counters GC'd). n, = adapter.ordered_lock_assign("strict:3") @@ -181,7 +181,7 @@ NonStrictOrderedReactor.run(account_id: 1, thing: :two) NonStrictOrderedReactor.run(account_id: 1, thing: :three) - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(OrderedLockCounters.runs).to eq(%i[one two three]) end @@ -190,16 +190,16 @@ NonStrictOrderedReactor.run(account_id: 2, thing: :one, fail: true) NonStrictOrderedReactor.run(account_id: 2, thing: :two) - first_job = RubyReactor::SidekiqWorkers::Worker.jobs.first - RubyReactor::SidekiqWorkers::Worker.jobs.shift - RubyReactor::SidekiqWorkers::Worker.new.perform(*first_job["args"]) + first_job = RubyReactor::Adapters::Sidekiq::Worker.jobs.first + RubyReactor::Adapters::Sidekiq::Worker.jobs.shift + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*first_job["args"]) expect(adapter.ordered_lock_peek("lenient:2")[:first_failed]).to eq(1) # Now run nonce 2 — it should NOT be skipped despite first_failed > 0. - second_job = RubyReactor::SidekiqWorkers::Worker.jobs.first - RubyReactor::SidekiqWorkers::Worker.jobs.clear - RubyReactor::SidekiqWorkers::Worker.new.perform(*second_job["args"]) + second_job = RubyReactor::Adapters::Sidekiq::Worker.jobs.first + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*second_job["args"]) expect(OrderedLockCounters.runs).to eq(%i[one two]) end @@ -208,7 +208,7 @@ describe "drain + reset" do it "restarts at nonce 1 after a full drain" do 2.times { |i| OrderedReactor.run(account_id: 5, thing: i + 1) } - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain n, = adapter.ordered_lock_assign("orders:5") expect(n).to eq(1) @@ -252,8 +252,8 @@ # Batch A: a single nonce, run to completion so the key fully drains # (counters GC'd, generation epoch kept). OrderedReactor.run(account_id: 77, thing: :a) - straggler = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.drain + straggler = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.drain expect(adapter.ordered_lock_peek("orders:77")).to be_ordered_lock_drained # Batch B reuses nonce 1 under a NEW epoch; leave its job queued. @@ -262,7 +262,7 @@ OrderedLockCounters.runs.clear # Replay batch A's straggler (same serialized context -> stale epoch). - RubyReactor::SidekiqWorkers::Worker.new.perform(*straggler["args"]) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*straggler["args"]) # It must neither run its step nor advance batch B's cursor. expect(OrderedLockCounters.runs).to be_empty @@ -275,15 +275,15 @@ # Batch A: single nonce, run to completion (counters GC'd, epoch kept). OrderedReactor.run(account_id: 78, thing: :a) - duplicate = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.drain + duplicate = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.drain expect(adapter.ordered_lock_peek("orders:78")).to be_ordered_lock_drained # Sidekiq at-least-once redelivery of nonce 1 in the drain->next-batch # window: SAME epoch, so the stale-batch fence cannot catch it. Its # terminal advance (my == last + 1 against the GC'd cursor) must not # resurrect last_completed. - RubyReactor::SidekiqWorkers::Worker.new.perform(*duplicate["args"]) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*duplicate["args"]) expect(adapter.ordered_lock_peek("orders:78")[:last_completed]).to eq(0) # Batch B: nonce 2 alone must still snooze behind nonce 1 (pre-fix the @@ -292,13 +292,13 @@ OrderedReactor.run(account_id: 78, thing: :b2) # nonce 2 OrderedLockCounters.runs.clear - second = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.jobs.clear - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in).and_call_original - RubyReactor::SidekiqWorkers::Worker.new.perform(*second["args"]) + second = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in).and_call_original + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*second["args"]) expect(OrderedLockCounters.runs).to be_empty - expect(RubyReactor::SidekiqWorkers::Worker).to have_received(:perform_in) + expect(RubyReactor::Adapters::Sidekiq::Worker).to have_received(:perform_in) end it "does not downgrade an already-completed context when a stale duplicate is redelivered" do @@ -306,9 +306,9 @@ RubyReactor.configuration.lock_snooze_jitter = 0 OrderedReactor.run(account_id: 88, thing: :a) - duplicate = RubyReactor::SidekiqWorkers::Worker.jobs.last + duplicate = RubyReactor::Adapters::Sidekiq::Worker.jobs.last ctx_id = duplicate["args"].first # identity-only payload: args.first IS the id - RubyReactor::SidekiqWorkers::Worker.drain + RubyReactor::Adapters::Sidekiq::Worker.drain expect(adapter.retrieve_context(ctx_id, "OrderedReactor")["status"]).to eq("completed") # New batch bumps the epoch, so the duplicate's gate resolves to stale. @@ -316,7 +316,7 @@ # Redeliver the completed job. Its stale-batch short-circuit must NOT # overwrite the stored :completed record with :skipped. - RubyReactor::SidekiqWorkers::Worker.new.perform(*duplicate["args"]) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*duplicate["args"]) expect(adapter.retrieve_context(ctx_id, "OrderedReactor")["status"]).to eq("completed") end @@ -330,20 +330,20 @@ OrderedLockWithLockReactor.run(account_id: 1, thing: :a) # nonce 1 OrderedLockWithLockReactor.run(account_id: 1, thing: :b) # nonce 2 - jobs = RubyReactor::SidekiqWorkers::Worker.jobs + jobs = RubyReactor::Adapters::Sidekiq::Worker.jobs _job1 = jobs.shift job2 = jobs.shift # Put nonce 2 back; do NOT execute nonce 1. Ordered gate must reject # nonce 2 since last_completed == 0. - RubyReactor::SidekiqWorkers::Worker.jobs.clear - RubyReactor::SidekiqWorkers::Worker.jobs << job2 + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear + RubyReactor::Adapters::Sidekiq::Worker.jobs << job2 RubyReactor.configuration.lock_snooze_base_delay = 0.01 RubyReactor.configuration.lock_snooze_jitter = 0 RubyReactor.configuration.lock_snooze_max_attempts = 2 - expect { RubyReactor::SidekiqWorkers::Worker.drain }.not_to raise_error + expect { RubyReactor::Adapters::Sidekiq::Worker.drain }.not_to raise_error # The lock was never acquired because the ordered gate rejected first. expect(adapter.lock_info("lock:combo:1")).to be_nil @@ -366,11 +366,11 @@ OrderedLockWithLockReactor.run(account_id: 1, thing: :one) # nonce 1 OrderedLockWithLockReactor.run(account_id: 1, thing: :two) # nonce 2 - job1 = RubyReactor::SidekiqWorkers::Worker.jobs.first - RubyReactor::SidekiqWorkers::Worker.jobs.clear + job1 = RubyReactor::Adapters::Sidekiq::Worker.jobs.first + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear # Drive nonce 1 at the cap so the next snooze escalates. - RubyReactor::SidekiqWorkers::Worker.new.perform(*job1["args"].first(2), 2) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*job1["args"].first(2), 2) state = adapter.ordered_lock_peek("combo:1") # Escalation must advance the cursor (nonce 1 no longer in flight) AND @@ -397,13 +397,13 @@ # nonce 1 + 2; run nonce 2's job alone — its ordered gate raises WaitError. OrderedReactor.run(account_id: 555, thing: :first) OrderedReactor.run(account_id: 555, thing: :second) - job2 = RubyReactor::SidekiqWorkers::Worker.jobs.last - RubyReactor::SidekiqWorkers::Worker.jobs.clear + job2 = RubyReactor::Adapters::Sidekiq::Worker.jobs.last + RubyReactor::Adapters::Sidekiq::Worker.jobs.clear RubyReactor.configuration.lock_snooze_base_delay = 0.01 RubyReactor.configuration.lock_snooze_jitter = 0 - allow(RubyReactor::SidekiqWorkers::Worker).to receive(:perform_in) - RubyReactor::SidekiqWorkers::Worker.new.perform(*job2["args"]) + allow(RubyReactor::Adapters::Sidekiq::Worker).to receive(:perform_in) + RubyReactor::Adapters::Sidekiq::Worker.new.perform(*job2["args"]) expect(events).to include(:snooze_reactor) expect(events).not_to include(:failed_reactor) diff --git a/spec/ruby_reactor/map/dispatcher_spec.rb b/spec/ruby_reactor/map/dispatcher_spec.rb index 69a54941..7be97f08 100644 --- a/spec/ruby_reactor/map/dispatcher_spec.rb +++ b/spec/ruby_reactor/map/dispatcher_spec.rb @@ -4,7 +4,7 @@ RSpec.describe RubyReactor::Map::Dispatcher do let(:storage) { instance_double(RubyReactor::Storage::RedisAdapter) } - let(:async_router) { class_double(RubyReactor::SidekiqAdapter) } + let(:async_router) { class_double(RubyReactor::Adapters::Sidekiq::Router) } before do allow(RubyReactor.configuration).to receive_messages(storage_adapter: storage, async_router: async_router) diff --git a/spec/ruby_reactor/step/map_step_spec.rb b/spec/ruby_reactor/step/map_step_spec.rb index b3517539..da0522df 100644 --- a/spec/ruby_reactor/step/map_step_spec.rb +++ b/spec/ruby_reactor/step/map_step_spec.rb @@ -26,7 +26,7 @@ let(:storage_adapter) { RubyReactor.configuration.storage_adapter } # Async router is fine to be mocked as we don't want to actually enqueue sidekiq jobs here unless necessary - let(:async_router) { class_double(RubyReactor::SidekiqAdapter) } + let(:async_router) { class_double(RubyReactor::Adapters::Sidekiq::Router) } before do allow(RubyReactor.configuration).to receive(:async_router).and_return(async_router) diff --git a/spec/ruby_reactor/sweeper_entrypoints_spec.rb b/spec/ruby_reactor/sweeper_entrypoints_spec.rb index a5aa2803..e3a15e04 100644 --- a/spec/ruby_reactor/sweeper_entrypoints_spec.rb +++ b/spec/ruby_reactor/sweeper_entrypoints_spec.rb @@ -2,12 +2,12 @@ require "spec_helper" require "sidekiq/testing" -require "ruby_reactor/sidekiq_workers/sweeper_worker" +require "ruby_reactor/adapters/sidekiq/sweeper_worker" # Module-level sweeper entrypoints: the host kick (start_sweeper!) and the # synchronous escape hatch (sweep_once). RSpec.describe "RubyReactor sweeper entrypoints" do - let(:worker) { RubyReactor::SidekiqWorkers::SweeperWorker } + let(:worker) { RubyReactor::Adapters::Sidekiq::SweeperWorker } before do worker.clear diff --git a/spec/ruby_reactor/telemetry_spec.rb b/spec/ruby_reactor/telemetry_spec.rb index 7dc16031..1f23d643 100644 --- a/spec/ruby_reactor/telemetry_spec.rb +++ b/spec/ruby_reactor/telemetry_spec.rb @@ -405,8 +405,8 @@ class TelemetryAsyncMapRetryReactor < RubyReactor::Reactor TelemetryAsyncRetryReactor.new.run 5.times do - RubyReactor::SidekiqWorkers::Worker.drain - break if RubyReactor::SidekiqWorkers::Worker.jobs.empty? + RubyReactor::Adapters::Sidekiq::Worker.drain + break if RubyReactor::Adapters::Sidekiq::Worker.jobs.empty? end spans = exporter.finished_spans @@ -428,10 +428,10 @@ class TelemetryAsyncMapRetryReactor < RubyReactor::Reactor it "traces a requeued async map element attempt as ERROR and the successful attempt as OK" do TelemetryAsyncMapRetryReactor.new.run(items: [3], fail_until_attempt: 2) 8.times do - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain - break if RubyReactor::SidekiqWorkers::MapElementWorker.jobs.empty? && - RubyReactor::SidekiqWorkers::MapCollectorWorker.jobs.empty? + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain + break if RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.empty? && + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.jobs.empty? end spans = exporter.finished_spans diff --git a/spec/ruby_reactor_spec.rb b/spec/ruby_reactor_spec.rb index 4c0e871c..5b41dda3 100644 --- a/spec/ruby_reactor_spec.rb +++ b/spec/ruby_reactor_spec.rb @@ -812,7 +812,7 @@ class TestRetryOuterReactor < RubyReactor::Reactor end before do - allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::SidekiqAdapter) + allow(RubyReactor::Configuration.instance).to receive(:async_router).and_return(RubyReactor::Adapters::Sidekiq::Router) end it "returns job_id and intermediate_results correctly" do diff --git a/spec/single_worker_map_spec.rb b/spec/single_worker_map_spec.rb index 3f3de676..64fb3ecb 100644 --- a/spec/single_worker_map_spec.rb +++ b/spec/single_worker_map_spec.rb @@ -17,11 +17,11 @@ # An async map without batch_size now fans out per-element workers # (batch_size defaults to the full source size) instead of a single # MapExecutionWorker. One worker per element. - expect(RubyReactor::SidekiqWorkers::MapElementWorker.jobs.size).to eq(3) + expect(RubyReactor::Adapters::Sidekiq::MapElementWorker.jobs.size).to eq(3) # Process the elements and the collector that aggregates them. - RubyReactor::SidekiqWorkers::MapElementWorker.drain - RubyReactor::SidekiqWorkers::MapCollectorWorker.drain + RubyReactor::Adapters::Sidekiq::MapElementWorker.drain + RubyReactor::Adapters::Sidekiq::MapCollectorWorker.drain # Verify result in Redis storage = RubyReactor.configuration.storage_adapter diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 1194ca0f..527937cc 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -67,7 +67,7 @@ def redis RubyReactor.configure do |config| config.storage.adapter = :redis config.storage.redis_url = REDIS_TEST_URL - config.async_router = RubyReactor::SidekiqAdapter + config.async_router = RubyReactor::Adapters::Sidekiq::Router end RSpec.configure do |config| diff --git a/specs/active_job.md b/specs/active_job.md new file mode 100644 index 00000000..34661c46 --- /dev/null +++ b/specs/active_job.md @@ -0,0 +1,259 @@ +# Pluggable Background Adapters (Sidekiq → ActiveJob) + +## Goal + +Today RubyReactor is hardwired to Sidekiq for everything async: enqueuing, +resuming, map-element fan-out/collection, and the recovery sweeper. We want +to support other background processors — starting with ActiveJob — without +duplicating the reactor-resume/snooze/escalate logic per adapter. + +## What's already abstracted (good news) + +The **enqueue side** is already behind a seam: + +- `RubyReactor.configuration.async_router` (default `RubyReactor::SidekiqAdapter`, + [configuration.rb:106-108](../lib/ruby_reactor/configuration.rb#L106-L108)) is the + only thing the core engine calls to go async. Call sites: + [reactor.rb:117](../lib/ruby_reactor/reactor.rb#L117), + [reactor.rb:319](../lib/ruby_reactor/reactor.rb#L319), + [step/map_step.rb:274](../lib/ruby_reactor/step/map_step.rb#L274), + [step/map_step.rb:285](../lib/ruby_reactor/step/map_step.rb#L285), + [map/dispatcher.rb:177](../lib/ruby_reactor/map/dispatcher.rb#L177), + [map/element_executor.rb:155](../lib/ruby_reactor/map/element_executor.rb#L155), + [map/element_executor.rb:176](../lib/ruby_reactor/map/element_executor.rb#L176), + [executor/retry_manager.rb:59,80](../lib/ruby_reactor/executor/retry_manager.rb#L59), + [executor/step_executor.rb:219](../lib/ruby_reactor/executor/step_executor.rb#L219), + [sweeper.rb:48](../lib/ruby_reactor/sweeper.rb#L48), + [map/sweeper.rb:99](../lib/ruby_reactor/map/sweeper.rb#L99). +- The router contract is 5 class methods on `SidekiqAdapter` + ([sidekiq_adapter.rb](../lib/ruby_reactor/sidekiq_adapter.rb)): + `perform_async`, `perform_in`, `perform_map_element_async`, + `perform_map_element_in`, `perform_map_collection_async`. All return + `RubyReactor::AsyncResult`. +- An adapter for any other queueing backend just needs to implement that + same 5-method contract and assign it to `config.async_router`. **This part + needs no rework.** + +## What's NOT abstracted (the actual gap) + +The **worker/job side** — the classes the queue invokes — bakes Sidekiq in +directly. **Decision: these move into `RubyReactor::Adapters::Sidekiq::*`** +(renamed from `RubyReactor::SidekiqWorkers::*`), with a sibling +`RubyReactor::Adapters::ActiveJob::*` for the new adapter — see +[Namespace](#namespace) below. + +| Class (today) | File | Sidekiq coupling | +|---|---|---| +| `SidekiqWorkers::Worker` | [worker.rb](../lib/ruby_reactor/sidekiq_workers/worker.rb) | `include ::Sidekiq::Worker`, `sidekiq_options`, `sidekiq_retries_exhausted`, and internally calls `self.class.perform_in(...)` to reschedule snoozes (lines 116, 91-117) | +| `SidekiqWorkers::MapElementWorker` | [map_element_worker.rb](../lib/ruby_reactor/sidekiq_workers/map_element_worker.rb) | `include ::Sidekiq::Worker`; body is a 1-line delegate to `Map::ElementExecutor.perform` | +| `SidekiqWorkers::MapCollectorWorker` | [map_collector_worker.rb](../lib/ruby_reactor/sidekiq_workers/map_collector_worker.rb) | same — 1-line delegate to `Map::Collector.perform` | +| `SidekiqWorkers::SweeperWorker` | [sweeper_worker.rb](../lib/ruby_reactor/sidekiq_workers/sweeper_worker.rb) | `include ::Sidekiq::Worker`, `sidekiq_options retry: false`, self-reschedules via `perform_in` / class-level `perform_in` in `schedule_next` | + +Plus test-side coupling: + +- [rspec/sidekiq_helpers.rb](../lib/ruby_reactor/rspec/sidekiq_helpers.rb) hardcodes + the 3 Sidekiq worker classes and `Sidekiq::Testing` fake-mode draining. +- [rspec/test_subject.rb:194-203,433](../lib/ruby_reactor/rspec/test_subject.rb#L194-L203) + gates job-processing on `defined?(Sidekiq::Testing)` and forces + `async_router` back to `SidekiqAdapter`. +- [ruby_reactor.rb:22-27](../lib/ruby_reactor.rb#L22-L27) optionally requires + `sidekiq`; [ruby_reactor.rb:359](../lib/ruby_reactor.rb#L359) calls + `SidekiqWorkers::SweeperWorker.schedule_next` directly from + `RubyReactor.start_sweeper!`. + +The real logic worth extracting lives almost entirely in +`SidekiqWorkers::Worker#perform` (~65 lines): rehydrate context from +storage, deserialize, resolve `reactor_class`, mark +`inline_async_execution`, run `Executor#resume_execution`, and on +lock/semaphore/rate-limit/ordered-lock contention either snooze (re-enqueue +with a computed delay) or escalate to `failed`. None of that is +Sidekiq-specific — it only *touches* Sidekiq via `self.class.perform_in` to +reschedule. + +## Proposal + +### 1. Normalize the enqueue API at the job-class boundary, not in the shared logic + +`Sidekiq::Worker` gives every job class `.perform_async` / `.perform_in` for +free. ActiveJob doesn't — it has `.perform_later` and +`.set(wait: delay).perform_later`. Rather than teach the shared logic two +different reschedule calls, give every framework-specific job class the +same two class methods, so the shared mixin can keep calling +`self.class.perform_in(...)` unchanged: + +```ruby +module RubyReactor + module Adapters + module ActiveJob + module Compat + def perform_async(*args) = perform_later(*args) + def perform_in(delay, *args) = set(wait: delay).perform_later(*args) + end + end + end +end +``` + +### 2. Extract `RubyReactor::Worker` — the framework-agnostic mixin + +Move the body of `Adapters::Sidekiq::Worker#perform` (and its private snooze/ +escalate/deserialization-failure helpers) into a plain module with no +`Sidekiq` reference: + +```ruby +module RubyReactor + module Worker + def perform(context_id, reactor_class_name = nil, snooze_count = 0) + # ...exact same logic as today's SidekiqWorkers::Worker#perform... + end + + private + # handle_snooze, compute_snooze_delay, hinted_retry?, escalate_snooze, + # log_infrastructure_failure, handle_deserialization_failure, + # build_failed_context_payload — unchanged, moved verbatim. + end +end +``` + +`Adapters::Sidekiq::Worker` then becomes: + +```ruby +module RubyReactor + module Adapters + module Sidekiq + class Worker + include ::Sidekiq::Worker + include RubyReactor::Worker + + sidekiq_options retry: RubyReactor.configuration.sidekiq_retry_count, + dead: false, queue: RubyReactor.configuration.sidekiq_queue + end + end + end +end +``` + +And a new `Adapters::ActiveJob::Worker`: + +```ruby +module RubyReactor + module Adapters + module ActiveJob + class Worker < ::ActiveJob::Base + extend Compat + include RubyReactor::Worker + + queue_as { RubyReactor.configuration.sidekiq_queue } # or a renamed generic config + end + end + end +end +``` + +Same pattern applies to `MapElementWorker` / `MapCollectorWorker` — they're +already a 1-line delegate, so genericizing is just swapping the include; no +logic to extract. + +### 3. Sweeper + +`SweeperWorker`'s window-claim-lock + self-reschedule logic +([sweeper_worker.rb:30-70](../lib/ruby_reactor/sidekiq_workers/sweeper_worker.rb#L30-L70)) +is also framework-agnostic except for the `perform_in` call in +`schedule_next`. Extract the same way into `RubyReactor::SweeperJob`, +included into both `Adapters::Sidekiq::SweeperWorker` (`sidekiq_options retry: false`) +and an `Adapters::ActiveJob::SweeperWorker`. `RubyReactor.start_sweeper!` +([ruby_reactor.rb:356-360](../lib/ruby_reactor.rb#L356-L360)) needs to call +through whichever sweeper job class matches the configured adapter instead +of hardcoding `Adapters::Sidekiq::SweeperWorker`. + +### 4. New `RubyReactor::Adapters::ActiveJob::Router` + +Mirrors today's `RubyReactor::Adapters::Sidekiq::Router` (renamed from +`SidekiqAdapter`) exactly — same 5 methods, just pointing at the +`Adapters::ActiveJob::*` job classes instead of `Adapters::Sidekiq::*`. No +changes needed to any core call site; swap is purely +`config.async_router = RubyReactor::Adapters::ActiveJob::Router`. + +### 5. Optional config sugar + +`configuration.rb` already does this pattern for storage +([configuration.rb:114-121](../lib/ruby_reactor/configuration.rb#L114-L121)): +a single `storage.adapter` symbol resolves to a concrete adapter instance. +Could mirror it — `config.queue_adapter = :sidekiq | :active_job` resolving +both `async_router` and the sweeper job class — but this is sugar, not +required for the feature to work. Confirm whether you want it. + +### 6. Test helpers + +`rspec/sidekiq_helpers.rb` and the `Sidekiq::Testing` branch in +`rspec/test_subject.rb` only fire for Sidekiq today. For ActiveJob, Rails +already ships `ActiveJob::TestHelper` (`perform_enqueued_jobs`, +`have_enqueued_job`) which covers most of this generically. We'd still want +something equivalent to `drain_async_jobs` (loops until self-rescheduling +jobs — e.g. ordered-lock snoozes — stop producing new ones), so +`test_subject.rb`'s job-processing gate needs to branch on which +testing framework is active (or be driven by `configuration.async_router`) +rather than hardcoding `defined?(Sidekiq::Testing)`. + +### 7. Loading + +`ruby_reactor.rb:22-27` optionally `require "sidekiq"`. Add the same +optional-require pattern for `active_job`, so neither dependency is forced +on users who only need one. + +## Namespace + +**Decided: `RubyReactor::Adapters::Sidekiq::*` / `RubyReactor::Adapters::ActiveJob::*`.** + +Renames/moves required (mechanical, but touches every reference): + +| Today | Becomes | +|---|---| +| `RubyReactor::SidekiqAdapter` | `RubyReactor::Adapters::Sidekiq::Router` | +| `RubyReactor::SidekiqWorkers::Worker` | `RubyReactor::Adapters::Sidekiq::Worker` | +| `RubyReactor::SidekiqWorkers::MapElementWorker` | `RubyReactor::Adapters::Sidekiq::MapElementWorker` | +| `RubyReactor::SidekiqWorkers::MapCollectorWorker` | `RubyReactor::Adapters::Sidekiq::MapCollectorWorker` | +| `RubyReactor::SidekiqWorkers::SweeperWorker` | `RubyReactor::Adapters::Sidekiq::SweeperWorker` | +| (new) | `RubyReactor::Adapters::ActiveJob::Router` | +| (new) | `RubyReactor::Adapters::ActiveJob::{Worker,MapElementWorker,MapCollectorWorker,SweeperWorker}` | + +Files move from `lib/ruby_reactor/sidekiq_workers/*.rb` + +`lib/ruby_reactor/sidekiq_adapter.rb` to +`lib/ruby_reactor/adapters/sidekiq/*.rb` (Zeitwerk-driven, so the directory +move IS the rename — no manual `module` boilerplate beyond nesting). New +ActiveJob side lives in `lib/ruby_reactor/adapters/active_job/*.rb`. + +References that need updating for the rename: +- [configuration.rb:107](../lib/ruby_reactor/configuration.rb#L107) — `@async_router ||= RubyReactor::SidekiqAdapter` +- [ruby_reactor.rb:359](../lib/ruby_reactor.rb#L359) — `SidekiqWorkers::SweeperWorker.schedule_next` +- [rspec/sidekiq_helpers.rb](../lib/ruby_reactor/rspec/sidekiq_helpers.rb) — `worker_classes` list +- [rspec/test_subject.rb:196](../lib/ruby_reactor/rspec/test_subject.rb#L196) — stub target +- [spec/ruby_reactor/sidekiq_workers/worker_spec.rb](../spec/ruby_reactor/sidekiq_workers/worker_spec.rb), + [spec/ruby_reactor/sidekiq_workers/sweeper_worker_spec.rb](../spec/ruby_reactor/sidekiq_workers/sweeper_worker_spec.rb) — + `described_class` references, move to `spec/ruby_reactor/adapters/sidekiq/` + +## Retry config — decided: generic + +`sidekiq_retry_count` / `sidekiq_queue` rename to `config.job_retry_count` / +`config.queue_name`, used by both adapters: + +- `Adapters::Sidekiq::Worker` → `sidekiq_options retry: config.job_retry_count, dead: false, queue: config.queue_name` +- `Adapters::ActiveJob::Worker` → maps to `retry_on StandardError, attempts: config.job_retry_count` (infra + failures only — reactor-specific errors are already caught by the shared + snooze/escalate logic in `RubyReactor::Worker` before they'd ever reach + the framework's retry layer) and `queue_as { config.queue_name }`. +- `sidekiq_retries_exhausted` (currently an empty hook) → ActiveJob + equivalent is `retry_on ... do |job, error| ... end` / `discard_on`. + +`configuration.rb` changes: add `job_retry_count`/`queue_name` as the +canonical attrs; keep `sidekiq_retry_count`/`sidekiq_queue` as deprecated +aliases delegating to the new names so existing configs don't break. + +None of this requires touching the enqueue-side call sites in +`reactor.rb`, `executor/*`, `map/*` — that seam already works (it just calls +through `config.async_router`, whose value changes, not its call sites). +The work is isolated to: `lib/ruby_reactor/worker.rb` (new), +`lib/ruby_reactor/sweeper_job.rb` (new), the `sidekiq_workers/` → +`adapters/sidekiq/` move + rename, `adapters/active_job/*.rb` (new, 5 +classes: `Router` + 4 job classes), and the test-helper / +`start_sweeper!` branching described above.