Introduce robust loop helper.

Author: samuel-williams-shopify

fixtures/async/container/supervisor/a_monitor.rb

@@ -29,6 +29,16 @@ module Supervisor
 						type: be == :remove,
 					)
 				end
+				
+				it "can respond to status calls" do
+					worker = Worker.new(endpoint: endpoint)
+					connection = worker.connect
+					
+					response = connection.call(do: :status)
+					
+					# Maybe we could be a bit more specific.
+					expect(response).to be_a(Array)
+				end
 			end
 		end
 	end

lib/async/container/supervisor/loop.rb

@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+module Async
+	module Container
+		module Supervisor
+			# A helper for running loops at aligned intervals.
+			module Loop
+				# A robust loop that executes a block at aligned intervals.
+				#
+				# The alignment is modulo the current clock in seconds.
+				#
+				# If an error occurs during the execution of the block, it is logged and the loop continues.
+				#
+				# @parameter interval [Integer] The interval in seconds between executions of the block.
+				def self.run(interval: 60, &block)
+					while true
+						# Compute the wait time to the next interval:
+						wait = interval - (Time.now.to_f % interval)
+						if wait.positive?
+							# Sleep until the next interval boundary:
+							sleep(wait)
+						end
+						
+						begin
+							yield
+						rescue => error
+							Console.error(self, "Loop error:", error)
+							raise
+						end
+					end
+				end
+			end
+		end
+	end
+end

lib/async/container/supervisor/memory_monitor.rb

@@ -6,6 +6,8 @@
 require "memory/leak/cluster"
 require "set"
 
+require_relative "loop"
+
 module Async
 	module Container
 		module Supervisor
@@ -32,6 +34,9 @@ def initialize(interval: 10, total_size_limit: nil, memory_sample: MEMORY_SAMPLE
 					@processes = Hash.new{|hash, key| hash[key] = Set.new.compare_by_identity}
 				end
 
+				# @attribute [Memory::Leak::Cluster] The cluster of processes being monitored.
+				attr_reader :cluster
+				
 				# Add a process to the memory monitor. You may override this to control how processes are added to the cluster.
 				#
 				# @parameter process_id [Integer] The process ID to add.
@@ -98,8 +103,14 @@ def memory_leak_detected(process_id, monitor)
 					end
 
 					# Kill the process gently:
-					Console.info(self, "Killing process!", child: {process_id: process_id})
-					Process.kill(:INT, process_id)
+					begin
+						Console.info(self, "Killing process!", child: {process_id: process_id})
+						Process.kill(:INT, process_id)
+					rescue Errno::ESRCH
+						# No such process - he's dead Jim.
+					rescue => error
+						Console.warn(self, "Failed to kill process!", child: {process_id: process_id}, exception: error)
+					end
 
 					true
 				end
@@ -109,14 +120,17 @@ def memory_leak_detected(process_id, monitor)
 				# @returns [Async::Task] The task that is running the memory monitor.
 				def run
 					Async do
-						while true
+						Loop.run(interval: @interval) do
 							# This block must return true if the process was killed.
 							@cluster.check! do |process_id, monitor|
 								Console.error(self, "Memory leak detected!", child: {process_id: process_id}, monitor: monitor)
-								memory_leak_detected(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
-							
-							sleep(@interval)
 						end
 					end
 				end

lib/async/container/supervisor/process_monitor.rb

@@ -4,6 +4,9 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "process/metrics"
+require_relative "loop"
+
+require_relative "loop"
 
 module Async
 	module Container
@@ -70,15 +73,13 @@ def status(call)
 				# @returns [Async::Task] The task that is running the process monitor.
 				def run
 					Async do
-						while true
+						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
-							
-							sleep(@interval)
 						end
 					end
 				end

test/async/container/memory_monitor.rb

@@ -21,6 +21,17 @@
 			task = monitor.run
 			expect(task).to be(:running?)
 		end
+		
+		it "can handle failures" do
+			expect(monitor.cluster).to receive(:check!).and_raise(Errno::ESRCH)
+			
+			task = monitor.run
+			expect(task).to be(:running?)
+			
+			sleep 1
+			
+			expect(task).to be(:running?)
+		end
 	end
 end
 

test/async/container/process_monitor.rb

@@ -11,60 +11,32 @@
 describe Async::Container::Supervisor::ProcessMonitor do
 	include Sus::Fixtures::Console::CapturedLogger
 
-	let(:monitor) {subject.new(interval: 10)}
+	let(:monitor) {subject.new(interval: 1)}
 	it_behaves_like Async::Container::Supervisor::AMonitor
 
-	it "has a parent process id" do
-		expect(monitor.ppid).to be == Process.ppid
-	end
-	
-	it "can capture process metrics" do
-		metrics = monitor.metrics
-		
-		# Should capture at least the current process
-		expect(metrics).to be_a(Hash)
-		expect(metrics).not.to be(:empty?)
-		
-		# Check that we have a metric for the current process
-		metric = metrics[Process.pid]
-		expect(metric).not.to be_nil
-		expect(metric.process_id).to be == Process.pid
-		expect(metric.command).to be_a(String)
+	with "#ppid" do
+		it "defaults to the current parent process id" do
+			expect(monitor.ppid).to be == Process.ppid
+		end
 	end
 
-	it "can respond to status calls" do
-		# Create a mock connection and call
-		stream = StringIO.new
-		connection = Async::Container::Supervisor::Connection.new(stream)
+	with "#run" do
+		include Sus::Fixtures::Async::SchedulerContext
 
-		# Create a mock call
-		call_messages = []
-		call = Object.new
-		def call.push(**message)
-			@messages ||= []
-			@messages << message
-		end
-		def call.messages
-			@messages || []
+		it "can run the monitor" do
+			task = monitor.run
+			expect(task).to be(:running?)
 		end
 
-		monitor.status(call)
-		
-		expect(call.messages).not.to be(:empty?)
-		status = call.messages.first
-		expect(status).to have_keys(:process_monitor)
-		expect(status[:process_monitor]).to have_keys(:ppid, :metrics)
-	end
-	
-	it "can register and remove connections" do
-		stream = StringIO.new
-		connection = Async::Container::Supervisor::Connection.new(stream, 0, process_id: Process.pid)
-		
-		# These should not raise errors
-		expect do
-			monitor.register(connection)
-			monitor.remove(connection)
-		end.not.to raise_exception
+		it "can handle failures" do
+			expect(monitor).to receive(:metrics).and_raise(Errno::ESRCH)
+			
+			task = monitor.run
+			expect(task).to be(:running?)
+			
+			sleep 1
+			
+			expect(task).to be(:running?)
+		end
 	end
 end
-