Fix reload reconciliation of keyed children

lib/async/container/channel.rb

@@ -13,6 +13,7 @@ class Channel
 			def initialize(timeout: 1.0)
 				@in, @out = ::IO.pipe
 				@in.timeout = timeout
+				@stopping = false
 			end
 
 			# The input end of the pipe.
@@ -23,6 +24,17 @@ def initialize(timeout: 1.0)
 			# @attribute [IO]
 			attr :out
 
+			# Whether this child is being deliberately stopped and should not be restarted.
+			# @returns [Boolean]
+			def stopping?
+				@stopping
+			end
+
+			# Mark this child as being deliberately stopped, so its supervising fiber will not restart it.
+			def stopping!
+				@stopping = true
+			end
+
 			# Close the input end of the pipe.
 			def close_read
 				@in.close

lib/async/container/generic.rb

@@ -144,8 +144,7 @@ def wait_until_ready
 						end
 					end
 
-					self.sleep
-
+					# Check readiness before sleeping: after a reload that only reuses or removes children, no new child sends a readiness message, so `sleep` would otherwise block indefinitely even though everything is already ready.
 					if self.status?(:ready)
 						Console.debug(self) do |buffer|
 							buffer.puts "All ready:"
@@ -156,6 +155,8 @@ def wait_until_ready
 
 						return true
 					end
+
+					self.sleep
 				end
 			end
 
@@ -294,7 +295,7 @@ def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startu
 							Console.error(self, "Policy error in child_exit!", exception: error)
 						end
 
-						if restart && !@stopping
+						if restart && !@stopping && !child.stopping?
 							@statistics.restart!
 						else
 							break
@@ -327,20 +328,39 @@ def async(**options, &block)
 			end
 
 			# Reload the container's keyed instances.
+			#
+			# Any keyed child which is not re-marked during {yield} is considered obsolete and is
+			# stopped. Its supervising fiber removes it from {@keyed} once it has exited.
+			#
+			# @returns [Boolean] Whether any keyed instances were stopped.
 			def reload
 				@keyed.each_value(&:clear!)
 
 				yield
 
 				dirty = false
 
-				@keyed.delete_if do |key, value|
-					value.stop? && (dirty = true)
+				# Snapshot the values so we can stop obsolete children without mutating `@keyed` mid-iteration (the supervising fiber deletes the entry when the child exits):
+				@keyed.values.each do |keyed|
+					unless keyed.marked?
+						stop_child(keyed.value)
+						dirty = true
+					end
 				end
 
 				return dirty
 			end
 
+			# Stop a single child instance and prevent it from being restarted.
+			# @parameter child [Channel] The child instance to stop.
+			# @parameter graceful [Boolean | Numeric] Whether to stop the child gracefully.
+			def stop_child(child, graceful = true)
+				# Prevent the supervising fiber from restarting the child once it exits:
+				child.stopping!
+
+				@group.stop_child(child, graceful)
+			end
+
 			# Mark the container's keyed instance which ensures that it won't be discarded.
 			def mark?(key)
 				if key

lib/async/container/group.rb

@@ -180,6 +180,48 @@ def stop(graceful = true)
 				end
 			end
 
+			# Stop a single running child, gracefully if possible, then forcefully.
+			#
+			# Mirrors the multi-phase sequence of {stop}, but scoped to one child: send SIGINT and
+			# wait up to `graceful` seconds, then send SIGKILL and wait for it to exit.
+			#
+			# @parameter channel [Channel] The channel of the child to stop.
+			# @parameter graceful [Boolean | Numeric] Whether to interrupt first, or a specific timeout.
+			def stop_child(channel, graceful = true)
+				io = channel.in
+				fiber = @running[io]
+
+				return unless fiber
+
+				if graceful
+					# Send SIGINT to the child:
+					fiber.resume(Interrupt)
+
+					if graceful == true
+						graceful = DEFAULT_GRACEFUL_TIMEOUT
+					end
+
+					clock = Clock.start
+
+					# Wait for the child to exit:
+					while @running.key?(io)
+						duration = graceful - clock.total
+						break if duration < 0
+
+						wait_for_children(duration)
+					end
+				end
+			ensure
+				# Force kill if it's still running:
+				if fiber && @running.key?(io)
+					fiber.resume(Kill)
+
+					while @running.key?(io)
+						wait_for_children(nil)
+					end
+				end
+			end
+
 			# Wait for a message in the specified {Channel}.
 			def wait_for(channel)
 				io = channel.in

lib/async/container/keyed.rb

@@ -38,16 +38,6 @@ def mark!
 			def clear!
 				@marked = false
 			end
-
-			# Stop the instance if it was not marked.
-			#
-			# @returns [Boolean] True if the instance was stopped.
-			def stop?
-				unless @marked
-					@value.stop
-					return true
-				end
-			end
 		end
 	end
 end

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - **Fixed**: Reloading a container now correctly stops keyed children whose keys are no longer configured. Previously this raised `NoMethodError` (the child has no `stop` method) and would not have stopped `restart: true` children even if it had. `Generic#reload` now stops obsolete children via a new `Group#stop_child`, and a per-child stopping flag prevents them from being respawned.
+
 ## v0.37.0
 
   - Rename `ASYNC_CONTAINER_GRACEFUL_TIMEOUT` to `ASYNC_CONTAINER_GRACEFUL_STOP` and apply it at the controller level as `GRACEFUL_STOP`. `Group#stop` now only applies the shutdown policy it is given.

test/async/container/controller.rb

@@ -82,6 +82,64 @@ def controller.setup(container)
 
 			controller.wait
 		end
+
+		it "spawns a newly configured keyed child on reload" do
+			keys = ["a"]
+
+			controller.define_singleton_method(:setup) do |container|
+				container.reload do
+					keys.each do |key|
+						container.spawn(key: key) do |instance|
+							instance.ready!
+							sleep
+						end
+					end
+				end
+			end
+
+			controller.start
+
+			expect(controller.container["a"]).not.to be_nil
+			expect(controller.container["b"]).to be_nil
+
+			# The configuration now includes an additional keyed child:
+			keys << "b"
+			controller.reload
+
+			expect(controller.container["a"]).not.to be_nil
+			expect(controller.container["b"]).not.to be_nil
+		ensure
+			controller.stop(false)
+		end
+
+		it "stops a keyed child that is no longer configured on reload" do
+			keys = ["a", "b"]
+
+			controller.define_singleton_method(:setup) do |container|
+				container.reload do
+					keys.each do |key|
+						container.spawn(key: key) do |instance|
+							instance.ready!
+							sleep
+						end
+					end
+				end
+			end
+
+			controller.start
+
+			expect(controller.container["a"]).not.to be_nil
+			expect(controller.container["b"]).not.to be_nil
+
+			# The configuration no longer includes "b", so reload should stop it:
+			keys.delete("b")
+			controller.reload
+
+			expect(controller.container["a"]).not.to be_nil
+			expect(controller.container["b"]).to be_nil
+		ensure
+			controller.stop(false)
+		end
 	end
 
 	with "notify" do