Add and clean up monitor tests.

Author: samuel-williams-shopify

fixtures/async/container/supervisor/a_monitor.rb

@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/container/supervisor/a_server"
+
+module Async
+	module Container
+		module Supervisor
+			AMonitor = Sus::Shared("a monitor") do
+				include_context AServer
+				
+				let(:monitors) {[monitor, registration_monitor]}
+				
+				it "can add and remove connections" do
+					worker = Worker.new(endpoint: endpoint)
+					connection = worker.connect
+					
+					event = registration_monitor.pop
+					expect(event).to have_attributes(
+						type: be == :register,
+					)
+					
+					connection.close
+					
+					event = registration_monitor.pop
+					expect(event).to have_attributes(
+						type: be == :remove,
+					)
+				end
+			end
+		end
+	end
+end

fixtures/async/container/supervisor/a_server.rb

@@ -13,21 +13,31 @@ module Async
 	module Container
 		module Supervisor
 			class RegistrationMonitor
+				Event = Struct.new(:type, :connection)
+				
 				def initialize
-					@registrations = []
+					@registrations = ::Thread::Queue.new
 				end
 
 				attr :registrations
 
 				def run
 				end
 
+				def status(call)
+					call.push(registrations: @registrations.size)
+				end
+				
+				def pop(...)
+					@registrations.pop(...)
+				end
+				
 				def register(connection)
-					@registrations << connection
+					@registrations << Event.new(:register, connection)
 				end
 
 				def remove(connection)
-					@registrations.delete(connection)
+					@registrations << Event.new(:remove, connection)
 				end
 			end
 

lib/async/container/supervisor/server.rb

@@ -39,7 +39,9 @@ def initialize(monitors: [], endpoint: Supervisor.endpoint)
 				# @parameter call [Connection::Call] The registration call.
 				# @parameter call[:state] [Hash] The worker state to merge (e.g. process_id).
 				def do_register(call)
-					call.connection.state.merge!(call.message[:state])
+					if state = call.message[:state]
+						call.connection.state.merge!(state)
+					end
 
 					connection_id = SecureRandom.uuid
 					call.connection.state[:connection_id] = connection_id
@@ -52,7 +54,7 @@ def do_register(call)
 						Console.error(self, "Error while registering process!", monitor: monitor, exception: error)
 					end
 				ensure
-					call.finish
+					call.finish(connection_id: connection_id)
 				end
 
 				# Forward an operation to a worker connection.

lib/async/container/supervisor/worker.rb

@@ -25,9 +25,9 @@ def self.run(...)
 				#
 				# @parameter state [Hash] The worker state to register with the supervisor.
 				# @parameter endpoint [IO::Endpoint] The supervisor endpoint to connect to.
-				def initialize(state, endpoint: Supervisor.endpoint)
+				def initialize(state = nil, endpoint: Supervisor.endpoint)
+					super(endpoint: endpoint)
 					@state = state
-					@endpoint = endpoint
 				end
 
 				include Dispatchable
@@ -164,6 +164,7 @@ def do_garbage_profile_stop(call)
 
 					# Register the worker with the supervisor:
 					connection.call(do: :register, state: @state)
+					# We ignore the response (it contains the `connection_id`).
 				end
 			end
 		end

test/async/container/memory_monitor.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/container/supervisor/memory_monitor"
+require "async/container/supervisor/a_monitor"
+
+require "sus/fixtures/console/captured_logger"
+
+describe Async::Container::Supervisor::MemoryMonitor do
+	include Sus::Fixtures::Console::CapturedLogger
+	
+	let(:monitor) {subject.new(interval: 1, memory_sample: {duration: 1, timeout: 5})}
+	it_behaves_like Async::Container::Supervisor::AMonitor
+	
+	with "#run" do
+		include Sus::Fixtures::Async::SchedulerContext
+		
+		it "can run the monitor" do
+			task = monitor.run
+			expect(task).to be(:running?)
+		end
+		
+		it "can handle failures" do
+			expect(monitor.cluster).to receive(:check!).and_raise(Errno::ESRCH)
+			
+			task = monitor.run
+			expect(task).to be(:running?)
+			
+			sleep 1
+			
+			expect(task).to be(:running?)
+		end
+	end
+end
+

test/async/container/process_monitor.rb

@@ -4,14 +4,17 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "async/container/supervisor/process_monitor"
-require "async/container/supervisor/connection"
+require "async/container/supervisor/a_monitor"
+
 require "sus/fixtures/console/captured_logger"
 
 describe Async::Container::Supervisor::ProcessMonitor do
 	include Sus::Fixtures::Console::CapturedLogger
+	
 	let(:monitor) {subject.new(interval: 10)}
+	it_behaves_like Async::Container::Supervisor::AMonitor
 
-	it "has a ppid" do
+	it "has a parent process id" do
 		expect(monitor.ppid).to be == Process.ppid
 	end
 

test/async/container/supervised.rb

@@ -42,9 +42,10 @@ def setup(container)
 		worker = evaluator.make_supervised_worker(state)
 		worker_task = worker.run
 
-		sleep(0.001) until registration_monitor.registrations.any?
+		# Wait for the worker to register with the supervisor.
+		event = registration_monitor.pop
+		connection = event.connection
 
-		connection = registration_monitor.registrations.first
 		expect(connection.state).to have_keys(
 			process_id: be == ::Process.pid
 		)

test/async/container/supervisor.rb

@@ -17,9 +17,9 @@
 		connection = worker.connect
 
 		# Wait for the client to connect to the server:
-		sleep(0.001) until registration_monitor.registrations.any?
+		event = registration_monitor.pop
+		connection = event.connection
 
-		connection = registration_monitor.registrations.first
 		expect(connection.state).to have_keys(
 			process_id: be == ::Process.pid
 		)
@@ -32,10 +32,10 @@
 			worker = Async::Container::Supervisor::Worker.new(state, endpoint: endpoint)
 			worker_task = worker.run
 
-			sleep(0.001) until registration_monitor.registrations.any?
+			event = registration_monitor.pop
+			connection = event.connection
 
 			path = File.join(@root, "memory.json")
-			connection = registration_monitor.registrations.first
 			connection.call(do: :memory_dump, path: path)
 
 			expect(File.size(path)).to be > 0
@@ -110,9 +110,8 @@ def reader_target.dispatch(call); end
 			worker = Async::Container::Supervisor::Worker.new(state, endpoint: endpoint)
 			worker_task = worker.run
 
-			sleep(0.001) until registration_monitor.registrations.any?
-			
-			connection = registration_monitor.registrations.first
+			event = registration_monitor.pop
+			connection = event.connection
 
 			# Sample for a short duration (1 second for test speed)
 			result = connection.call(do: :memory_sample, duration: 1)