Class: Pgbus::Process::Dispatcher

Inherits:
Object
  • Object
show all
Includes:
SignalHandler
Defined in:
lib/pgbus/process/dispatcher.rb

Defined Under Namespace

Classes: LoopOutcome, MaintenanceResult

Constant Summary collapse

CLEANUP_INTERVAL =

Maintenance runs on coarser intervals than the main loop

3600
REAP_INTERVAL =

Run idempotency cleanup every hour

300
CONCURRENCY_INTERVAL =

Run stale process reaping every 5 minutes

300
BATCH_CLEANUP_INTERVAL =

Run concurrency cleanup every 5 minutes

3600
RECURRING_CLEANUP_INTERVAL =

Run batch cleanup every hour

3600
ARCHIVE_COMPACTION_INTERVAL =

Run recurring execution cleanup every hour

3600
OUTBOX_CLEANUP_INTERVAL =

Run archive compaction every hour

3600
JOB_LOCK_CLEANUP_INTERVAL =

Run outbox cleanup every hour

300
STATS_CLEANUP_INTERVAL =

Run job lock cleanup every 5 minutes

3600
ORPHAN_STREAM_SWEEP_INTERVAL =

Run stats cleanup every hour

3600
TABLE_MAINTENANCE_INTERVAL =

Run orphan stream sweep every hour

Pgbus::TableMaintenance::MAINTENANCE_INTERVAL
ARCHIVE_COMPACTION_BATCH_SIZE =

Page size for archive compaction. Each cycle deletes up to this many archived rows per queue. Tuned via constant rather than configuration because the value rarely needs adjusting and a too-small value just delays cleanup, never breaks anything.

1000
MAINTENANCE_BACKOFF_BASE =

Maintenance backoff during systemic outages. When every attempted maintenance task fails for two consecutive cycles (e.g. the database is down), skip maintenance entirely for an exponentially growing window instead of retrying ~12 tasks every dispatch interval and flooding logs and error trackers. The window doubles per consecutive failed cycle and is capped. Constants (not config) per the precedent above: the values rarely need tuning and only affect noise, never correctness. Backoff exits the moment any task succeeds again.

30
MAINTENANCE_BACKOFF_MAX =

seconds; first backoff window

600
TERMINATED_CONNECTION_SIGNAL =

A pooled backend the server terminated while the dispatcher sat idle between cycles surfaces with this exact PostgreSQL text on the next query. It is expected and self-healing — release_maintenance_connections returns the dead socket to the pool and the next cycle reconnects — so a task that hits it logs a calm INFO rather than an alarming WARN. Matched case-insensitively as a substring of the exception message. Common in local development (Postgres restart, rails db:migrate terminating idle backends, foreman/overmind restarting the DB), rare in production.

"terminating connection due to administrator command"
MAINTENANCE_TIMESTAMPS =

The per-task monotonic timestamps run_maintenance consults via run_if_due. Enumerated so set_maintenance_timestamp can validate its argument.

%i[
  @last_cleanup_at @last_reap_at @last_concurrency_at @last_batch_cleanup_at
  @last_recurring_cleanup_at @last_archive_compaction_at @last_stream_archive_compaction_at
  @last_outbox_cleanup_at @last_job_lock_cleanup_at @last_stats_cleanup_at
  @last_orphan_stream_sweep_at @last_table_maintenance_at
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from SignalHandler

#handled_signals, included, #interruptible_sleep, #process_signals, #restore_signals, #setup_signals

Constructor Details

#initialize(config: Pgbus.configuration) ⇒ Dispatcher

Returns a new instance of Dispatcher.



129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/pgbus/process/dispatcher.rb', line 129

def initialize(config: Pgbus.configuration)
  @config = config
  @shutting_down = false
  # Seed every per-task timestamp from the single source of truth so the
  # list can never drift from set_maintenance_timestamp's allow-list.
  now = monotonic_now
  MAINTENANCE_TIMESTAMPS.each { |ivar| instance_variable_set(ivar, now) }
  @maintenance_failure_streak = 0
  @maintenance_backoff_until = nil
  @loop_tick_at = Concurrent::AtomicReference.new(nil)
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



94
95
96
 
# File 'lib/pgbus/process/dispatcher.rb', line 94

def config
  @config
end

#maintenance_backoff_until ⇒ Object (readonly)

Returns the value of attribute maintenance_backoff_until.



94
95
96
 
# File 'lib/pgbus/process/dispatcher.rb', line 94

def maintenance_backoff_until
  @maintenance_backoff_until
end

#maintenance_failure_streak ⇒ Object (readonly)

Returns the value of attribute maintenance_failure_streak.



94
95
96
 
# File 'lib/pgbus/process/dispatcher.rb', line 94

def maintenance_failure_streak
  @maintenance_failure_streak
end

Instance Method Details

#graceful_shutdown ⇒ Object 



164
165
166
 
# File 'lib/pgbus/process/dispatcher.rb', line 164

def graceful_shutdown
  @shutting_down = true
end

#immediate_shutdown ⇒ Object 



168
169
170
 
# File 'lib/pgbus/process/dispatcher.rb', line 168

def immediate_shutdown
  @shutting_down = true
end

#last_loop_tick ⇒ Object

The last wall-clock timestamp (Time.now.to_f) stamped by the run loop via the heartbeat's loop_tick_supplier. Wall-clock — NOT monotonic — so it stays comparable across the parent/child process boundary the supervisor watchdog reads it over. nil until the first stamp_loop_tick.



104
105
106
 
# File 'lib/pgbus/process/dispatcher.rb', line 104

def last_loop_tick
  @loop_tick_at.get
end

#maintenance_timestamp(ivar) ⇒ Object

Read one of the @last_*_at maintenance timestamps (companion to set_maintenance_timestamp) so a test can assert a timestamp did or did not advance without reaching into the ivar directly.

Raises:

  • (ArgumentError) 


123
124
125
126
127
 
# File 'lib/pgbus/process/dispatcher.rb', line 123

def maintenance_timestamp(ivar)
  raise ArgumentError, "unknown maintenance timestamp #{ivar}" unless MAINTENANCE_TIMESTAMPS.include?(ivar.to_sym)

  instance_variable_get(ivar)
end

#run ⇒ Object 



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/pgbus/process/dispatcher.rb', line 141

def run
  setup_signals
  start_heartbeat
  Pgbus.logger.info do
    "[Pgbus] Dispatcher started: interval=#{config.dispatch_interval}s"
  end

  loop do
    stamp_loop_tick
    break if @shutting_down

    process_signals
    break if @shutting_down

    run_maintenance
    break if @shutting_down

    interruptible_sleep(config.dispatch_interval)
  end

  shutdown
end

#set_maintenance_timestamp(ivar, monotonic_value) ⇒ Object

Test seam: set one of the @last_at maintenance timestamps so a task becomes (or stops being) due on the next run_maintenance. The timestamps are per-task monotonic clocks with no constructor injection point (all 12 default to monotonic_now at construction), so a post-construction setter is the minimal way to drive the due/not-due logic deterministically. ivar must be one of the recognized @last_at names.

Raises:

  • (ArgumentError) 


114
115
116
117
118
 
# File 'lib/pgbus/process/dispatcher.rb', line 114

def set_maintenance_timestamp(ivar, monotonic_value)
  raise ArgumentError, "unknown maintenance timestamp #{ivar}" unless MAINTENANCE_TIMESTAMPS.include?(ivar.to_sym)

  instance_variable_set(ivar, monotonic_value)
end

#shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


96
97
98
 
# File 'lib/pgbus/process/dispatcher.rb', line 96

def shutting_down?
  @shutting_down
end