RubyReactor

A dynamic, dependency-resolving saga orchestrator for Ruby. Ruby Reactor implements the Saga pattern with compensation-based error handling and DAG-based execution planning. It leverages Sidekiq or ActiveJob for background execution and Redis for state persistence.

Building complex business transactions often results in spaghetti code or brittle "god classes." Ruby Reactor solves this by implementing the Saga Pattern in a lightweight, developer-friendly package. It lets you define workflows as clear, dependency-driven steps without the boilerplate of heavy enterprise frameworks.

The key value is Reliability: if any part of your workflow fails with an error, Ruby Reactor automatically triggers compensation logic to undo previous steps, ensuring your system never ends up in a corrupted half-state. Only an interruption (a signal, an exit, out of memory) runs no rollback code; the run is recorded as aborted so a manual undo can roll it back. Whether you're coordinating microservices or monolith modules, you get atomic-like consistency with background processing built-in.

• DAG-based Execution: Steps are executed based on their dependencies, allowing for parallel execution of independent steps.
• Background Execution: Hand a whole reactor, or everything after a declared step, to a background job via Sidekiq or ActiveJob (so any ActiveJob-compatible queue — Resque, Solid Queue, GoodJob, etc. — works too).
• Async Steps & Reactors: async_step and async_reactor dispatch independent units of work while the reactor keeps running; steps that read their result wait for it.
• Map & Parallel Execution: Iterate over collections in parallel with the map step, distributing work across multiple workers.
• Retries: per-step retry policies (declared on the step class or step block) with exponential, linear, or fixed backoff.
• Compensation: Automatic rollback of completed steps when any error occurs after them, including a raising argument transform or an exception that is not a StandardError.
• Interrupts: Pause and resume workflows to wait for external events (webhooks, user approvals).
• Input Validation: Integrated with dry-validation for robust input checking.
• Distributed Locks, Semaphores, Rate Limits, Periods & Ordered Locks: Coordinate across processes with Redis-backed primitives — exclusive locks for at-most-one-runner, semaphores for capacity caps, fixed-window rate limits for external APIs (single or multi-window like "3/sec AND 100/min"), with_period to dedup reactors to once per calendar bucket, and with_ordered_lock for strict transaction ordering via a monotonically increasing nonce assigned at enqueue. Background jobs snooze on contention with smart retry_after instead of consuming retry budget.

• E-commerce Checkout: Orchestrate inventory reservation, payment authorization, and shipping label generation. If payment fails, automatically release inventory and cancel the shipping request.
• Data Import Pipelines: Ingest optional massive CSVs using map steps to validate and upsert records in parallel. If data validation fails for a chunk, fail fast or collect errors while letting valid chunks succeed.
• Subscription Billing: Coordinate Stripe charges, invoice email generation, and internal entitlement updates. Use interrupts to pause the workflow when 3rd-party APIs are required to continue the workflow or when specific customer approval is needed.

• Features
• Comparison
• Real-World Use Cases
• Installation
• Configuration
• Quick Start
• Defining Steps
• Web Dashboard
  • Rails Installation
• Usage
  • Basic Example: User Registration
  • Background Execution
    • Full Background
    • Background Hand-off
  • Async Steps & Reactors
    • async_step
    • async_reactor
  • Durability & Recovery
  • Interrupts (Pause & Resume)
  • Locks, Semaphores & Ordered Locks
  • Map & Parallel Execution
    • Map with Dynamic Source (ActiveRecord)
  • Input Validation
  • Complex Workflows with Dependencies
  • Error Handling and Compensation
  • Using Pre-defined Schemas
  • Testing
• Documentation
• Future improvements
• Development
• Contributing
• Code of Conduct

Add this line to your application's Gemfile:

gem 'ruby_reactor'

And then execute:

$ bundle install

Or install it yourself as:

$ gem install ruby_reactor

Every setting is optional — RubyReactor ships with the defaults shown. Drop this into an initializer (e.g. config/initializers/ruby_reactor.rb); pasted as-is it changes nothing, so it doubles as a reference of every knob.

Reading the block: lines starting with ## are documentation. Lines starting with a single # (a config.… call) are real settings commented at their default — uncomment one to enable it.

RubyReactor.configure do |config|
  ## === Storage (Redis) ===

  ## Storage adapter. Default: :redis (the only adapter shipped today).
  # config.storage.adapter = :redis

  ## Redis URL. Default: "redis://localhost:6379/0".
  config.storage.redis_url = ENV.fetch("REDIS_URL", "redis://localhost:6379/0")

  ## Extra options passed to Redis.new. Default: {}.
  # config.storage.redis_options = { timeout: 1 }

  ## === Background job backend (Sidekiq by default, or ActiveJob) ===

  ## Queue used by RubyReactor's background worker. Default: :default.
  # config.queue_name = :default

  ## Retry count for infrastructure failures only (deserialization, Redis,
  ## network). Step retries are managed separately. Default: 3.
  # config.job_retry_count = 3

  ## `sidekiq_queue` / `sidekiq_retry_count` still work as deprecated aliases
  ## for `queue_name` / `job_retry_count` above.

  ## === Contention snooze (locks / semaphores / rate limits / ordered locks) ===

  ## When a worker cannot acquire a primitive it re-enqueues itself with
  ## `lock_snooze_base_delay + rand(0..lock_snooze_jitter)` seconds (rate-limit
  ## uses a precise `retry_after_seconds` hint from the error; ordered-lock waits
  ## re-poll at the base delay so a successor catches its blocker finishing fast),
  ## up to `lock_snooze_max_attempts` times before marking the context :failed.
  ## Set max_attempts to :infinity to never give up.
  # config.lock_snooze_base_delay = 5
  # config.lock_snooze_jitter = 5
  # config.lock_snooze_max_attempts = 20

  ## === Waiting on async results (`result(:name)` on async_step / async_reactor) ===

  ## How long a SYNCHRONOUS caller blocks on a still-pending async result before
  ## failing with Error::AsyncWaitTimeoutError. Kept tight: a blocking wait pins
  ## a thread and must stay under typical host timeouts. Default: 30.
  # config.async_wait_timeout = 30

  ## Total parked time allowed when the reader runs INSIDE a worker: the job
  ## re-enqueues itself instead of blocking (locks stay held across the gap),
  ## so this bound can be generous. Measured from the unit's dispatch.
  ## :infinity removes the bound. Default: 3600.
  # config.async_park_timeout = 3600

  ## === Durability & crash recovery (see "Durability & Recovery" below) ===

  ## Retention TTL (seconds) for stored reactor/map state. Must exceed your
  ## worst-case snooze/retry window; re-stamped on every write. Default: 86_400.
  # config.context_ttl = 86_400

  ## TTL (seconds) for the per-context liveness lock. A live worker auto-extends
  ## it; its absence is the sweeper's "worker died" signal. Must exceed the
  ## longest a single step can run without yielding the GIL. Default: 60.
  # config.context_lock_ttl = 60

  ## Minimum seconds between per-step checkpoints within one run. 0 = checkpoint
  ## after every step (strongest guarantee). Raise to coalesce mid-run writes for
  ## long reactors — only safe when steps are idempotent. Default: 0.
  # config.checkpoint_min_interval = 0

  ## Recovery sweeper (the chain is kicked once by `RubyReactor.start_sweeper!`).
  # config.sweeper_enabled = true    # run recovery by default
  # config.sweeper_interval = 30     # seconds between sweeps = recovery-latency bound
  # config.sweeper_limit = 1000      # max contexts/maps inspected per sweep

  ## === Misc ===

  ## Logger. Default: Logger.new($stdout).
  # config.logger = Logger.new($stdout)

  ## Background job router. Default: RubyReactor::Adapters::Sidekiq::Router. Swap in the
  ## built-in ActiveJob adapter (see "Background Execution" below), or point at any
  ## custom adapter — it only needs to respond to
  ## `perform_async(context_id, reactor_class_name, **)`.
  # config.async_router = RubyReactor::Adapters::ActiveJob::Router

  ## === Examples (no default — set these to use the feature) ===

  ## Named rate limits shared across reactors. Reference with `with_rate_limit(:stripe)`.
  # config.rate_limits.register(:stripe, limits: { second: 3, minute: 100 })

  ## OpenTelemetry / custom middlewares. Default: [].
  # config.middlewares = [RubyReactor::OpenTelemetry]
end

You can also leave out the configure block entirely — defaults work for local development against a Redis on localhost:6379.

Crash recovery needs a kick. The sweeper_* settings above only configure the recovery sweeper — they do not start it. Call RubyReactor.start_sweeper! once at boot (ideally from a worker-process startup hook — a Sidekiq on(:startup) hook, or a Rails initializer for ActiveJob) or no crashed reactor will ever resume. See Durability & Recovery.

class HelloReactor < RubyReactor::Reactor
  step :greet do
    run { Success("Hello from Ruby Reactor!") }
  end
  returns :greet
end

result = HelloReactor.run
puts result.value  # => "Hello from Ruby Reactor!"

Note: Examples in this README use inline step blocks where a step is trivial. For production workflows, prefer class-based steps.

RubyReactor supports two ways to define step logic:

Whichever style you use, a step's run returns one of four signals — all exposed as bare helpers in both class steps and inline blocks:

• Success(value) — step succeeded; value flows to dependent steps.
• Failure(error) — step failed; the reactor rolls back completed steps (compensate/undo).
• Halt(reason:) — clean halt: stop the reactor, keep partial progress, no rollback. See Halting a reactor cleanly.
• Skipped(value) — mark this one step skipped. The run continues exactly as for Success — value flows to dependants — and the step is never undone; the execution trace marks it skipped. See Skipping a single step.

One-line helpers end a step immediately from any call depth: success!(value), fail!(error, retry: true), halt!(reason:), skip!(value) — equivalent to returning the matching signal, usable in run, compensate, and undo bodies.

Class steps subclass RubyReactor::Step and implement run, and optionally compensate and undo, as instance methods reading inputs and context:

class ReserveInventoryStep < RubyReactor::Step
  # The step's input contract: enforced before `run` on every execution path.
  input :order, :hash
  # The step's retry policy: applies in every reactor that uses this step.
  retries max_attempts: 3

  def run
    reservation_id = InventoryService.reserve(inputs.order[:items])
    Success(reservation_id: reservation_id)
  end

  def compensate
    InventoryService.release_partial(inputs.order[:items])
    Success()
  end

  def undo
    InventoryService.release(result[:reservation_id])
    Success()
  end
end

class OrderProcessingReactor < RubyReactor::Reactor
  step :reserve_inventory, ReserveInventoryStep do
    argument :order, result(:validate_order)
  end
end

Why prefer class steps?

• Testability — unit-test run, compensate, and undo in isolation without booting the whole reactor
• Composability — share the same step class across multiple reactors and compose larger workflows from small, focused units
• Readability — reactor files stay orchestration-only; business logic lives in named classes instead of growing inline blocks

See Core Concepts — Step Classes for the full reference. Usage examples below mix class and inline steps — inline where the logic is trivial.

RubyReactor ships with a built-in web dashboard to inspect reactor executions, view logs, and retry failed steps. The dashboard is a Rack app (a Roda application) bundled inside the gem with its pre-compiled JS/CSS assets — no extra install or asset build step is required.

Add the gem to your Gemfile:

gem "ruby_reactor"

Then mount the dashboard in your config/routes.rb:

Rails.application.routes.draw do
  # ... other routes
  mount RubyReactor::Web::Application => "/ruby_reactor"
end

That's it — visit /ruby_reactor and the UI loads. RubyReactor::Web::Application is autoloaded by Zeitwerk on first reference, so no extra require is needed.

Because it's a plain Rack app, you can mount it anywhere call(env) is accepted:

# config.ru
require "ruby_reactor/web/application"
run RubyReactor::Web::Application

You can secure the dashboard using standard Rails authentication methods (e.g., wrapping the mount line in an authenticate block with Devise, or in a constraints block).

RubyReactor allows you to define complex workflows as "reactors" with steps that can depend on each other, handle failures with compensations, and validate inputs. Examples in this section mix class steps with inline blocks; see Defining Steps for guidance.

require 'ruby_reactor'

class ValidateEmailStep < RubyReactor::Step
  def run
    email = inputs.email
    email&.include?('@') ? Success(email.strip) : Failure("Email must contain @")
  end
end

class CreateUserStep < RubyReactor::Step
  def run
    Success(
      id: rand(10000),
      email: inputs.email,
      password_hash: inputs.password_hash,
      created_at: Time.now
    )
  end

  def compensate
    Notify.to(inputs.email)
    Success()
  end
end

class UserRegistrationReactor < RubyReactor::Reactor
  input :email
  input :password

  step :validate_email, ValidateEmailStep do
    argument :email, input(:email)
  end

  step :hash_password do
    argument :password, input(:password)
    run do |inputs, _context|
      require 'digest'
      Success(Digest::SHA256.hexdigest(inputs.password))
    end
  end

  step :create_user, CreateUserStep do
    argument :email, result(:validate_email)
    argument :password_hash, result(:hash_password)
  end

  step :notify_user do
    argument :email, result(:validate_email)
    wait_for :create_user

    run do |inputs, _context|
      Email.send!(inputs.email, "verify your email")
      Success()
    end

    compensate do |_error, inputs, _context|
      Email.send("support@acme.com", "Email verification for #{inputs.email} couldn't be sent")
      Success()
    end
  end

  returns :create_user
end

# Run the reactor
result = UserRegistrationReactor.run(
  email: 'alice@example.com',
  password: 'secret123'
)

if result.success?
  puts "User created: #{result.value[:email]}"
else
  puts "Failed: #{result.error}"
end

Execute reactors in the background using Sidekiq or ActiveJob. The backend is chosen via config.async_router (defaults to the Sidekiq adapter); to run on ActiveJob instead:

RubyReactor.configure do |config|
  config.async_router = RubyReactor::Adapters::ActiveJob::Router
end

That's the only switch — everything below (background execution, async_step / async_reactor, durability, retries, snoozing) works identically on either backend.

class BackgroundReactor < RubyReactor::Reactor
  background all: true # Entire reactor runs in background

  step :long_running_task do
    run { perform_heavy_work }
  end
end

# Returns immediately with DispatchResult
result = BackgroundReactor.run(params)

A reactor can name one point where execution stops running in the caller's process and is handed to a worker. Everything before it runs in the caller; everything after it runs in a single background job. The cut point is nameable from either side.

class CreateUserReactor < RubyReactor::Reactor
  input :params

  step :validate_inputs do
    run { |inputs| validate(inputs.params) }
  end

  step :create_user do
    argument :params, result(:validate_inputs)
    run { |inputs| User.create(inputs.params) }
  end

  # :create_user is the LAST step to run in the calling process.
  # Equivalently here: `background before: :open_account`.
  background after: :create_user

  step :open_account do
    argument :user, result(:create_user)
    run { |inputs| Bank.open_account(inputs.user) }
  end

  step :report_new_user do
    argument :user, result(:create_user)
    wait_for :open_account
    run { |inputs| Analytics.track(inputs.user) }
  end
end

# Usage
def create(params)
  # Returns an DispatchResult immediately once :create_user completes
  result = CreateUserReactor.run(params)

  # Access synchronous results immediately
  user = result.intermediate_results[:create_user]

  # do something with user
end

after: :x guarantees :x runs in the calling process and is the last to do so. before: :x guarantees :x runs in the worker and is the first to do so. They coincide in a linear chain; in a DAG they pin different steps, so pick whichever step you actually need pinned. Compensation is unchanged — background changes where code runs, not the saga contract.

Breaking change: the per-step async true flag has been removed. It was ambiguous — only the first flagged step in a reactor ever took effect and the rest were silently ignored — and it now raises at class-definition time. The exact replacement for a flagged step :x is background before: :x. The same applies to async inside a compose block.

Breaking change: whole-reactor async true has also been removed — it named the same idea as background's cut point with a different word, right next to async_step/async_reactor, whose names mean something else. It now raises at class-definition time. The exact replacement is background all: true.

async_step and async_reactor dispatch an independent unit of work to its own job while the reactor keeps running. Unlike background, they do not move the reactor itself — they are separate units with their own outcome and compensation.

Where background relocates the rest of a reactor, async_step dispatches one step's work to its own job while the reactor keeps running every other ready step.

class SignupReactor < RubyReactor::Reactor
  input :email

  async_step :send_email do
    argument :to, input(:email)
    run { |inputs| Mailer.welcome(inputs.to).deliver_now; Success(:sent) }
  end

  # Does NOT wait — it has no dependency on :send_email.
  step :record_signup do
    argument :email, input(:email)
    run { |inputs| Success(Signup.create!(email: inputs.email)) }
  end

  # DOES wait, because it reads the result.
  step :confirm_delivery do
    argument :delivery, result(:send_email)
    run { |inputs| Success("confirmed") }
  end
end

Reading result(:send_email) is what makes a step wait — never unbounded, and never thread-hungry: a synchronous caller blocks, bounded by config.async_wait_timeout (default 30s); a reader inside a worker parks instead — after a short grace the job re-enqueues itself, freeing the thread and keeping any held lock/semaphore checked out, bounded by config.async_park_timeout (default 1h) so legitimately slow units get their time. On success the reader gets the raw value; on failure it gets the Failure object, so it can inspect it and decide.

Compensation is opt-in. If a dispatched step fails and nothing reads its result, the reactor is not compensated — it was dispatched precisely so the reactor would not depend on it. The unit's own compensate does run: once, in the unit's job, after its final attempt fails, recorded on the unit's record as compensation. A reader that returns Failure compensates itself and undoes the reactor's completed steps; the unit is not compensated again. The independence cuts both ways: async dispatches never enter the parent's undo stack, so a parent rolling back for its own reasons never "undoes" a unit that runs (and may still succeed) elsewhere. An inline undo on an async_step would never run, so it raises at class-definition time (a step class's undo is warned about).

class SignupReactor < RubyReactor::Reactor
  input :user_id

  # Fire-and-forget: nothing reads it, so its failure never affects this reactor.
  async_reactor :backfill_profile, ProfileBackfillReactor do
    argument :user_id, input(:user_id)
  end

  async_reactor :provision_account, AccountProvisioningReactor do
    argument :user_id, input(:user_id)
  end

  step :verify do
    argument :account, result(:provision_account)   # waits for the child
    run do |inputs|
      inputs.account.success? ? Success(inputs.account.value) : Failure(inputs.account.error)
    end
  end
end

The child is an ordinary, independently addressable execution, linked to the parent by execution id (drillable in the dashboard) but excluded from its compensation graph. Reach for compose instead when the child belongs to this unit of work — its result is available immediately and its failure rolls the parent back.

Lock ownership is never shared across the async boundary: a child declaring a key the parent holds fails at dispatch with an explanatory error rather than deadlocking. See Background & Async Execution for the full rules.

Background reactors are durable: state lives in Redis, not in the job payload. Before any background job is enqueued the root context is persisted, and after every completed step a checkpoint advances the stored blob — so a crash re-runs at most one step, never the whole reactor. Each running reactor also holds a short liveness lock that a live worker auto-extends; its absence is how a dead worker is detected. A run executing in the caller's process (a synchronous Reactor.run, an inline continue) holds the same lock while it executes, so a sweep never runs it a second time, however long a step takes; if that process is killed, the lock lapses within context_lock_ttl and the run is recovered like a worker's. A worker takes the run's lock before it reads the run, so it always resumes from the last save of whoever held it.

Recovery is not automatic until you start the sweeper. A crashed worker's reactor only resumes when the recovery sweeper notices the lapsed liveness lock and re-enqueues it. The sweeper is a self-rescheduling chain — kick it once per process boot:

The recommended spot is a worker-process startup hook, so only the worker process runs recovery (not your web/console/client processes). On Sidekiq:

# config/initializers/sidekiq.rb
Sidekiq.configure_server do |config|
  config.on(:startup) { RubyReactor.start_sweeper! }
end

ActiveJob has no equivalent server-only hook — call it from wherever your queue adapter's worker process boots (e.g. bin/jobs for Solid Queue, or a dedicated initializer guarded by an env var so it doesn't also run in web processes):

# config/initializers/ruby_reactor.rb
RubyReactor.start_sweeper! if ENV["RUBY_REACTOR_WORKER"]

Anywhere that runs once at boot works too (idempotent, so it's safe even if every process calls it) — e.g. an unconditional Rails initializer:

# config/initializers/ruby_reactor.rb
RubyReactor.start_sweeper!

That's all that's required: start_sweeper! is idempotent (safe to call on every boot — duplicate kicks collapse to one chain), runs both the top-level reactor sweeper and the map sweeper every config.sweeper_interval seconds, and stops if you set config.sweeper_enabled = false. The interval is your recovery-latency bound. Both cover a fan-out map's distributed rollback too: the map sweeper re-dispatches a lost element rollback job, and the reactor sweeper re-enqueues a rolling_back run (like a running one) whose resume was lost.

Sidekiq Enterprise super_fetch compatibility: the chain is safe under reliable fetch. super_fetch re-runs a job whose worker died mid-execution, so a tick that crashes after enqueuing its successor but before acking would, with naive single-flight, be recovered alongside that successor and fork the chain (doubling every interval). RubyReactor avoids this: it never relies on "one job in the chain" — each next tick is claimed by a per-time-window lock, so a super_fetch-recovered tick computes the same window, loses the claim, and collapses back to a single successor. The startup hook above is likewise idempotent across multiple super_fetch server processes.

Prefer your own scheduler? Set config.sweeper_enabled = false (which makes start_sweeper! a no-op) and drive recovery from cron, a Kubernetes CronJob, sidekiq-cron, sidekiq-scheduler, or Rails recurring tasks. Each tick is one call:

RubyReactor.sweep_once # => { reactors: <n re-enqueued>, maps: <n recovered> }

For example, a rake task a system cron / CronJob can invoke:

# lib/tasks/ruby_reactor.rake
namespace :ruby_reactor do
  task sweep: :environment do
    RubyReactor.sweep_once
  end
end

Pause execution to wait for external events like webhooks or user approvals.

class ApprovalReactor < RubyReactor::Reactor
  step :submit_request do
    run { |inputs| RequestService.submit(inputs.to_h) }
  end

  interrupt :wait_for_manager do
    wait_for :submit_request
    # Resume using this ID
    correlation_id { |ctx| "req-#{ctx.result(:submit_request)[:id]}" }
  end

  step :process_decision do
    argument :decision, result(:wait_for_manager)
    run do |inputs| 
      inputs.decision == 'approved' ? Success() : Failure("Rejected")
    end
  end
end

# Usage:
# 1. Start execution
execution = ApprovalReactor.run(params) # => Returns Paused status

# 2. Later, resume it via correlation ID
ApprovalReactor.continue_by_correlation_id(
  correlation_id: "req-123",
  payload: "approved",
  step_name: :wait_for_manager
)

An interrupt inside a composed child pauses the top-level run; resume it there by the step path, e.g. step_name: [:approval, :wait_for_manager]. See Interrupts inside composed reactors. continue validates the payload in the calling process first: an invalid one is answered with its validation failure and nothing is stored. A valid one claims its interrupt, so of two resumes of the same interrupt exactly one is accepted (the other raises "already resumed"). When the run is busy, another interrupt's resume is still running, or the reactor's with_lock/with_semaphore is held, the accepted resume is handed to a worker and continue returns a DispatchResult instead of raising, so a webhook's resume is never lost. See Resuming Execution.

Coordinate across processes with Redis-backed primitives:

• with_lock — at-most-one runner per key at a time (concurrency control).
• with_semaphore — cap total concurrent runners per key (capacity control).
• with_rate_limit — fixed-window rate limit, single or multi-window ("3/sec AND 100/min"). Inline per-reactor, or reference a named limit registered once in RubyReactor.configure and shared across reactors.
• with_period — run at most once per calendar bucket (dedup / once-per-day, once-per-month, etc).
• with_ordered_lock — strict transaction ordering via a monotonically increasing nonce assigned at enqueue. Workers can only proceed when their nonce equals last_completed + 1.

An interrupt's resume that meets the reactor's held lock or semaphore is accepted and deferred to a worker, which waits for it (continue returns a DispatchResult); a run already admitted is never failed by the snooze limit for waiting.

class RefundOrderReactor < RubyReactor::Reactor
  input :order_id

  # Only one refund per order at a time. Auto-extend keeps the TTL fresh while
  # the reactor runs, so long steps cannot let the lock expire mid-flight.
  with_lock(ttl: 60) { |inputs| "order:#{inputs[:order_id]}" }

  step :refund do
    argument :order_id, input(:order_id)
    run { |inputs| PaymentGateway.refund(inputs.order_id) }
  end
end

class ChargeStep < RubyReactor::Step
  input :account_id

  # All five macros also work declared on a STEP, not just the reactor —
  # keying the critical section down to this one operation instead of the
  # whole workflow. Surrounding steps (audit, notify, ...) keep overlapping
  # across concurrent executions; only :charge serializes. Its undo re-takes
  # the key, waiting up to `rollback_wait:` (default: `ttl`; 60 s for a
  # step `with_semaphore`) — an undo that still can't run is reported on
  # `Failure#rollback_failures`, never dropped.
  with_lock(ttl: 60, rollback_wait: nil) { |args| "acct:#{args[:account_id]}" }

  def run
    Success(charge!(inputs))
  end
end

class GeocodeReactor < RubyReactor::Reactor
  input :address

  # At most 5 geocode calls in flight across the fleet.
  with_semaphore(limit: 5) { |inputs| "geocode_api" }

  step :geocode do
    argument :address, input(:address)
    run { |inputs| Geocoder.lookup(inputs.address) }
  end
end

class MonthlyBillingReactor < RubyReactor::Reactor
  input :org_id

  # Run at most once per UTC month per org. Subsequent calls in the same month
  # return RubyReactor::Halt without executing any step. Pair with
  # with_lock for strict at-most-one even under concurrent racers.
  with_period(every: :month) { |inputs| "monthly_billing:#{inputs[:org_id]}" }

  step :build do
    argument :org_id, input(:org_id)
    run { |inputs| Billing.generate(inputs.org_id) }
  end
end

class ChargeReactor < RubyReactor::Reactor
  input :account_id

  # Respect upstream Stripe rate limits: 3/sec and 100/min.
  # Background workers snooze for exactly retry_after seconds instead of
  # consuming the backend's retry budget.
  with_rate_limit(
    limits: { second: 3, minute: 100 }
  ) { |inputs| "stripe:#{inputs[:account_id]}" }

  step :charge do
    argument :account_id, input(:account_id)
    run { |inputs| Stripe.charge(inputs.account_id) }
  end
end

class OrderedTransactionReactor < RubyReactor::Reactor
  background all: true
  input :account_id
  input :transaction

  # Strict order: a monotonically increasing nonce is assigned at enqueue
  # time (inside `Reactor.run`). Workers only execute when their nonce
  # equals last_completed + 1; otherwise they snooze. After the sequence
  # fully drains the counter resets to 0.
  with_ordered_lock(poison_pill_timeout: 300) { |inputs| "txs:#{inputs[:account_id]}" }

  step :apply do
    argument :transaction, input(:transaction)
    run { |inputs| Ledger.apply(inputs.transaction) }
  end
end

# Caller-side order is preserved; the worker pool may pick jobs in any order
# but the gate enforces sequential execution per key.
[tx1, tx2, tx3].each { |tx| OrderedTransactionReactor.run(account_id: 42, transaction: tx) }

Named global limits. When several reactors hit the same external service, register the limit once and reference it by name. The name is the shared key base, so every reactor throttles against one bucket:

RubyReactor.configure do |config|
  config.rate_limits.register(:stripe, limits: { second: 3, minute: 100 })
  config.rate_limits.register(:twilio, limit: 10, period: :second)
end

class ChargeReactor < RubyReactor::Reactor
  input :account_id

  with_rate_limit(:stripe)   # shared :stripe quota across every reactor

  step :charge do
    argument :account_id, input(:account_id)
    run { |inputs| Stripe.charge(inputs.account_id) }
  end
end

Referencing an unregistered name raises RubyReactor::RateLimitRegistry::UnknownLimitError. Named limits hit the same enforcement path as inline ones, so background snooze behavior is identical.

On contention:

• Inline (Reactor.run) raises RubyReactor::Lock::AcquisitionError / RubyReactor::Semaphore::AcquisitionError / RubyReactor::RateLimit::ExceededError / RubyReactor::OrderedLock::WaitError.
• Background (Sidekiq or ActiveJob) snoozes the job via perform_in(delay, ...). For rate limits the delay uses the error's retry_after_seconds hint (precise wakeup — the bucket roll time is known exactly); for locks, semaphores, and ordered-lock waits it's lock_snooze_base_delay + jitter (a short re-poll, since a held lock or a live blocker nonce typically clears in milliseconds). Snoozes do not count against the backend's retry budget. After lock_snooze_max_attempts snoozes the context is marked failed (ordered-lock waits bypass the cap — see the ordered-lock docs).

On dedup hits (period gate already marked), the reactor returns a RubyReactor::Halt result instead — no steps run, no exception:

result = MonthlyBillingReactor.run(org_id: 42)
result.success?  # true (Halt is a Success subclass)
result.halted?   # true on dedup hit, false otherwise

A step's run block can also return Halt(reason: "...") to stop the reactor cleanly — remaining steps don't execute, and already-completed steps are NOT compensated. Use it when the rest of the workflow is unnecessary and partial progress should be kept (Failure is for "stop and roll back"). Halt is a bare helper just like Success/Failure (or use the fully-qualified RubyReactor.Halt(...)).

step :ensure_active do
  argument :user, result(:fetch_user)
  run do |inputs, _ctx|
    next Halt(reason: "user_opted_out") if inputs.user.opted_out?
    Success(inputs.user)
  end
end

To mark a single step as having had nothing to do — the rest of the workflow runs as usual — return Skipped(value) instead. Skipped is only an instrumentation mark, so an engineer reviewing the execution can see the step did not need to run. It never changes how the run executes: the value flows to dependants, and a background hand-off and a with_period bucket treat it as a completed step. A skipped step had nothing to do, so it is never undone. (where/guard were removed; skipping from the body is the only way.)

step :maybe_sync do
  argument :user, result(:fetch_user)
  run do |inputs, _ctx|
    next Skipped(inputs.user) if inputs.user.already_synced?
    Success(sync!(inputs.user))
  end
end

See Locks, Semaphores, Rate Limits, Periods & Ordered Locks for re-entrancy, auto-extend, multi-window quotas, bucket semantics, owner identity, snooze tuning, ordered-lock assignment + poison-pill semantics, step-scoped coordination, and operational notes.

Process collections in parallel using the map step:

class DataProcessingReactor < RubyReactor::Reactor
  input :items

  map :process_items do
    source input(:items)
    argument :item, element(:process_items)
    
    # Run elements as background jobs, in batches
    fan_out batch_size: 50

    step :transform do
      argument :item, input(:item)
      run { |inputs| transform_item(inputs.item) }
    end

    returns :transform
  end
end

A fan_out map is a hand-off point: the reactor stops at the map (the caller gets a DispatchResult), every element runs as its own background job, and once all outcomes are collected the reactor resumes in a worker with the steps after the map. It fans out the same way when the reactor is already running in a worker (e.g. background all: true), and inside a composed child, where the top-level run is the one that resumes.

By using fan_out with batch_size, the system applies Back Pressure to efficiently manage resources. Read more about Back Pressure & Resource Management.

Rollback. A map rolls back like a composed reactor: the undos already declared on the element reactor's steps are each element's rollback, unless the map declares undo_all (below). When the map fails (an element fails in an atomic map, the default, or collect raises), and when a later step fails or the run is undone manually, every element that completed is rolled back, newest-started first, and the run ends with the same Failure in inline and fan-out mode alike. A map rolls back the way it ran: an inline map in process, reading element states 100 at a time; a fan-out map with one rollback job per started element, batch_size per throw, the same back pressure as its forward run. Until the last element reports, the run is rolling_back, and only then are the steps before the map undone. A failing atomic fan-out map lets elements already in flight finish before it reports the failure. Make element undos idempotent: one cut off by a killed worker runs again. See Rollback.

One bulk undo. When an API can roll back many elements in one call (a bulk refund), declare undo_all on the map. Every rollback of the map then calls it once with the completed elements' results, instead of replaying each element's undos:

map :charges, ChargeReactor do
  source input(:payments)
  argument :payment, element(:charges)
  fan_out batch_size: 100

  # A lazy Enumerable of the completed elements' results, in index order.
  undo_all do |charges|
    Payments.bulk_refund(charges.map { |charge| charge[:id] }.to_a)
  end
end

A fan-out map dispatches no element rollback jobs for it, and nothing holds every result at once. Failed elements still roll themselves back; a block that raises or returns a Failure is reported as one rollback failure (kind: :undo_all), and the steps before the map are still undone. See undo_all.

batch_size is optional: with fan_out alone, no throw enqueues more than 50 element jobs (RubyReactor::Map::DEFAULT_BATCH_SIZE), forward and rollback, and the outcomes are aggregated into a ResultEnumerator. Set batch_size to change the throw size. See fan_out Without batch_size.

Breaking change: async true inside a map block has been removed — it read like async_step/async_reactor, which dispatch independent units the reactor does not stop for. It now raises at class-definition time. The exact replacement is fan_out (fan_out batch_size: N).

You can use a block for source to dynamically fetch data, such as from ActiveRecord queries. The result is wrapped in a ResultEnumerator for easy access to successes and failures.

map :archive_old_users do
  argument :days, input(:days)

  # Dynamic source using ActiveRecord
  source do |args|
    User.where("last_login_at < ?", args[:days].days.ago)
  end
  
  argument :user, element(:archive_old_users)
  fan_out batch_size: 100

  step :archive do
    argument :user, input(:user)
    run { |inputs| inputs.user.archive! }
  end
  
  returns :archive
end

step :summary do
  argument :results, result(:archive_old_users)
  run do |inputs|
    puts "Archived: #{inputs.results.successes.count}"
    puts "Failed: #{inputs.results.failures.count}"
    Success()
  end
end

RubyReactor integrates with dry-validation for input validation. A single input method escalates from a bare declaration to a full nested schema. Name and optionality are declared once.

Form 0 — declaration only (no validation, value passes through as-is):

input :user
input :payload, redact: true

Form 1 — inline scalar (the common case). The type is an optional positional; any keyword ending in ? is a dry-schema predicate:

input :name,  :string,  min_size?: 2
input :email, :string,  format?: /\A[^@\s]+@[^@\s]+\z/
input :age,   :integer, gteq?: 18

# optional: true flips required(...).filled -> optional(...).maybe
input :bio,   :string,  optional: true, max_size?: 100

Form 1b — class / module maps to a type? instance check:

input :user,  User                 # required(:user).filled(type?: User)
input :items, Array, min_size?: 1  # instance check + predicate

Form 2 — macro block for nested / complex schemas. The block receives the input's macro (already bound to the name and optionality), and because it is a block argument your class constants and helpers stay reachable:

EMAIL = /\A[^@\s]+@[^@\s]+\z/

input :order do |i|
  i.hash do
    required(:customer).hash do
      required(:email).filled(:string, format?: EMAIL)
    end
    required(:items).each do
      schema { required(:product_id).filled(:string) }
    end
  end
end

Form 3 — pre-built schema / contract (also the home for coercing dry-types):

UserSchema = Dry::Schema.Params do
  required(:user).hash { required(:email).filled(:string) }
end

input :user, validate: UserSchema

Putting the inline forms together:

class ValidatedUserReactor < RubyReactor::Reactor
  input :name,  :string,  min_size?: 2
  input :email, :string
  input :age,   :integer, gteq?: 18
  input :bio,   :string,  optional: true, max_size?: 100

  step :create_profile do
    argument :name, input(:name)
    argument :email, input(:email)
    argument :age, input(:age)
    argument :bio, input(:bio)

    run do |inputs, context|
      profile = {
        name: inputs.name,
        email: inputs.email,
        age: inputs.age,
        bio: inputs.bio || "No bio provided",
        created_at: Time.now
      }
      Success(profile)
    end
  end

  returns :create_profile
end

# Valid input
result = ValidatedUserReactor.run(
  name: "Alice Johnson",
  email: "alice@example.com",
  age: 25,
  bio: "Software developer"
)

# Invalid input - will return validation errors
result = ValidatedUserReactor.run(
  name: "A",  # Too short
  email: "",  # Empty
  age: 15     # Too young
)

All validation failures — inputs, step arguments, step output, and interrupt payloads — surface uniformly as a RubyReactor::Error::InputValidationError with a field_errors hash:

result = ValidatedUserReactor.run(name: "A")
result.error                       # => RubyReactor::Error::InputValidationError
result.error.field_errors[:name]   # => "size cannot be less than 2"

A step declares the inputs it accepts; the reactor only says where each value comes from. Rules live with the unit of work, so a step reused by three reactors is validated the same way in all three.

class ChargeStep < RubyReactor::Step
  input :amount,   :integer, gteq?: 1
  input :currency, :string,  included_in?: %w[USD EUR GBP]
  input :user,     User                                   # type? instance check
  input :note,     :string,  optional: true, default: "", max_size?: 100
  input :token,    :string,  redact: true                 # "[REDACTED]" in failures and traces

  # Cross-field rules over the whole argument hash, applied last
  validate_inputs do
    required(:amount).filled(:integer, lt?: 10_000)
  end

  def run
    Success(charge!(inputs))
  end
end

input takes the same forms as a reactor input (inline type and predicates, do |i| ... end macro block, validate: Schema). The contract is enforced before run on every path: inline execution, each retry attempt, async_step and background workers, resume, every map element, and a direct ChargeStep.run(args, context) call. A violation raises RubyReactor::Error::InputValidationError; inside a reactor that becomes a Failure with validation_errors and the step's name, after completed steps are rolled back, and have_validation_error(:amount) matches it.

Wiring. argument only maps values:

class ChargeReactor < RubyReactor::Reactor
  input :amount
  input :currency
  input :user

  step :charge, ChargeStep do
    argument :user, result(:load_user)   # explicit wiring always wins
  end                                    # :amount and :currency resolve by name
end

A declared input with no argument resolves from the reactor input of the same name. Step results are never used for this. A required input that is neither wired nor a reactor input raises RubyReactor::Error::ValidationError before any step runs, naming the reactor, the step, the input, and both fixes. ChargeReactor.validate_definition! runs the same check without running the reactor, e.g. in an initializer or a CI task.

For a step that owns a contract, these raise RubyReactor::Error::ValidationError when the step line is evaluated:

• a type or predicate on argument (argument :amount, input(:amount), :integer)
• validate_args
• an argument naming an input the step does not declare

There are no per-reactor overrides. If two reactors need different bounds for the same value, write two steps (or a parameterized step), or relax the contract.

Inline steps declare the same lines inside inputs do ... end. Inside the block input declares; outside it, input(:x) is still the template reference:

step :charge do
  inputs do
    input :amount,   :integer, gteq?: 1
    input :currency, :string,  included_in?: %w[USD EUR GBP]
  end

  argument :amount,   input(:amount)
  argument :currency, input(:currency)

  run { |inputs, _| charge!(inputs) }
end

Moving an inline step into a class is deleting the inputs do wrapper.

Presence. A value is provided when its key exists, not when it is truthy:

Migrating from rules on argument. Rules on argument and validate_args still work for steps with no contract, and print a one-time deprecation notice per declaration site. Removal is no earlier than the next major version.

# Before
step :charge, ChargeStep do
  argument :amount, input(:amount), :integer, gteq?: 1
  validate_args { required(:amount).filled(:integer, lt?: 10_000) }
end

# After: the rules move into the step
class ChargeStep < RubyReactor::Step
  input :amount, :integer, gteq?: 1
  validate_inputs { required(:amount).filled(:integer, lt?: 10_000) }
end

step :charge, ChargeStep   # or keep `argument :amount, input(:amount)` for wiring

For an inline step, put the same input / validate_inputs lines inside inputs do ... end.

Output validation is scalar-aware — pass a type/predicates for a single value, or a block for a hash output:

validate_output :integer, gteq?: 0   # scalar return value
validate_output do                   # hash return value
  required(:id).filled(:string)
end

Validate the payload supplied on resume with validate_payload (the older validate is kept as a deprecated alias):

interrupt :await_approval do
  wait_for :submit_request

  validate_payload do
    required(:approved).filled(:bool)
    optional(:note).maybe(:string)
  end
end

Steps can depend on results from multiple other steps:

class OrderProcessingReactor < RubyReactor::Reactor
  input :user_id
  input :product_ids, Array, min_size?: 1

  step :validate_user do
    argument :user_id, input(:user_id)

    run do |inputs, context|
      # Check if user exists and has permission to purchase
      user = find_user(inputs.user_id)
      user ? Success(user) : Failure("User not found")
    end
  end

  step :validate_products do
    argument :product_ids, input(:product_ids)

    run do |inputs, context|
      products = inputs.product_ids.map { |id| find_product(id) }
      if products.all?
        Success(products)
      else
        Failure("Some products not found")
      end
    end
  end

  step :calculate_total do
    argument :products, result(:validate_products)

    run do |inputs, context|
      total = inputs.products.sum { |p| p[:price] }
      Success(total)
    end
  end

  step :check_inventory do
    argument :products, result(:validate_products)

    run do |inputs, context|
      available = inputs.products.all? { |p| p[:stock] > 0 }
      available ? Success(true) : Failure("Out of stock")
    end
  end

  step :process_payment do
    argument :user, result(:validate_user)
    argument :total, result(:calculate_total)

    run do |inputs, context|
      # Process payment logic here
      payment_id = process_payment(inputs.user[:id], inputs.total)
      Success(payment_id)
    end

    undo do |error, inputs, context|
      # Refund payment on failure
      refund_payment(inputs.payment_id)
      Success()
    end
  end

  step :create_order do
    argument :user, result(:validate_user)
    argument :products, result(:validate_products)
    argument :payment_id, result(:process_payment)

    run do |inputs, context|
      order = create_order_record(inputs.user, inputs.products, inputs.payment_id)
      Success(order)
    end

    undo do |error, inputs, context|
      # Cancel order and update inventory
      cancel_order(inputs.order[:id])
      Success()
    end
  end

  step :update_inventory do
    argument :products, result(:validate_products)

    run do |inputs, context|
      inputs.products.each { |p| decrement_stock(p[:id]) }
      Success(true)
    end

    undo do |error, inputs, context|
      # Restock products
      inputs.products.each { |p| increment_stock(p[:id]) }
      Success()
    end
  end

  step :send_confirmation do
    argument :user, result(:validate_user)
    argument :order, result(:create_order)

    run do |inputs, context|
      send_email(inputs.user[:email], "Order confirmed", order_details(inputs.order))
      Success(true)
    end
  end

  returns :send_confirmation
end

When a step fails, RubyReactor automatically undoes completed steps in reverse order, compensate only runs in the failing step and backwalks the executed steps undo blocks. Composed reactors and maps are completed steps too: undoing one replays its child's (or each completed element's) own step undos:

class TransactionReactor < RubyReactor::Reactor
  input :from_account
  input :to_account
  input :amount

  step :validate_accounts do
    argument :from_account, input(:from_account)
    argument :to_account, input(:to_account)

    run do |inputs, context|
      from = find_account(inputs.from_account)
      to = find_account(inputs.to_account)

      if from && to && from != to
        Success({from: from, to: to})
      else
        Failure("Invalid accounts")
      end
    end
  end

  step :check_balance do
    argument :accounts, result(:validate_accounts)
    argument :amount, input(:amount)

    run do |inputs, context|
      if inputs.accounts[:from][:balance] >= inputs.amount
        Success(inputs.accounts)
      else
        Failure("Insufficient funds")
      end
    end
  end

  step :debit_account do
    argument :accounts, result(:check_balance)
    argument :amount, input(:amount)

    run do |inputs, context|
      debit(inputs.accounts[:from][:id], inputs.amount)
      Success(inputs.accounts)
    end

    undo do |error, inputs, context|
      # Credit the amount back
      credit(inputs.accounts[:from][:id], inputs.amount)
      Success()
    end
  end

  step :credit_account do
    argument :accounts, result(:debit_account)
    argument :amount, input(:amount)

    run do |inputs, context|
      credit(inputs.accounts[:to][:id], inputs.amount)
      Success({transaction_id: generate_transaction_id()})
    end

    undo do |error, inputs, context|
      # Debit the amount back from recipient
      debit(inputs.accounts[:to][:id], inputs.amount)
      Success()
    end
  end

  step :notify do
    argument :accounts, result(:validate_accounts)
    wait_for :credit_account, :debit_account

    run do |inputs, context|
      Notify.to(inputs.accounts[:from])
      Notify.to(inputs.accounts[:to])
    end
     
  end

  returns :credit_account
end

# If credit_account fails, RubyReactor will:
# 1. Compensate credit_account (debit the recipient)
# 2. Undo debit_account (credit the sender)
# Result: Complete rollback of the transaction

A rollback that did not complete is never silent. Failure#rollback_failures lists every undo or compensation that raised, returned a Failure, or could not re-take its step's lock/semaphore within rollback_wait: — including those of composed children (flattened). It is always an Array, empty when rollback completed, and is part of Failure#to_h:

result = TransactionReactor.run(from_account: 1, to_account: 2, amount: 10)
result.rollback_failures
# => [{ step: :debit_account, kind: :undo, key: nil, reason: :raised, message: "gateway timeout" }]

reason is :coordination_unavailable, :returned_failure, or :raised; kind is :undo or :compensate. An entry from a map element's rollback also carries map_step: and element_index:, and a map reports an element it could not roll back with reason: :context_unavailable (its stored context, or the map's element index, expired) or :element_in_flight (a duplicate of it was still running). See Step Rollback.

Every error after a completed step rolls back, not only a step body's failure. An argument source, transform or result path that raises fails that step with RubyReactor::Error::ArgumentResolutionError. The step is not compensated (its body never started) nor retried; completed steps are undone. The Failure names the step, and exception_class reports the original error's class:

result.step_name        # => :charge
result.exception_class  # => "ArgumentError"

That holds for every exception, not only a StandardError: a step class that never implemented run (NotImplementedError) or a custom Exception subclass fails its step and rolls back the same way, and exception_class names it.

Only an interruption (a signal such as Interrupt, SystemExit, NoMemoryError, or an enclosing Timeout.timeout) reaches the caller unchanged and runs no rollback code. A run in the caller's process is stored with status aborted, the steps not yet undone still outstanding; MyReactor.undo(id) rolls them back. A run in a worker is redelivered instead.

MyReactor.undo(id) holds the run's context lock while it undoes, and raises RubyReactor::Lock::AcquisitionError if a live run or rollback holds it (retry later). When the undo reaches a completed fan-out map, it undoes the steps after the map, dispatches the map's element rollbacks and returns with the run rolling_back; a worker finishes the rest and marks the run cancelled. While a run is rolling_back, both undo and cancel raise RubyReactor::Error::ValidationError: cancelling would mark it finished and strand the steps before the map.

You can use existing dry-validation schemas:

require 'dry/schema'

user_schema = Dry::Schema.Params do
  required(:user).hash do
    required(:name).filled(:string, min_size?: 2)
    required(:email).filled(:string)
    optional(:phone).maybe(:string)
  end
end

class SchemaValidatedReactor < RubyReactor::Reactor
  input :user, validate: user_schema

  step :process_user do
    argument :user, input(:user)

    run do |inputs, context|
      Success(inputs.user)
    end
  end

  returns :process_user
end

RubyReactor provides testing utilities for RSpec. See the Testing with RSpec guide for comprehensive documentation — including unit-testing class-based steps directly.

RSpec.describe PaymentReactor, type: :reactor do
  it "processes payment successfully" do
    subject = test_reactor(PaymentReactor, order_id: 123, amount: 99.99)
    
    expect(subject).to be_success
    expect(subject).to have_run_step(:charge_card).after(:validate_order)
    expect(subject.step_result(:charge_card)).to include(status: "completed")
  end

  it "handles payment failures with mocked steps" do
    subject = test_reactor(PaymentReactor, order_id: 123, amount: 99.99)
      .mock_step(:charge_card) { Failure("Card declined") }
    
    expect(subject).to be_failure
    expect(subject.error).to include("Card declined")
  end
end

For detailed documentation, see the following guides:

Learn about the fundamental building blocks of RubyReactor: Reactors, Steps, Context, and Results. Covers class-based steps (the preferred approach) and inline blocks, how data flows between steps, and how the context maintains state throughout execution.

Deep dive into how RubyReactor manages dependencies. This guide explains how the Directed Acyclic Graph is constructed to ensure steps execute in the correct topological order, enabling automatic parallelization of independent steps.

Explore the ways to move work off the calling process: background execution (background all: true and the background after:/before: hand-off) and async units (async_step, async_reactor). Learn how RubyReactor leverages Sidekiq or ActiveJob for background processing, non-blocking execution, and scalable worker management.

Discover how to build complex, modular workflows by composing reactors within other reactors. This guide covers inline composition, class-based composition, and how to manage dependencies between composed workflows.

Master the map feature for processing collections. Learn about parallel execution, batch processing for large datasets, and error handling strategies like atomic maps vs. partial result collection.

Configure robust retry policies for your steps. This guide details the available backoff strategies (exponential, linear, fixed), how to declare retries on a step class or inline step, and how background retries work without blocking workers.

Learn how to pause and resume reactors to handle long-running processes, manual approvals, and asynchronous callbacks. Patterns for correlation IDs, timeouts, and payload validation.

Comprehensive guide to testing reactors with RubyReactor's testing utilities. Learn about the TestSubject class for reactor execution and introspection, step mocking for isolating dependencies, testing nested and composed reactors, and custom RSpec matchers like be_success, have_run_step, and have_retried_step.

Coordinate access to shared resources across processes with Redis-backed primitives: exclusive locks (with_lock), concurrency-limiting semaphores (with_semaphore), fixed-window rate limits with multi-window quotas (with_rate_limit), calendar-bucketed dedup (with_period, returning Halt results), and strict sequential ordering via a monotonically increasing nonce assigned at enqueue (with_ordered_lock). Every primitive is also available declared on a step instead of the whole reactor, keying the critical section down to one operation. Covers re-entrancy across composed reactors, TTL auto-extend, inline-vs-async contention behavior, smart retry_after snoozes for rate limits, snooze tuning, the token-based semaphore safety model, once-per-day/month/year scheduling patterns, ordered-lock counter reset on drain, poison-pill timeouts, and deadlock-safe composition rules.

Hook into the execution lifecycle with observer middlewares. Covers the full set of lifecycle events (reactor, step, retry, compensation/undo, background hand-off, locks/semaphores), writing and registering custom middlewares (global and per-reactor), and the built-in RubyReactor::OpenTelemetry tracing middleware — span structure, input/argument redaction, distributed trace propagation across background/retry boundaries, and custom exporters.

• Order Processing - Complete order processing workflow example
• Payment Processing - Payment handling with compensation
• Inventory Management - Inventory management system example

• Global id to serialize ActiveRecord classes
• Descriptive errors
• map step to iterate over arrays in parallel
• compose special step to execute reactors as step
• interrupt to pause and resume reactors
• Middlewares
• Async ruby to parallelize same level steps
• Web dashboard to inspect reactor results and errors
• Multiple storage adapters
  • Redis
  • ActiveRecord
• Multiple background job adapters
  • Sidekiq
  • ActiveJob
• OpenTelemetry support
• locks

After checking out the repo, run bin/setup to install dependencies. Then, run rake spec to run the tests. You can also run bin/console for an interactive prompt that will allow you to experiment.

To install this gem onto your local machine, run bundle exec rake install. To release a new version, update the version number in version.rb, and then run bundle exec rake release, which will create a git tag for the version, push git commits and the created tag, and push the .gem file to rubygems.org.

The gem's RSpec suite expects Redis on port 6780 (see spec/spec_helper.rb). Start it via Docker Compose:

docker compose up -d redis-test

Then run the suite:

bundle exec rspec

Stop it when done:

docker compose stop redis-test

The demo Rails app under demo_app/ has its own Redis (port 6380) and bind-mounts the repo so edits to lib/ are live. Two ways to run it:

Option A — fully containerized (Redis + Rails + Sidekiq):

docker compose up demo-redis demo-app demo-sidekiq

App available at http://localhost:3789.

Option B — Redis in Docker, Rails on host:

docker compose up -d demo-redis

cd demo_app
bin/rails db:prepare
REDIS_URL=redis://localhost:6380/1 bin/rails server
# in another shell, if you need Sidekiq:
REDIS_URL=redis://localhost:6380/1 bundle exec sidekiq

To run the demo app specs:

cd demo_app
bundle exec rspec

Tear everything down:

docker compose down          # stop containers
docker compose down -v       # also remove demo_redis_data + bundle_cache volumes

Bug reports and pull requests are welcome on GitHub at https://github.com/arturictus/ruby_reactor. This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the code of conduct.

Everyone interacting in the RubyReactor project's codebases, issue trackers, chat rooms and mailing lists is expected to follow the code of conduct.