Fix child lifecycle timeout handling

Assisted-By: devx/a01026a8-dd63-4b91-bae4-aec7ee8f7fa3

Author: samuel-williams-shopify

lib/async/container/channel.rb

@@ -10,9 +10,9 @@ module Container
 		# Provides a basic multi-thread/multi-process uni-directional communication channel.
 		class Channel
 			# Initialize the channel using a pipe.
-			def initialize(timeout: 1.0)
+			def initialize(timeout: nil)
 				@in, @out = ::IO.pipe
-				@in.timeout = timeout
+				@in.timeout = timeout if timeout
 			end
 
 			# The input end of the pipe.
@@ -46,8 +46,7 @@ def receive
 				if data = @in.gets
 					return JSON.parse(data, symbolize_names: true)
 				end
-			rescue => error
-				Console.error(self, "Error during channel receive!", error)
+			rescue JSON::ParserError
 				return nil
 			end
 		end

lib/async/container/generic.rb

@@ -98,7 +98,7 @@ def running?
 
 			# Whether the container is stopping.
 			def stopping?
-				@stopping
+				@mutex.synchronize{@stopping}
 			end
 
 			# Sleep until a group event occurs or the specified duration elapses.
@@ -139,7 +139,7 @@ def wait
 
 			# Interrupt all children and enter the stopping state.
 			def interrupt
-				@stopping = true
+				stopping!
 				@group.interrupt!
 			end
 
@@ -155,15 +155,15 @@ def wait_until_ready
 					return false if failed? && !lifecycle_running?
 					return true if !running? && !lifecycle_running?
 
-					@group.wait
+					@group.wait(0.1)
 				end
 			end
 
 			# Stop all children.
 			def stop(timeout = true)
-				return if @stopping && !running?
+				return if stopping? && !running?
 
-				@stopping = true
+				stopping!
 				@group.stop(timeout)
 				self.wait
 			end
@@ -229,6 +229,10 @@ def lifecycle_running?
 				@mutex.synchronize{@threads.any?(&:alive?)}
 			end
 
+			def stopping!
+				@mutex.synchronize{@stopping = true}
+			end
+			
 			def start_child(name:, **options, &block)
 				@child_type.call(name: name, **options, &block)
 			end
@@ -265,7 +269,7 @@ def manage_child(child, name:, key:, restart:, health_check_timeout:, startup_ti
 					unregister_child(child, status, key: key)
 					notify_child_exit(child, status, name: name, key: key)
 
-					if restart && !@stopping
+					if restart && !stopping?
 						@statistics.restart!
 
 						child = start_child(name: name, **options, &block)
@@ -275,7 +279,7 @@ def manage_child(child, name:, key:, restart:, health_check_timeout:, startup_ti
 					end
 				end
 			rescue => error
-				Console.error(self, "Error during child lifecycle management!", exception: error, stopping: @stopping) if defined?(Console)
+				Console.error(self, "Error during child lifecycle management!", exception: error, stopping: stopping?) if defined?(Console)
 			ensure
 				if @group.children.include?(child)
 					begin
@@ -322,13 +326,15 @@ def monitor_child_with_timeouts(child, health_check_timeout:, startup_timeout:)
 			end
 
 			def record_exit(child, status, name:, key:)
+				stopping = stopping?
+				
 				if status&.success?
-					Console.debug(self, "Child exited successfully.", status: status, stopping: @stopping) if defined?(Console)
-				elsif @stopping
-					Console.debug(self, "Child exited while stopping.", status: status, stopping: @stopping) if defined?(Console)
+					Console.debug(self, "Child exited successfully.", status: status, stopping: stopping) if defined?(Console)
+				elsif stopping
+					Console.debug(self, "Child exited while stopping.", status: status, stopping: stopping) if defined?(Console)
 				else
 					@statistics.failure!
-					Console.error(self, "Child exited with error!", status: status, stopping: @stopping) if defined?(Console)
+					Console.error(self, "Child exited with error!", status: status, stopping: stopping) if defined?(Console)
 				end
 			end
 

test/async/container/channel.rb

@@ -59,8 +59,10 @@
 	with "timeout" do
 		let(:channel) {subject.new(timeout: 0.001)}
 
-		it "fails gracefully on timeout" do
-			expect(channel.receive).to be_nil
+		it "does not treat timeouts as channel closure" do
+			expect do
+				channel.receive
+			end.to raise_exception(IO::TimeoutError)
 		end
 	end
 end

test/async/container/generic.rb

@@ -37,6 +37,15 @@
 		expect(container).to be(:failed?)
 	end
 
+	it "does not block waiting for readiness after a child fails to start" do
+		container.spawn do
+			raise "boom"
+		end
+		
+		expect(container.wait_until_ready).to be == false
+		expect(container).not.to be(:running?)
+	end
+	
 	it "restarts children when requested" do
 		count = 0
 
@@ -54,6 +63,21 @@
 		expect(container.statistics).to have_attributes(spawns: be == 2, restarts: be == 1, failures: be == 1)
 	end
 
+	it "does not restart children while stopping" do
+		container.spawn(restart: true) do |instance|
+			instance.ready!
+			sleep
+		rescue Interrupt
+			# Graceful shutdown.
+		end
+		
+		expect(container.wait_until_ready).to be == true
+		
+		container.stop(true)
+		
+		expect(container.statistics).to have_attributes(spawns: be == 1, restarts: be == 0)
+	end
+	
 	it "supports keyed children" do
 		expect(container.spawn(key: :worker) do |instance|
 			instance.ready!