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_router ⇒ Object 



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

def async_router
  @async_router ||= RubyReactor::Adapters::Sidekiq::Router
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).



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

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.



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

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.



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

def context_ttl
  @context_ttl ||= 86_400
end

#job_retry_count ⇒ Object 



92
93
94
 
# File 'lib/ruby_reactor/configuration.rb', line 92

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.



107
108
109
 
# File 'lib/ruby_reactor/configuration.rb', line 107

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.



112
113
114
 
# File 'lib/ruby_reactor/configuration.rb', line 112

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.



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

def lock_snooze_max_attempts
  @lock_snooze_max_attempts ||= 20
end

#logger ⇒ Object 



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

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

#middlewares ⇒ Object 



143
144
145
 
# File 'lib/ruby_reactor/configuration.rb', line 143

def middlewares
  @middlewares ||= []
end

#queue_name ⇒ Object 



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

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.



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

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.



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

def sweeper_interval
  @sweeper_interval ||= 30
end

#sweeper_limit ⇒ Object

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



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

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).



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

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.



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

def sidekiq_queue
  queue_name
end

#sidekiq_queue=(value) ⇒ Object 



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

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.



98
99
100
 
# File 'lib/ruby_reactor/configuration.rb', line 98

def sidekiq_retry_count
  job_retry_count
end

#sidekiq_retry_count=(value) ⇒ Object 



102
103
104
 
# File 'lib/ruby_reactor/configuration.rb', line 102

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

#storage ⇒ Object 



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

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

#storage_adapter ⇒ Object 



134
135
136
137
138
139
140
141
 
# File 'lib/ruby_reactor/configuration.rb', line 134

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