Better handling of failures.

Author: ioquatix

lib/async/container/supervisor/memory_monitor.rb

@@ -91,9 +91,14 @@ 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
 
@@ -113,7 +118,12 @@ def run
 							# 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)

lib/async/container/supervisor/process_monitor.rb

@@ -15,14 +15,14 @@ module Supervisor
 			# by tracking the parent process ID (ppid), which is more efficient than tracking
 			# individual processes.
 			class ProcessMonitor
-			# 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
-				@ppid = ppid || Process.ppid
-			end
+				# 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
+					@ppid = ppid || Process.ppid
+				end
 
 				# @attribute [Integer] The parent process ID being monitored.
 				attr :ppid
@@ -68,21 +68,21 @@ def status(call)
 				# 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
-					while true
-						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)
+				def run
+					Async do
+						while true
+							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
-						
-						sleep(@interval)
 					end
 				end
 			end
-			end
 		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
+