Class: RubyReactor::Configuration

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/ruby_reactor/configuration.rb

Overview

Configuration class for RubyReactor settings

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#async_park_timeout ⇒ Object

Total seconds a WORKER-side result(:name) wait may stay parked before failing with Error::AsyncWaitTimeoutError, measured from the unit's dispatched_at. Inside a worker a pending result does not block the thread for async_wait_timeout — after a short in-thread grace the job parks (re-enqueues itself, locks kept held) and re-checks on redelivery, so this bound can be generous where the blocking one must stay tight. :infinity removes the bound (the context TTL then remains the only backstop), mirroring lock_snooze_max_attempts.



117
118
119
 
# File 'lib/ruby_reactor/configuration.rb', line 117

def async_park_timeout
  @async_park_timeout ||= 3600
end

#async_router ⇒ Object 



155
156
157
 
# File 'lib/ruby_reactor/configuration.rb', line 155

def async_router
  @async_router ||= RubyReactor::Adapters::Sidekiq::Router
end

#async_wait_timeout ⇒ Object

Seconds a step blocks in the notified wait when it reads result(:name) for an async_step / async_reactor that has not finished yet. Never unbounded: on expiry the referencing step fails with an Error::AsyncWaitTimeoutError.

30s must comfortably exceed dispatch -> worker pickup -> completion for a small unit under a healthy queue, while staying under the request/job timeouts of typical hosts (Sidekiq's 25s shutdown grace, Puma's 60s) so a stuck wait fails loudly on our terms instead of being killed from outside.

The wait's fallback re-check interval is DERIVED from this, not configured: async_wait_timeout / 10 clamped to 1..5s (see AsyncWaiter).



105
106
107
 
# File 'lib/ruby_reactor/configuration.rb', line 105

def async_wait_timeout
  @async_wait_timeout ||= 30
end

#checkpoint_min_interval ⇒ Object

Minimum wall-clock seconds between two PER-STEP durable checkpoints within a single worker run. The save-per-step checkpoint (on_step_complete) bounds crash re-execution to one step, but re-serializes and re-writes the WHOLE root blob after every Success — O(steps × context_size) writes for a long, large reactor. This throttle coalesces the mid-run intermediate checkpoints: a checkpoint is written only if at least this many seconds have elapsed since the last one. The final terminal/handoff state is ALWAYS persisted (by the run's ensure-save and the pre-enqueue checkpoint), so throttling only affects mid-run granularity. Tradeoff: with interval > 0, a crash may re-run every step completed inside the last interval — safe only when those steps are idempotent or side-effect-free.

Default 0 -> checkpoint after EVERY step (strongest guarantee, no coalescing).



50
51
52
 
# File 'lib/ruby_reactor/configuration.rb', line 50

def checkpoint_min_interval
  @checkpoint_min_interval ||= 0
end

#context_lock_ttl ⇒ Object

TTL (seconds) for the per-context liveness lock (async:<id>). Short by design — it is a liveness signal, not retention. A live worker auto-extends it (every ttl/3 s, from a background thread); its absence is the sweeper's "worker died" signal.

SAFETY CONSTRAINT: this MUST exceed the longest a single step can run WITHOUT letting the auto-extend thread make progress. Under MRI the extender shares the GIL, so a step that holds the GIL continuously for longer than this TTL (a long CPU-bound pure-Ruby loop, a C extension that never releases the GIL, or a stop-the-world GC pause) lets the lock lapse. A lapsed lock looks "dead" to the sweeper, which may re-enqueue a duplicate that runs CONCURRENTLY with the still-live original — a double-run. I/O-bound steps release the GIL and keep the lock fresh, so the default 60s suits typical workloads; raise it if you run long synchronous CPU-bound steps.



89
90
91
 
# File 'lib/ruby_reactor/configuration.rb', line 89

def context_lock_ttl
  @context_lock_ttl ||= 60
end

#context_ttl ⇒ Object

Retention TTL (seconds) for a stored reactor context. Storage is load-bearing for resume, so this must comfortably exceed the worst-case snooze/retry window. Refreshed on every checkpoint write.



33
34
35
 
# File 'lib/ruby_reactor/configuration.rb', line 33

def context_ttl
  @context_ttl ||= 86_400
end

#job_retry_count ⇒ Object 



121
122
123
 
# File 'lib/ruby_reactor/configuration.rb', line 121

def job_retry_count
  @job_retry_count ||= 3
end

#lock_snooze_base_delay ⇒ Object

Base seconds the Sidekiq worker waits before re-checking a contended lock.



136
137
138
 
# File 'lib/ruby_reactor/configuration.rb', line 136

def lock_snooze_base_delay
  @lock_snooze_base_delay ||= 5
end

#lock_snooze_jitter ⇒ Object

Extra random seconds added to the base delay to avoid thundering herd.



141
142
143
 
# File 'lib/ruby_reactor/configuration.rb', line 141

def lock_snooze_jitter
  @lock_snooze_jitter ||= 5
end

#lock_snooze_max_attempts ⇒ Object

How many times a single job can snooze on lock contention before it is marked as failed. Set to :infinity to never escalate.



147
148
149
 
# File 'lib/ruby_reactor/configuration.rb', line 147

def lock_snooze_max_attempts
  @lock_snooze_max_attempts ||= 20
end

#logger ⇒ Object 



151
152
153
 
# File 'lib/ruby_reactor/configuration.rb', line 151

def logger
  @logger ||= Logger.new($stdout)
end

#middlewares ⇒ Object 



172
173
174
 
# File 'lib/ruby_reactor/configuration.rb', line 172

def middlewares
  @middlewares ||= []
end

#queue_name ⇒ Object 



16
17
18
 
# File 'lib/ruby_reactor/configuration.rb', line 16

def queue_name
  @queue_name ||= :default
end

#sweeper_enabled ⇒ Object

Whether the recovery sweepers run. The host kicks the self-rescheduling chain once (RubyReactor.start_sweeper!, e.g. from an initializer); each tick re-checks this flag, so flipping it to false stops the chain at the next tick. Default on: durability is inert without a running sweeper, so recovery must work out of the box.



59
60
61
62
 
# File 'lib/ruby_reactor/configuration.rb', line 59

def sweeper_enabled
  @sweeper_enabled = true if @sweeper_enabled.nil?
  @sweeper_enabled
end

#sweeper_interval ⇒ Object

Seconds between sweeps. This is the upper bound on recovery latency for a dead worker — lower it for faster recovery, raise it to cut scan load.



66
67
68
 
# File 'lib/ruby_reactor/configuration.rb', line 66

def sweeper_interval
  @sweeper_interval ||= 30
end

#sweeper_limit ⇒ Object

Max contexts/maps inspected per sweep (passed to each sweeper's run_once).



71
72
73
 
# File 'lib/ruby_reactor/configuration.rb', line 71

def sweeper_limit
  @sweeper_limit ||= 1000
end

Instance Method Details

#rate_limits ⇒ Object

Registry of named rate limits shared across reactors. Configure entries with config.rate_limits.register(:name, ...) and reference them from a reactor via with_rate_limit(:name).



179
180
181
 
# File 'lib/ruby_reactor/configuration.rb', line 179

def rate_limits
  @rate_limits ||= RubyReactor::RateLimitRegistry.new
end

#sidekiq_queue ⇒ Object

Deprecated alias for queue_name — kept so existing Sidekiq-only configs don't break.



22
23
24
 
# File 'lib/ruby_reactor/configuration.rb', line 22

def sidekiq_queue
  queue_name
end

#sidekiq_queue=(value) ⇒ Object 



26
27
28
 
# File 'lib/ruby_reactor/configuration.rb', line 26

def sidekiq_queue=(value)
  self.queue_name = value
end

#sidekiq_retry_count ⇒ Object

Deprecated alias for job_retry_count — kept so existing Sidekiq-only configs don't break.



127
128
129
 
# File 'lib/ruby_reactor/configuration.rb', line 127

def sidekiq_retry_count
  job_retry_count
end

#sidekiq_retry_count=(value) ⇒ Object 



131
132
133
 
# File 'lib/ruby_reactor/configuration.rb', line 131

def sidekiq_retry_count=(value)
  self.job_retry_count = value
end

#storage ⇒ Object 



159
160
161
 
# File 'lib/ruby_reactor/configuration.rb', line 159

def storage
  @storage ||= RubyReactor::Storage::Configuration.new
end

#storage_adapter ⇒ Object 



163
164
165
166
167
168
169
170
 
# File 'lib/ruby_reactor/configuration.rb', line 163

def storage_adapter
  @storage_adapter ||= case storage.adapter
                       when :redis
                         RubyReactor::Storage::RedisAdapter.new(url: storage.redis_url, **storage.redis_options)
                       else
                         raise "Unknown storage adapter: #{storage.adapter}"
                       end
end