@@ -4,7 +4,10 @@ module SolidObjects
44 class Supervisor
55 # @rbs @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
66 # @rbs @threads: Array[Thread]
7+ # @rbs @monitor: Thread?
78 # @rbs @started: bool
9+ # @rbs @cleaned_up_at: Float
10+ # @rbs @lifecycle: Thread::Mutex
811
912 # @rbs (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void
1013 def initialize (
@@ -20,7 +23,10 @@ def initialize(
2023 reminder_scheduler_count :
2124 )
2225 @threads = [ ]
26+ @monitor = nil
2327 @started = false
28+ @cleaned_up_at = nil
29+ @lifecycle = Thread ::Mutex . new
2430 end
2531
2632 # @rbs () -> void
@@ -36,7 +42,8 @@ def start
3642 return if @started
3743
3844 @started = true
39- @threads = components . map { |component | Thread . new { component . run } }
45+ @threads = components . map { |component | supervise ( component ) }
46+ @monitor = Thread . new { monitor_loop }
4047 SolidObjects . instrument ( :"supervisor.started" , component_count : components . length )
4148 end
4249
@@ -45,14 +52,19 @@ def stop
4552 return unless @started
4653
4754 begin
55+ # Flipping the flag under the same lock replacement takes means a
56+ # replacement either completes before shutdown reads the component
57+ # list, or never starts.
58+ @lifecycle . synchronize { @started = false }
59+ stop_monitor
4860 components . each ( &:request_shutdown )
4961 join_until_timeout
5062 components . reject ( &:stopped? ) . each ( &:stop )
5163 ensure
5264 # Connections held outside the pool must be released even when a
5365 # component fails to stop, or they accumulate across restarts.
5466 release_wake_up
55- @started = false
67+ @monitor = nil
5668 SolidObjects . instrument ( :"supervisor.stopped" , component_count : components . length )
5769 end
5870 end
@@ -61,6 +73,95 @@ def stop
6173
6274 attr_reader :components , :threads
6375
76+ # A role that raises leaves its thread dead. Without replacement the
77+ # process keeps running while quietly doing less work, so the supervisor
78+ # watches its threads and restarts any that stopped before shutdown.
79+ # A failing pass must not stop supervision, and must not retry without
80+ # pacing either: a persistently failing database would otherwise spin.
81+ # @rbs () -> void
82+ def monitor_loop
83+ while @started
84+ begin
85+ replace_dead_roles
86+ cleanup_dead_processes
87+ rescue => error
88+ SolidObjects . instrument (
89+ :"supervisor.monitor_failed" ,
90+ error_class : error . class . name ,
91+ error_message : error . message
92+ )
93+ end
94+ sleep SolidObjects . configuration . supervisor_monitor_interval
95+ end
96+ end
97+
98+ # A role that raises runs its own shutdown cleanup on the way out, so a
99+ # crashed component reports itself stopped exactly like one that was asked
100+ # to stop. While the supervisor is still running, a dead thread can only
101+ # mean a crash, so replacement keys on the supervisor rather than on the
102+ # component. The crashed instance has already released its process record,
103+ # so a fresh one takes its place.
104+ # @rbs () -> void
105+ def replace_dead_roles
106+ components . each_with_index do |component , index |
107+ thread = threads [ index ]
108+ next if thread &.alive?
109+
110+ replaced = @lifecycle . synchronize do
111+ next false unless @started
112+
113+ replacement = component . class . new
114+ components [ index ] = replacement
115+ threads [ index ] = supervise ( replacement )
116+ replacement
117+ end
118+ break unless replaced
119+
120+ SolidObjects . instrument (
121+ :"supervisor.role_replaced" ,
122+ role : replaced . class . name ,
123+ error_class : thread_error ( thread )
124+ )
125+ end
126+ end
127+
128+ # @rbs (Thread?) -> String?
129+ def thread_error ( thread )
130+ thread &.join
131+ nil
132+ rescue => error
133+ error . class . name
134+ end
135+
136+ # @rbs () -> void
137+ def cleanup_dead_processes
138+ interval = SolidObjects . configuration . dead_process_cleanup_interval
139+ return unless interval . positive?
140+ return if @cleaned_up_at && monotonic_now - @cleaned_up_at < interval
141+
142+ @cleaned_up_at = monotonic_now
143+ ProcessRegistry . cleanup_dead
144+ end
145+
146+ # @rbs (untyped) -> Thread
147+ def supervise ( component )
148+ Thread . new { component . run }
149+ end
150+
151+ # The monitor only performs maintenance, so shutdown must never return while
152+ # it is still alive: a pass blocked on the database would otherwise outlive
153+ # the supervisor that owns it.
154+ # @rbs () -> void
155+ def stop_monitor
156+ monitor = @monitor
157+ @monitor = nil
158+ return unless monitor
159+
160+ monitor . join ( SolidObjects . configuration . shutdown_timeout )
161+ monitor . kill if monitor . alive?
162+ monitor . join ( SolidObjects . configuration . supervisor_monitor_interval )
163+ end
164+
64165 # A wake-up adapter may hold connections outside the pool, which would
65166 # otherwise accumulate across restarts in one process.
66167 # @rbs () -> void
0 commit comments