Improve server and monitor robustness.

Author: ioquatix

async-service-supervisor.gemspec

@@ -24,6 +24,7 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.3"
 
+	spec.add_dependency "async", "~> 2.38"
 	spec.add_dependency "async-bus"
 	spec.add_dependency "async-service", "~> 0.15"
 	spec.add_dependency "async-utilization", "~> 0.3"

lib/async/service/supervisor/loop.rb

@@ -6,23 +6,23 @@
 require "memory/leak/cluster"
 require "set"
 
-require_relative "loop"
+require_relative "monitor"
 
 module Async
 	module Service
 		module Supervisor
 			# Monitors worker memory usage and restarts workers that exceed limits.
 			#
 			# Uses the `memory` gem to track process memory and detect leaks.
-			class MemoryMonitor
+			class MemoryMonitor < Monitor
 				# Create a new memory monitor.
 				#
 				# @parameter interval [Integer] The interval at which to check for memory leaks.
 				# @parameter total_size_limit [Integer] The total size limit of all processes, or nil for no limit.
 				# @parameter free_size_minimum [Integer] The minimum free memory threshold, or nil for no threshold.
 				# @parameter options [Hash] Options to pass to the cluster when adding processes.
 				def initialize(interval: 10, total_size_limit: nil, free_size_minimum: nil, **options)
-					@interval = interval
+					super(interval: interval)
 					@cluster = Memory::Leak::Cluster.new(total_size_limit: total_size_limit, free_size_minimum: free_size_minimum)
 
 					# We use these options when adding processes to the cluster:
@@ -85,28 +85,11 @@ def remove(supervisor_controller)
 					end
 				end
 
-				# The key used when this monitor's status is aggregated with others.
-				def self.monitor_type
-					:memory_monitor
-				end
-				
 				# Serialize memory cluster data for JSON.
 				def as_json
 					@cluster.as_json
 				end
 
-				# Serialize to JSON string.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Get status for the memory monitor.
-				#
-				# @returns [Hash] Hash with type and data keys.
-				def status
-					{type: self.class.monitor_type, data: as_json}
-				end
-				
 				# Invoked when a memory leak is detected.
 				#
 				# @parameter process_id [Integer] The process ID of the process that has a memory leak.
@@ -128,21 +111,15 @@ def memory_leak_detected(process_id, monitor)
 					true
 				end
 
-				# Run the memory monitor.
-				#
-				# @returns [Async::Task] The task that is running the memory monitor.
-				def run
-					Async do
-						Loop.run(interval: @interval) do
-							@guard.synchronize do
-								# This block must return true if the process was killed.
-								@cluster.check! do |process_id, monitor|
-									begin
-										memory_leak_detected(process_id, monitor)
-									rescue => error
-										Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
-									end
-								end
+				# Run one iteration of the memory monitor.
+				def run_once
+					@guard.synchronize do
+						# This block must return true if the process was killed.
+						@cluster.check! do |process_id, monitor|
+							begin
+								memory_leak_detected(process_id, monitor)
+							rescue => error
+								Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
 							end
 						end
 					end

lib/async/service/supervisor/memory_monitor.rb

@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/loop"
+
+module Async
+	module Service
+		module Supervisor
+			class Monitor
+				def initialize(interval: 1.0)
+					@interval = interval
+				end
+				
+				def as_json(...)
+					{}
+				end
+				
+				# Serialize to JSON string.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
+				# Get aggregated utilization status by service name.
+				#
+				# Reads utilization data from all registered workers and aggregates it
+				# by service name (from supervisor_controller.state[:name]).
+				#
+				# @returns [Hash] Hash with type and data keys.
+				def status
+					{type: self.class.name, data: as_json}
+				end
+				
+				# Run one iteration of the monitor.
+				def run_once
+					# This method can be overridden by subclasses to implement specific monitoring logic.
+				end
+				
+				# Run the utilization monitor.
+				#
+				# Periodically aggregates utilization data from all workers.
+				#
+				# @returns [Async::Task] The task that is running the utilization monitor.
+				def run(parent: Async::Task.current)
+					parent.async do
+						Loop.periodic(interval: @interval) do
+							self.run_once
+						end
+					end
+				end
+			end
+		end
+	end
+end

lib/async/service/supervisor/monitor.rb

@@ -4,7 +4,7 @@
 # Copyright, 2025-2026, by Samuel Williams.
 
 require "process/metrics"
-require_relative "loop"
+require_relative "monitor"
 
 module Async
 	module Service
@@ -15,13 +15,13 @@ module Supervisor
 			# Unlike {MemoryMonitor}, this monitor captures metrics for the entire process tree
 			# by tracking the parent process ID (ppid), which is more efficient than tracking
 			# individual processes.
-			class ProcessMonitor
+			class ProcessMonitor < Monitor
 				# Create a new process monitor.
 				#
 				# @parameter interval [Integer] The interval in seconds at which to log process metrics.
 				# @parameter ppid [Integer] The parent process ID to monitor. If nil, uses the current process to capture its children.
 				def initialize(interval: 60, ppid: nil)
-					@interval = interval
+					super(interval: interval)
 					@ppid = ppid || Process.ppid
 				end
 
@@ -67,33 +67,13 @@ def as_json
 					{ppid: @ppid, metrics: self.metrics}
 				end
 
-				# Serialize to JSON string.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Get status for the process monitor.
-				#
-				# @returns [Hash] Hash with type and data keys.
-				def status
-					{type: self.class.monitor_type, data: as_json}
-				end
-				
-				# Run the process monitor.
-				#
-				# Periodically captures and logs process metrics for the entire process tree.
-				#
-				# @returns [Async::Task] The task that is running the process monitor.
-				def run
-					Async do
-						Loop.run(interval: @interval) do
-							metrics = self.metrics
-							
-							# Log each process individually for better searchability in log platforms:
-							metrics.each do |process_id, general|
-								Console.info(self, "Process metrics captured.", general: general)
-							end
-						end
+				# Run one iteration of the process monitor.
+				def run_once
+					metrics = self.metrics
+					
+					# Log each process individually for better searchability in log platforms:
+					metrics.each do |process_id, general|
+						Console.info(self, "Process metrics captured.", general: general)
 					end
 				end
 			end

lib/async/service/supervisor/process_monitor.rb

@@ -87,30 +87,34 @@ def remove(controller)
 				# @parameter parent [Async::Task] The parent task to run under.
 				def run
 					Sync do |task|
+						barrier = Async::Barrier.new
+						
 						# Start all monitors:
 						@monitors.each do |monitor|
-							monitor.run
+							monitor.run(parent: barrier)
 						rescue => error
 							Console.error(self, "Error while starting monitor!", monitor: monitor, exception: error)
 						end
 
-						# Accept connections from workers:
-						self.accept do |connection|
-							# Create a supervisor controller for this connection:
-							supervisor_controller = SupervisorController.new(self, connection)
-							
-							# Bind supervisor controller:
-							connection.bind(:supervisor, supervisor_controller)
-							
-							# Run the connection:
-							connection.run
-						ensure
-							self.remove(supervisor_controller)
+						barrier.async do
+							# Accept connections from workers:
+							self.accept do |connection|
+								# Create a supervisor controller for this connection:
+								supervisor_controller = SupervisorController.new(self, connection)
+								
+								# Bind supervisor controller:
+								connection.bind(:supervisor, supervisor_controller)
+								
+								# Run the connection:
+								connection.run
+							ensure
+								self.remove(supervisor_controller)
+							end
 						end
 
-						task.children&.each(&:wait)
+						barrier.wait
 					ensure
-						task.stop
+						barrier&.stop
 					end
 				end
 			end

lib/async/service/supervisor/server.rb

@@ -5,7 +5,7 @@
 
 require "set"
 
-require_relative "loop"
+require_relative "monitor"
 require "async/utilization"
 
 module Async
@@ -15,7 +15,7 @@ module Supervisor
 			#
 			# Uses shared memory to efficiently collect utilization metrics from workers
 			# and aggregates them by service name for monitoring and reporting.
-			class UtilizationMonitor
+			class UtilizationMonitor < Monitor
 				# Allocates and manages shared memory segments for worker utilization data.
 				#
 				# Manages a shared memory file that workers can write utilization metrics to.
@@ -195,8 +195,8 @@ def close
 				# @parameter size [Integer] Total size of the shared memory buffer.
 				# @parameter segment_size [Integer] Size of each allocation segment (default: 512 bytes).
 				def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512)
+					super(interval: interval)
 					@path = path
-					@interval = interval
 					@segment_size = segment_size
 
 					@allocator = SegmentAllocator.new(path, size: size, segment_size: segment_size)
@@ -313,39 +313,16 @@ def as_json
 					end
 				end
 
-				# Serialize to JSON string.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Get aggregated utilization status by service name.
-				#
-				# Reads utilization data from all registered workers and aggregates it
-				# by service name (from supervisor_controller.state[:name]).
-				#
-				# @returns [Hash] Hash with type and data keys.
-				def status
-					{type: self.class.monitor_type, data: as_json}
-				end
-				
 				# Emit the utilization metrics.
 				#
 				# @parameter status [Hash] The utilization metrics.
 				def emit(metrics)
 					Console.info(self, "Utilization:", metrics: metrics)
 				end
 
-				# Run the utilization monitor.
-				#
-				# Periodically aggregates utilization data from all workers.
-				#
-				# @returns [Async::Task] The task that is running the utilization monitor.
-				def run
-					Async do
-						Loop.run(interval: @interval) do
-							self.emit(self.as_json)
-						end
-					end
+				# Run one iteration of the utilization monitor.
+				def run_once
+					self.emit(self.as_json)
 				end
 			end
 		end

lib/async/service/supervisor/utilization_monitor.rb

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Improve robustness and error handling of default monitors and server loop, ensuring that monitor failures either completely crash the server or retry appropriately, rather than leaving the server in a broken state.
+
 ## v0.14.0
 
   - Add `Worker#make_controller` as an override point for providing a custom worker controller with additional RPCs.