Class: Pgbus::Process::Supervisor

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

Constant Summary collapse

FORK_WAIT =

seconds between fork checks

1
WATCHDOG_INTERVAL =

seconds between stall checks

10
RESTART_STABLE_UPTIME =

A child that crashes within this many seconds of forking is treated as crash-looping and restarted with exponential backoff (base RESTART_BACKOFF_BASE, doubling per consecutive crash, capped at RESTART_BACKOFF_MAX). A child that ran at least this long — or that exited cleanly, like a recycling worker — restarts immediately and resets its crash streak.

30
RESTART_BACKOFF_BASE = 
1
RESTART_BACKOFF_MAX = 
60

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, forks: {}, shutting_down: false, pending_restarts: [], last_watchdog_at: nil) ⇒ Supervisor

The child-fork bookkeeping (forks, pending_restarts) and the shutting_down flag accept injected seeds so tests can drive the monitor/reap loops from a known state without poking private ivars. All default to the empty/false values production always starts from.



33
34
35
36
37
38
39
40
41
 
# File 'lib/pgbus/process/supervisor.rb', line 33

def initialize(config: Pgbus.configuration, forks: {}, shutting_down: false,
               pending_restarts: [], last_watchdog_at: nil)
  @config = config
  @forks = forks
  @shutting_down = shutting_down
  @last_watchdog_at = last_watchdog_at || monotonic_now
  @pending_restarts = pending_restarts
  @crash_counts = Hash.new(0)
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



23
24
25
 
# File 'lib/pgbus/process/supervisor.rb', line 23

def config
  @config
end

#forks ⇒ Object

forks is readable everywhere; the writer exists only so tests can seed a known set of children before exercising the reap/watchdog paths (in production forks is populated by fork_* as children spawn).



27
28
29
 
# File 'lib/pgbus/process/supervisor.rb', line 27

def forks
  @forks
end

#last_watchdog_at=(value) ⇒ Object (writeonly)

Test seam: reset the watchdog clock so check_stalled_workers runs on the next call instead of waiting out WATCHDOG_INTERVAL. Production advances this internally after each watchdog pass.



50
51
52
 
# File 'lib/pgbus/process/supervisor.rb', line 50

def last_watchdog_at=(value)
  @last_watchdog_at = value
end

Instance Method Details

#graceful_shutdown ⇒ Object 



85
86
87
88
89
 
# File 'lib/pgbus/process/supervisor.rb', line 85

def graceful_shutdown
  Pgbus.logger.info { "[Pgbus] Supervisor: graceful shutdown requested" }
  @shutting_down = true
  signal_children("TERM")
end

#immediate_shutdown ⇒ Object 



91
92
93
94
95
 
# File 'lib/pgbus/process/supervisor.rb', line 91

def immediate_shutdown
  Pgbus.logger.warn { "[Pgbus] Supervisor: immediate shutdown requested" }
  @shutting_down = true
  signal_children("QUIT")
end

#run ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/pgbus/process/supervisor.rb', line 52

def run
  setup_signals
  start_heartbeat
  start_health_server

  Pgbus.logger.info { "[Pgbus] Supervisor starting pid=#{::Process.pid}" }

  # Emit a one-block boot diagnostics banner so operators can read the
  # resolved deployment config (connection target, pool, notify flags,
  # roles, capsules) straight from the log instead of attaching a console.
  log_boot_banner

  # Fail fast on a bad database_url / connection_params. PGMQ's pool is
  # lazy, so an unreachable DB would otherwise only surface once forked
  # children crash-loop against it. verify_connection! raises
  # Pgbus::ConfigurationError with an actionable message; we let it
  # propagate so the supervisor exits instead of forking anything.
  Pgbus.client.verify_connection!

  # Bootstrap queues once in the parent process before forking children.
  # This avoids the deadlock that occurs when multiple forked children
  # race to call enable_notify_insert (DROP TRIGGER + CREATE TRIGGER)
  # concurrently on the same queue tables. Children still call
  # bootstrap_queues! post-fork but the idempotent check in
  # notify_trigger_current? makes those calls cheap no-ops.
  bootstrap_queues

  boot_processes
  monitor_loop
ensure
  shutdown
end

#shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


43
44
45
 
# File 'lib/pgbus/process/supervisor.rb', line 43

def shutting_down?
  @shutting_down
end