Restore worker state interface.

Author: samuel-williams-shopify

guides/migration/readme.md

@@ -309,3 +309,9 @@ The new approach provides:
 - Simpler ID management.
 - Better controller-based API.
 - More explicit worker lifecycle management.
+
+### Connection State Handling
+
+State has moved from `connection.state` to `supervisor_controller.state`. Monitors receive `supervisor_controller` objects instead of `connection` objects, and access state via `supervisor_controller.state[:name]` instead of `connection.state[:name]`.
+
+By default, the `Supervised` module automatically includes the service name in state via `supervisor_worker_state`. Override `supervisor_worker_state` in your service environment to customize the state.

lib/async/service/supervisor/supervised.rb

@@ -24,10 +24,16 @@ def supervisor_endpoint
 					::IO::Endpoint.unix(supervisor_ipc_path)
 				end
 
+				# The state to associate with the supervised worker.
+				# @returns [Hash]
+				def supervisor_worker_state
+					{name: self.name}
+				end
+				
 				# The supervised worker for the current process.
 				# @returns [Worker] The worker client.
 				def supervisor_worker
-					Worker.new(process_id: Process.pid, endpoint: supervisor_endpoint)
+					Worker.new(process_id: Process.pid, endpoint: supervisor_endpoint, state: self.supervisor_worker_state)
 				end
 
 				# Create a supervised worker for the given instance.

lib/async/service/supervisor/supervisor_controller.rb

@@ -19,6 +19,7 @@ def initialize(server, connection)
 					@id = nil
 					@process_id = nil
 					@worker = nil
+					@state = {}
 				end
 
 				# @attribute [Server] The server instance.
@@ -36,20 +37,25 @@ def initialize(server, connection)
 				# @attribute [Proxy] The proxy to the worker controller.
 				attr :worker
 
+				# @attribute [Hash] State associated with this worker connection (e.g., service name).
+				attr_accessor :state
+				
 				# Register a worker connection with the supervisor.
 				#
 				# Allocates a unique sequential ID, stores the worker controller proxy,
 				# and notifies all monitors of the new connection.
 				#
 				# @parameter worker [Proxy] The proxy to the worker controller.
 				# @parameter process_id [Integer] The process ID of the worker.
+				# @parameter state [Hash] Optional state to associate with this worker (e.g., service name).
 				# @returns [Integer] The connection ID assigned to the worker.
-				def register(worker, process_id:)
+				def register(worker, process_id:, state: {})
 					raise RuntimeError, "Already registered" if @id
 
 					@id = @server.next_id
 					@process_id = process_id
 					@worker = worker
+					@state.merge!(state)
 
 					@server.add(self)
 

lib/async/service/supervisor/worker.rb

@@ -25,11 +25,13 @@ def self.run(process_id: Process.pid, endpoint: Supervisor.endpoint)
 				#
 				# @parameter process_id [Integer] The process ID to register with the supervisor.
 				# @parameter endpoint [IO::Endpoint] The supervisor endpoint to connect to.
-				def initialize(process_id: Process.pid, endpoint: Supervisor.endpoint)
+				# @parameter state [Hash] Optional state to associate with this worker (e.g., service name).
+				def initialize(process_id: Process.pid, endpoint: Supervisor.endpoint, state: {})
 					super(endpoint: endpoint)
 
 					@id = nil
 					@process_id = process_id
+					@state = state
 				end
 
 				# @attribute [Integer] The ID assigned by the supervisor.
@@ -38,6 +40,9 @@ def initialize(process_id: Process.pid, endpoint: Supervisor.endpoint)
 				# @attribute [Integer] The process ID of the worker.
 				attr :process_id
 
+				# @attribute [Hash] State associated with this worker (e.g., service name).
+				attr_accessor :state
+				
 				protected def connected!(connection)
 					super
 
@@ -49,10 +54,9 @@ def initialize(process_id: Process.pid, endpoint: Supervisor.endpoint)
 					# The supervisor allocates a unique ID and returns it
 					# This is a synchronous RPC call that will complete before returning
 					supervisor = connection[:supervisor]
-					@id = supervisor.register(worker_proxy, process_id: @process_id)
+					@id = supervisor.register(worker_proxy, process_id: @process_id, state: @state)
 				end
 			end
 		end
 	end
 end
-

releases.md

@@ -1,5 +1,11 @@
 # Releases
 
+## Unreleased
+
+  - Add `state` attribute to `SupervisorController` to store per-worker metadata (e.g., service name).
+  - Add `state` parameter to `Worker#initialize` to allow workers to provide state during registration.
+  - State is now accessible via `supervisor_controller.state` instead of `connection.state` (as it was in `Async::Container::Supervisor`).
+
 ## v0.10.0
 
   - Serialize `register`/`remove` and `check!` operations in `MemoryMonitor` to prevent race conditions.

test/async/service/supervised.rb

@@ -33,6 +33,8 @@ def setup(container)
 
 	it "can define a supervised service" do
 		environment = Async::Service::Environment.build(root: @root) do
+			name "simple-service"
+
 			service_class {SimpleService}
 
 			include Async::Service::Supervisor::Supervised
@@ -48,6 +50,9 @@ def setup(container)
 		supervisor_controller = event.supervisor_controller
 
 		expect(supervisor_controller.process_id).to be == ::Process.pid
+		expect(supervisor_controller.state).to have_keys(
+			name: be == "simple-service"
+		)
 	ensure
 		worker_task&.stop
 	end

test/async/service/supervisor.rb

@@ -23,6 +23,67 @@
 		worker_task&.stop
 	end
 
+	with "state handling" do
+		it "stores state on supervisor_controller during registration" do
+			custom_state = {name: "test-service", version: "1.0"}
+			worker = Async::Service::Supervisor::Worker.new(
+				process_id: ::Process.pid,
+				endpoint: endpoint,
+				state: custom_state
+			)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			expect(supervisor_controller.state).to have_keys(
+				name: be == "test-service",
+				version: be == "1.0"
+			)
+		ensure
+			worker_task&.stop
+		end
+		
+		it "defaults to empty state hash" do
+			worker = Async::Service::Supervisor::Worker.new(process_id: ::Process.pid, endpoint: endpoint)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			expect(supervisor_controller.state).to be == {}
+		ensure
+			worker_task&.stop
+		end
+		
+		it "allows monitors to access state" do
+			custom_state = {name: "outbox"}
+			worker = Async::Service::Supervisor::Worker.new(
+				process_id: ::Process.pid,
+				endpoint: endpoint,
+				state: custom_state
+			)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			# Verify state can be accessed by monitors
+			expect(supervisor_controller.state[:name]).to be == "outbox"
+		ensure
+			worker_task&.stop
+		end
+	end
+	
 	with "memory_dump" do
 		it "can dump memory" do
 			worker = Async::Service::Supervisor::Worker.new(process_id: ::Process.pid, endpoint: endpoint)
@@ -113,5 +174,66 @@
 		end
 	end
 
+	with "state handling" do
+		it "stores state on supervisor_controller during registration" do
+			custom_state = {name: "test-service", version: "1.0"}
+			worker = Async::Service::Supervisor::Worker.new(
+				process_id: ::Process.pid,
+				endpoint: endpoint,
+				state: custom_state
+			)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			expect(supervisor_controller.state).to have_keys(
+				name: be == "test-service",
+				version: be == "1.0"
+			)
+		ensure
+			worker_task&.stop
+		end
+		
+		it "defaults to empty state hash" do
+			worker = Async::Service::Supervisor::Worker.new(process_id: ::Process.pid, endpoint: endpoint)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			expect(supervisor_controller.state).to be == {}
+		ensure
+			worker_task&.stop
+		end
+		
+		it "allows monitors to access state" do
+			custom_state = {name: "outbox"}
+			worker = Async::Service::Supervisor::Worker.new(
+				process_id: ::Process.pid,
+				endpoint: endpoint,
+				state: custom_state
+			)
+			worker_task = worker.run
+			
+			# Wait for registration via the registration monitor
+			event = registration_monitor.pop(timeout: 5)
+			
+			expect(event).to be_truthy
+			supervisor_controller = event.supervisor_controller
+			
+			# Verify state can be accessed by monitors
+			expect(supervisor_controller.state[:name]).to be == "outbox"
+		ensure
+			worker_task&.stop
+		end
+	end
+	
 end