Class: SolidObjects::Supervisor

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_objects/supervisor.rb,
sig/generated/lib/solid_objects/supervisor.rbs

Constant Summary collapse

MAXIMUM_RETENTION_BACKOFF_DOUBLINGS =

Returns:

  • (::Integer) 
16

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_count: SolidObjects.configuration.worker_count, effect_worker_count: SolidObjects.configuration.effect_worker_count, broadcast_worker_count: SolidObjects.configuration.broadcast_worker_count, reminder_scheduler_count: SolidObjects.configuration.reminder_scheduler_count) ⇒ Supervisor

Returns a new instance of Supervisor.

RBS:

  • (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void

Parameters:

  • worker_count: (Integer) (defaults to: SolidObjects.configuration.worker_count)
  • effect_worker_count: (Integer) (defaults to: SolidObjects.configuration.effect_worker_count)
  • broadcast_worker_count: (Integer) (defaults to: SolidObjects.configuration.broadcast_worker_count)
  • reminder_scheduler_count: (Integer) (defaults to: SolidObjects.configuration.reminder_scheduler_count) 


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/solid_objects/supervisor.rb', line 16

def initialize(
  worker_count: SolidObjects.configuration.worker_count,
  effect_worker_count: SolidObjects.configuration.effect_worker_count,
  broadcast_worker_count: SolidObjects.configuration.broadcast_worker_count,
  reminder_scheduler_count: SolidObjects.configuration.reminder_scheduler_count
)
  @builders = component_builders(
    worker_count:,
    effect_worker_count:,
    broadcast_worker_count:,
    reminder_scheduler_count:
  )
  @components = build_all(@builders)
  @threads = []
  @monitor = nil
  @started = false
  @cleaned_up_at = nil
  @retention = nil
  @lifecycle = Thread::Mutex.new
end

Instance Attribute Details

#builders ⇒ Object (readonly)

Returns the value of attribute builders.

Returns:

  • (Object) 


81
82
83
 
# File 'lib/solid_objects/supervisor.rb', line 81

def builders
  @builders
end

#components ⇒ Object (readonly)

Returns the value of attribute components.

Returns:

  • (Object) 


81
82
83
 
# File 'lib/solid_objects/supervisor.rb', line 81

def components
  @components
end

#threads ⇒ Object (readonly)

Returns the value of attribute threads.

Returns:

  • (Object) 


81
82
83
 
# File 'lib/solid_objects/supervisor.rb', line 81

def threads
  @threads
end

Instance Method Details

#build_all(builders) ⇒ Array[untyped]

A constructor can take a resource, and a later builder can raise. Without this, the components built first would be dropped while still holding whatever they took, and nothing would ever give it back.

RBS:

  • (Array[^() -> untyped]) -> Array[untyped]

Parameters:

  • (Array[^() -> untyped])

Returns:

  • (Array[untyped]) 


262
263
264
265
266
267
268
269
270
271
272
273
274
 
# File 'lib/solid_objects/supervisor.rb', line 262

def build_all(builders)
  built = []
  builders.each do |builder|
    # The component joins the list before the contract check, so a
    # component that fails the check is stopped along with the rest.
    built << (component = builder.call)
    SolidObjects.configuration.validate_component!(component)
  end
  built
rescue Exception # rubocop:disable Lint/RescueException
  built.each { |component| stop_after_failed_build(component) }
  raise
end

#cleanup_dead_processes ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



218
219
220
221
222
223
224
225
 
# File 'lib/solid_objects/supervisor.rb', line 218

def cleanup_dead_processes
  interval = SolidObjects.configuration.dead_process_cleanup_interval
  return unless interval.positive?
  return if @cleaned_up_at && monotonic_now - @cleaned_up_at < interval

  @cleaned_up_at = monotonic_now
  ProcessRegistry.cleanup_dead
end

#component_builders(worker_count:, effect_worker_count:, broadcast_worker_count:, reminder_scheduler_count:) ⇒ Array[^() -> untyped]

Each component keeps the builder that made it, so a replacement after a crash is built the same way as the original. Components registered through the configuration run beside the built in ones, under the same supervision, restart, and shutdown timeout.

RBS:

  • (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[^() -> untyped]

Parameters:

  • worker_count: (Integer)
  • effect_worker_count: (Integer)
  • broadcast_worker_count: (Integer)
  • reminder_scheduler_count: (Integer)

Returns:

  • (Array[^() -> untyped]) 


294
295
296
297
298
299
300
301
302
303
304
305
 
# File 'lib/solid_objects/supervisor.rb', line 294

def component_builders(
  worker_count:,
  effect_worker_count:,
  broadcast_worker_count:,
  reminder_scheduler_count:
)
  Array.new(worker_count) { -> { Worker.new } } +
    Array.new(effect_worker_count) { -> { EffectExecutor.new } } +
    Array.new(broadcast_worker_count) { -> { BroadcastExecutor.new } } +
    Array.new(reminder_scheduler_count) { -> { ReminderScheduler.new } } +
    SolidObjects.configuration.additional_components
end

#join_until_timeout ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



308
309
310
311
312
313
314
315
316
 
# File 'lib/solid_objects/supervisor.rb', line 308

def join_until_timeout
  deadline = monotonic_now + SolidObjects.configuration.shutdown_timeout
  threads.each do |thread|
    remaining = deadline - monotonic_now
    break unless remaining.positive?

    thread.join(remaining)
  end
end

#monitor_loop ⇒ void

This method returns an undefined value.

A role that raises leaves its thread dead. Without replacement the process keeps running while quietly doing less work, so the supervisor watches its threads and restarts any that stopped before shutdown. A failing pass must not stop supervision, and must not retry without pacing either: a persistently failing database would otherwise spin.

RBS:

  • () -> void



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/solid_objects/supervisor.rb', line 89

def monitor_loop
  while @started
    begin
      replace_dead_roles
      cleanup_dead_processes
    rescue => error
      SolidObjects.instrument(
        :"supervisor.monitor_failed",
        error_class: error.class.name,
        error_message: error.message
      )
    end
    sleep SolidObjects.configuration.supervisor_monitor_interval
  end
end

#monotonic_now ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


319
320
321
 
# File 'lib/solid_objects/supervisor.rb', line 319

def monotonic_now
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
end

#prune_expired_records ⇒ void

This method returns an undefined value.

Every actor call writes a durable message row, so retention that is only configured and never run leaves those rows to grow without bound. The supervisor runs it rather than requiring every application to schedule its own job.

RBS:

  • () -> void



185
186
187
188
189
190
 
# File 'lib/solid_objects/supervisor.rb', line 185

def prune_expired_records
  return unless SolidObjects.configuration.retention_interval.positive?

  MessagePruner.new.prune
  ProcessPruner.new.prune
end

#release_wake_up ⇒ void

This method returns an undefined value.

A wake-up adapter may hold connections outside the pool, which would otherwise accumulate across restarts in one process.

RBS:

  • () -> void



249
250
251
252
253
254
255
256
 
# File 'lib/solid_objects/supervisor.rb', line 249

def release_wake_up
  wake_up = SolidObjects.wake_up
  return unless wake_up.respond_to?(:stop)

  wake_up.stop
rescue
  nil
end

#replace_dead_roles ⇒ void

This method returns an undefined value.

A role that raises runs its own shutdown cleanup on the way out, so a crashed component reports itself stopped exactly like one that was asked to stop. While the supervisor is still running, a dead thread can only mean a crash, so replacement keys on the supervisor rather than on the component. The crashed instance has already released its process record, so a fresh one takes its place.

RBS:

  • () -> void



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/solid_objects/supervisor.rb', line 112

def replace_dead_roles
  components.each_with_index do |component, index|
    thread = threads[index]
    next if thread&.alive?

    replaced = @lifecycle.synchronize do
      next false unless @started

      # A component built by this supervisor has a builder, which carries
      # whatever the constructor was given. A component put in place by
      # other means has none, so the class is the only thing left to go on.
      builder = builders[index] || -> { component.class.new }
      replacement = builder.call
      components[index] = replacement
      threads[index] = supervise(replacement)
      replacement
    end
    break unless replaced

    SolidObjects.instrument(
      :"supervisor.role_replaced",
      role: replaced.class.name,
      error_class: thread_error(thread)
    )
  end
end

#retention_loop ⇒ void

This method returns an undefined value.

Retention gets its own thread rather than sharing the monitor's. A large backlog or a lock wait can make a pass slow, and role replacement must not wait behind housekeeping.

RBS:

  • () -> void



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/solid_objects/supervisor.rb', line 151

def retention_loop
  failures = 0
  while @started
    begin
      prune_expired_records
      failures = 0
    rescue => error
      failures += 1
      SolidObjects.instrument(
        :"supervisor.retention_failed",
        error_class: error.class.name,
        error_message: error.message
      )
    end
    wait_for_next_retention(failures)
  end
end

#retention_pause(failures) ⇒ Float

A transient lock or connection error must not defer retention for the whole interval, so a failed pass retries at monitor cadence. The pause then doubles per consecutive failure, capped by the interval, so a database that stays down is not polled once a second forever.

RBS:

  • (Integer) -> Float

Parameters:

  • (Integer)

Returns:

  • (Float) 


197
198
199
200
201
202
203
204
205
 
# File 'lib/solid_objects/supervisor.rb', line 197

def retention_pause(failures)
  interval = SolidObjects.configuration.retention_interval
  interval = SolidObjects.configuration.supervisor_monitor_interval unless interval.positive?
  return interval if failures.zero?

  backoff = SolidObjects.configuration.supervisor_monitor_interval *
    (2**[ failures - 1, MAXIMUM_RETENTION_BACKOFF_DOUBLINGS ].min)
  [ backoff, interval ].min
end

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



38
39
40
41
42
43
 
# File 'lib/solid_objects/supervisor.rb', line 38

def run
  start
  threads.each(&:join)
ensure
  stop
end

#start ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



46
47
48
49
50
51
52
53
54
 
# File 'lib/solid_objects/supervisor.rb', line 46

def start
  return if @started

  @started = true
  @threads = components.map { |component| supervise(component) }
  @monitor = Thread.new { monitor_loop }
  @retention = Thread.new { retention_loop }
  SolidObjects.instrument(:"supervisor.started", component_count: components.length)
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/solid_objects/supervisor.rb', line 57

def stop
  return unless @started

  begin
    # Flipping the flag under the same lock replacement takes means a
    # replacement either completes before shutdown reads the component
    # list, or never starts.
    @lifecycle.synchronize { @started = false }
    stop_monitor
    stop_retention
    components.each(&:request_shutdown)
    join_until_timeout
    components.reject(&:stopped?).each(&:stop)
  ensure
    # Connections held outside the pool must be released even when a
    # component fails to stop, or they accumulate across restarts.
    release_wake_up
    @monitor = nil
    SolidObjects.instrument(:"supervisor.stopped", component_count: components.length)
  end
end

#stop_after_failed_build(component) ⇒ void

This method returns an undefined value.

The failure that stopped the build is the one worth reporting, so a failure inside the cleanup never replaces it.

RBS:

  • (untyped) -> void

Parameters:

  • (Object) 


279
280
281
282
283
284
285
286
287
 
# File 'lib/solid_objects/supervisor.rb', line 279

def stop_after_failed_build(component)
  component.stop if component.respond_to?(:stop)
rescue Exception => error # rubocop:disable Lint/RescueException
  SolidObjects.instrument(
    :"supervisor.component_cleanup_failed",
    role: component.class.name,
    error_class: error.class.name
  )
end

#stop_monitor ⇒ void

This method returns an undefined value.

The monitor only performs maintenance, so shutdown must never return while it is still alive: a pass blocked on the database would otherwise outlive the supervisor that owns it.

RBS:

  • () -> void



236
237
238
239
240
241
242
243
244
 
# File 'lib/solid_objects/supervisor.rb', line 236

def stop_monitor
  monitor = @monitor
  @monitor = nil
  return unless monitor

  monitor.join(SolidObjects.configuration.shutdown_timeout)
  monitor.kill if monitor.alive?
  monitor.join(SolidObjects.configuration.supervisor_monitor_interval)
end

#stop_retention ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



208
209
210
211
212
213
214
215
 
# File 'lib/solid_objects/supervisor.rb', line 208

def stop_retention
  retention = @retention
  @retention = nil
  return unless retention

  retention.join(SolidObjects.configuration.shutdown_timeout)
  retention.kill if retention.alive?
end

#supervise(component) ⇒ Thread

RBS:

  • (untyped) -> Thread

Parameters:

  • (Object)

Returns:

  • (Thread) 


228
229
230
 
# File 'lib/solid_objects/supervisor.rb', line 228

def supervise(component)
  Thread.new { component.run }
end

#thread_error(thread) ⇒ String?

RBS:

  • (Thread?) -> String?

Parameters:

  • (Thread, nil)

Returns:

  • (String, nil) 


140
141
142
143
144
145
 
# File 'lib/solid_objects/supervisor.rb', line 140

def thread_error(thread)
  thread&.join
  nil
rescue => error
  error.class.name
end

#wait_for_next_retention(failures) ⇒ void

This method returns an undefined value.

Sleeping the whole interval would make shutdown wait out an hour-long nap, so the pause is taken in short steps that notice a stop request.

RBS:

  • (Integer) -> void

Parameters:

  • (Integer) 


172
173
174
175
176
177
178
 
# File 'lib/solid_objects/supervisor.rb', line 172

def wait_for_next_retention(failures)
  deadline = monotonic_now + retention_pause(failures)
  step = SolidObjects.configuration.supervisor_monitor_interval
  while @started && monotonic_now < deadline
    sleep [ step, deadline - monotonic_now ].min
  end
end