Better handling of write failures. Fixes #6.

Author: samuel-williams-shopify

lib/async/container/supervisor/connection.rb

@@ -4,6 +4,7 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "json"
+require "async"
 
 module Async
 	module Container
@@ -154,12 +155,14 @@ def self.dispatch(connection, target, id, message)
 								connection.write(id: id, **response)
 							end
 						ensure
-							# If the queue is closed, we don't need to send a finished message.
+							# Ensure the call is removed from the connection's calls hash, otherwise it will leak:
+							connection.calls.delete(id)
+							
+							# If the queue is closed, we don't need to send a finished message:
 							unless call.closed?
-								connection.write(id: id, finished: true)
+								# If the above write failed, this is likely to fail too, and we can safely ignore it.
+								connection.write(id: id, finished: true) rescue nil
 							end
-							
-							connection.calls.delete(id)
 						end
 					end
 

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Better handling of write failures in `Connection::Call.dispatch`, ensuring we don't leak calls.
+
 ## v0.8.0
 
   - Add `Async::Container::Supervisor::ProcessMonitor` for logging CPU and memory metrics periodically.

test/async/container/connection.rb

@@ -6,10 +6,41 @@
 require "async/container/supervisor/connection"
 require "stringio"
 
+class TestTarget
+	def initialize(&block)
+		@block = block
+	end
+	
+	def dispatch(call)
+		@block.call(call)
+	end
+end
+
 describe Async::Container::Supervisor::Connection do
 	let(:stream) {StringIO.new}
 	let(:connection) {Async::Container::Supervisor::Connection.new(stream)}
 
+	with "dispatch" do
+		it "handles failed writes" do
+			stream.write(JSON.dump({id: 1, do: :test}) << "\n")
+			stream.rewind
+			
+			expect(stream).to receive(:write).and_raise(IOError, "Test error")
+			
+			target = TestTarget.new do |call|
+				Async do
+					call.push(status: "working")
+					sleep(0) # Yield back to the dispatch to allow the write to fail.
+					call.finish(status: "done")
+				end
+			end
+			
+			connection.run(target)
+			
+			expect(connection.calls).to be(:empty?)
+		end
+	end
+	
 	with subject::Call do
 		let(:test_call) {Async::Container::Supervisor::Connection::Call.new(connection, 1, {do: :test, data: "value"})}
 

test/async/container/supervisor.rb

@@ -123,6 +123,5 @@ def reader_target.dispatch(call); end
 		ensure
 			worker_task&.stop
 		end
-		
 	end
 end