Better handling of failures.

Author: ioquatix

lib/async/container/supervisor/loop.rb

@@ -0,0 +1,31 @@
+module Async
+	module Container
+		module Supervisor
+			# 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.
+			module Loop
+				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)
+						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
@@ -91,15 +93,24 @@ def memory_leak_detected(process_id, monitor)
 
 						# Try to capture a memory sample:
 						connections.each do |connection|
-							result = connection.call(do: :memory_sample, **@memory_sample)
-							
-							Console.info(self, "Memory sample completed:", child: {process_id: process_id}, result: result)
+							begin
+								result = connection.call(do: :memory_sample, **@memory_sample)
+								
+								Console.info(self, "Memory sample completed:", child: {process_id: process_id}, result: result)
+							rescue Errno::EPIPE, IOError => error
+								# Connection already closed - process may have crashed
+								Console.warn(self, "Failed to capture memory sample (connection closed):", child: {process_id: process_id}, error: error)
+							end
 						end
 					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 => 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,7 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "process/metrics"
+require_relative "loop"
 
 module Async
 	module Container
@@ -70,15 +71,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

@@ -0,0 +1,185 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/container/supervisor/memory_monitor"
+require "async/container/supervisor/connection"
+require "sus/fixtures/console/captured_logger"
+
+describe Async::Container::Supervisor::MemoryMonitor do
+	include Sus::Fixtures::Console::CapturedLogger
+	
+	let(:monitor) {subject.new(interval: 10, memory_sample: {duration: 1, timeout: 5})}
+	
+	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
+	end
+	
+	it "can respond to status calls" do
+		# Create a mock call
+		call_messages = []
+		call = Object.new
+		def call.push(**message)
+			@messages ||= []
+			@messages << message
+		end
+		def call.messages
+			@messages || []
+		end
+		
+		monitor.status(call)
+		
+		expect(call.messages).not.to be(:empty?)
+		status = call.messages.first
+		expect(status).to have_keys(:memory_monitor)
+	end
+	
+	with "closed connection handling" do
+		it "handles Errno::EPIPE when connection is broken" do
+			# Use a fake PID that won't kill our test process
+			fake_pid = 999999
+			
+			# Create a mock connection that will raise EPIPE on write
+			stream = StringIO.new
+			connection = Async::Container::Supervisor::Connection.new(stream, 0, process_id: fake_pid)
+			
+			# Mock the call method to raise EPIPE
+			def connection.call(**message)
+				raise Errno::EPIPE, "Broken pipe"
+			end
+			
+			monitor.register(connection)
+			
+			# Mock a simple monitor object
+			mock_monitor = Object.new
+			def mock_monitor.to_s
+				"MockMonitor"
+			end
+			
+			# Mock Process.kill to avoid killing anything
+			expect(Process).to receive(:kill).and_return(nil)
+			
+			# This should handle the exception gracefully and not hang
+			result = monitor.memory_leak_detected(fake_pid, mock_monitor)
+			
+			# Verify the process would have been killed
+			expect(result).to be == true
+		end
+		
+		it "handles IOError when connection stream is closed" do
+			# Use a fake PID that won't kill our test process
+			fake_pid = 999998
+			
+			# Create a connection with a closed stream
+			stream = StringIO.new
+			stream.close
+			connection = Async::Container::Supervisor::Connection.new(stream, 0, process_id: fake_pid)
+			
+			# Mock the call method to raise IOError
+			def connection.call(**message)
+				raise IOError, "closed stream"
+			end
+			
+			monitor.register(connection)
+			
+			# Mock a simple monitor object
+			mock_monitor = Object.new
+			def mock_monitor.to_s
+				"MockMonitor"
+			end
+			
+			# Mock Process.kill to avoid killing anything
+			expect(Process).to receive(:kill).and_return(nil)
+			
+			# This should handle the exception gracefully and not hang
+			result = monitor.memory_leak_detected(fake_pid, mock_monitor)
+			
+			# Verify the process would have been killed
+			expect(result).to be == true
+		end
+		
+		it "handles write errors when stream is already closed" do
+			# Use a fake PID that won't kill our test process
+			fake_pid = 999997
+			
+			read_io, write_io = IO.pipe
+			
+			# Close the read end to simulate a broken pipe
+			read_io.close
+			
+			connection = Async::Container::Supervisor::Connection.new(write_io, 0, process_id: fake_pid)
+			monitor.register(connection)
+			
+			# Mock a simple monitor object
+			mock_monitor = Object.new
+			def mock_monitor.to_s
+				"MockMonitor"
+			end
+			
+			# Mock Process.kill to avoid killing anything
+			expect(Process).to receive(:kill).and_return(nil)
+			
+			# This should handle the exception gracefully
+			result = monitor.memory_leak_detected(fake_pid, mock_monitor)
+			
+			# Verify the process would have been killed
+			expect(result).to be == true
+		ensure
+			write_io&.close
+		end
+		
+		it "continues monitoring even if one connection fails" do
+			# Use a fake PID that won't kill our test process
+			fake_pid = 999996
+			
+			# Create two connections, one will fail
+			good_stream = StringIO.new
+			good_connection = Async::Container::Supervisor::Connection.new(good_stream, 0, process_id: fake_pid)
+			
+			bad_stream = StringIO.new
+			bad_connection = Async::Container::Supervisor::Connection.new(bad_stream, 2, process_id: fake_pid)
+			
+			# Make the bad connection raise on call
+			def bad_connection.call(**message)
+				raise Errno::EPIPE, "Broken pipe"
+			end
+			
+			# Make the good connection succeed
+			def good_connection.call(**message)
+				@call_count ||= 0
+				@call_count += 1
+				{success: true, count: @call_count}
+			end
+			def good_connection.call_count
+				@call_count || 0
+			end
+			
+			monitor.register(bad_connection)
+			monitor.register(good_connection)
+			
+			mock_monitor = Object.new
+			def mock_monitor.to_s
+				"MockMonitor"
+			end
+			
+			# Mock Process.kill to avoid killing anything
+			expect(Process).to receive(:kill).and_return(nil)
+			
+			# With the fix, this should NOT raise because bad_connection's exception is caught
+			result = monitor.memory_leak_detected(fake_pid, mock_monitor)
+			
+			# Both connections should have been attempted - bad one fails, good one succeeds
+			expect(good_connection.call_count).to be == 1
+			expect(result).to be == true
+		end
+	end
+end
+