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, notify_hub: nil) ⇒ Supervisor

The child-fork bookkeeping (forks, pending_restarts), the shutting_down flag, and notify_hub 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.



38
39
40
41
42
43
44
45
46
47
 
# File 'lib/pgbus/process/supervisor.rb', line 38

def initialize(config: Pgbus.configuration, forks: {}, shutting_down: false,
               pending_restarts: [], last_watchdog_at: nil, notify_hub: 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)
  @notify_hub = notify_hub
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).



31
32
33
 
# File 'lib/pgbus/process/supervisor.rb', line 31

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.



56
57
58
 
# File 'lib/pgbus/process/supervisor.rb', line 56

def last_watchdog_at=(value)
  @last_watchdog_at = value
end

#notify_hub ⇒ Object (readonly)

The host-level shared LISTEN hub (issue #381). nil under :fork scope, when notify wakeups are off entirely, or when no worker/consumer role is enabled. Readable as a test seam.



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

def notify_hub
  @notify_hub
end

Instance Method Details

#graceful_shutdown ⇒ Object 



105
106
107
108
109
 
# File 'lib/pgbus/process/supervisor.rb', line 105

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

#immediate_shutdown ⇒ Object 



111
112
113
114
115
 
# File 'lib/pgbus/process/supervisor.rb', line 111

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

#run ⇒ Object 



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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/pgbus/process/supervisor.rb', line 58

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

  # Optional in-process doctor preflight (issue #347): run the diagnostic
  # checks here — after config is loaded, the DB is verified reachable, and
  # queues are bootstrapped, but BEFORE any worker is forked — so an
  # entrypoint gets a single Rails boot instead of `pgbus doctor` + `pgbus
  # start`. :strict aborts the boot (raising, so nothing forks) on a
  # genuinely-fatal finding; :report only logs. Off by default.
  run_doctor_preflight unless config.doctor_on_boot.nil?

  # Host-level shared LISTEN (issue #381): under :supervisor scope, ONE
  # NotifyListener lives here and forks are woken over pipes — started
  # before any child forks so every fork_worker/fork_consumer can hand
  # its child a wake pipe.
  start_notify_hub

  boot_processes
  monitor_loop
ensure
  shutdown
end

#shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


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

def shutting_down?
  @shutting_down
end