100% documentation coverage.

Author: samuel-williams-shopify

lib/async/container/supervisor.rb

@@ -13,13 +13,3 @@
 
 require_relative "supervisor/environment"
 require_relative "supervisor/supervised"
-
-# @namespace
-module Async
-	# @namespace
-	module Container
-		# @namespace
-		module Supervisor
-		end
-	end
-end

lib/async/container/supervisor/client.rb

@@ -11,6 +11,9 @@ module Container
 		module Supervisor
 			# A client provides a mechanism to connect to a supervisor server in order to execute operations.
 			class Client
+				# Initialize a new client.
+				#
+				# @parameter endpoint [IO::Endpoint] The supervisor endpoint to connect to.
 				def initialize(endpoint: Supervisor.endpoint)
 					@endpoint = endpoint
 				end

lib/async/container/supervisor/connection.rb

@@ -8,8 +8,19 @@
 module Async
 	module Container
 		module Supervisor
+			# Represents a bidirectional communication channel between supervisor and worker.
+			#
+			# Handles message passing, call/response patterns, and connection lifecycle.
 			class Connection
+				# Represents a remote procedure call over a connection.
+				#
+				# Manages the call lifecycle, response queueing, and completion signaling.
 				class Call
+					# Initialize a new call.
+					#
+					# @parameter connection [Connection] The connection this call belongs to.
+					# @parameter id [Integer] The unique call identifier.
+					# @parameter message [Hash] The call message/parameters.
 					def initialize(connection, id, message)
 						@connection = connection
 						@id = id
@@ -18,10 +29,16 @@ def initialize(connection, id, message)
 						@queue = ::Thread::Queue.new
 					end
 
+					# Convert the call to a JSON-compatible hash.
+					#
+					# @returns [Hash] The message hash.
 					def as_json(...)
 						@message
 					end
 
+					# Convert the call to a JSON string.
+					#
+					# @returns [String] The JSON representation.
 					def to_json(...)
 						as_json.to_json(...)
 					end
@@ -32,14 +49,24 @@ def to_json(...)
 					# @attribute [Hash] The message that initiated the call.
 					attr :message
 
+					# Access a parameter from the call message.
+					#
+					# @parameter key [Symbol] The parameter name.
+					# @returns [Object] The parameter value.
 					def [] key
 						@message[key]
 					end
 
+					# Push a response into the call's queue.
+					#
+					# @parameter response [Hash] The response data to push.
 					def push(**response)
 						@queue.push(response)
 					end
 
+					# Pop a response from the call's queue.
+					#
+					# @returns [Hash, nil] The next response or nil if queue is closed.
 					def pop(...)
 						@queue.pop(...)
 					end
@@ -49,12 +76,20 @@ def close
 						@queue.close
 					end
 
+					# Iterate over all responses from the call.
+					#
+					# @yields {|response| ...} Each response from the queue.
 					def each(&block)
 						while response = self.pop
 							yield response
 						end
 					end
 
+					# Finish the call with a final response.
+					#
+					# Closes the response queue after pushing the final response.
+					#
+					# @parameter response [Hash] The final response data.
 					def finish(**response)
 						# If the remote end has already closed the connection, we don't need to send a finished message:
 						unless @queue.closed?
@@ -63,18 +98,25 @@ def finish(**response)
 						end
 					end
 
+					# Finish the call with a failure response.
+					#
+					# @parameter response [Hash] The error response data.
 					def fail(**response)
 						self.finish(failed: true, **response)
 					end
 
+					# Check if the call's queue is closed.
+					#
+					# @returns [Boolean] True if the queue is closed.
 					def closed?
 						@queue.closed?
 					end
 
 					# Forward this call to another connection, proxying all responses back.
 					#
 					# This provides true streaming forwarding - intermediate responses flow through
-					# in real-time rather than being buffered.
+					# in real-time rather than being buffered. The forwarding runs asynchronously
+					# to avoid blocking the dispatcher.
 					#
 					# @parameter target_connection [Connection] The connection to forward the call to.
 					# @parameter operation [Hash] The operation request to forward (must include :do key).
@@ -92,6 +134,15 @@ def forward(target_connection, operation)
 						end
 					end
 
+					# Dispatch a call to a target handler.
+					#
+					# Creates a call, dispatches it to the target, and streams responses back
+					# through the connection.
+					#
+					# @parameter connection [Connection] The connection to dispatch on.
+					# @parameter target [Dispatchable] The target handler.
+					# @parameter id [Integer] The call identifier.
+					# @parameter message [Hash] The call message.
 					def self.dispatch(connection, target, id, message)
 						Async do
 							call = self.new(connection, id, message)
@@ -112,6 +163,15 @@ def self.dispatch(connection, target, id, message)
 						end
 					end
 
+					# Make a call on a connection and wait for responses.
+					#
+					# If a block is provided, yields each response. Otherwise, buffers intermediate
+					# responses and returns the final response.
+					#
+					# @parameter connection [Connection] The connection to call on.
+					# @parameter message [Hash] The call message/parameters.
+					# @yields {|response| ...} Each intermediate response if block given.
+					# @returns [Hash, Array] The final response or array of intermediate responses.
 					def self.call(connection, **message, &block)
 						id = connection.next_id
 						call = self.new(connection, id, message)
@@ -149,6 +209,11 @@ def self.call(connection, **message, &block)
 					end
 				end
 
+				# Initialize a new connection.
+				#
+				# @parameter stream [IO] The underlying IO stream.
+				# @parameter id [Integer] The starting call ID (default: 0).
+				# @parameter state [Hash] Initial connection state.
 				def initialize(stream, id = 0, **state)
 					@stream = stream
 					@id = id
@@ -164,15 +229,26 @@ def initialize(stream, id = 0, **state)
 				# @attribute [Hash(Symbol, Object)] State associated with this connection, for example the process ID, etc.
 				attr_accessor :state
 
+				# Generate the next unique call ID.
+				#
+				# @returns [Integer] The next call identifier.
 				def next_id
 					@id += 2
 				end
 
+				# Write a message to the connection stream.
+				#
+				# @parameter message [Hash] The message to write.
 				def write(**message)
 					@stream.write(JSON.dump(message) << "\n")
 					@stream.flush
 				end
 
+				# Make a synchronous call and wait for a single response.
+				#
+				# @parameter timeout [Numeric, nil] Optional timeout for the call.
+				# @parameter message [Hash] The call message.
+				# @returns [Hash] The response.
 				def call(timeout: nil, **message)
 					id = next_id
 					calls[id] = ::Thread::Queue.new
@@ -184,22 +260,34 @@ def call(timeout: nil, **message)
 					calls.delete(id)
 				end
 
+				# Read a message from the connection stream.
+				#
+				# @returns [Hash, nil] The parsed message or nil if stream is closed.
 				def read
 					if line = @stream&.gets
 						JSON.parse(line, symbolize_names: true)
 					end
 				end
 
+				# Iterate over all messages from the connection.
+				#
+				# @yields {|message| ...} Each message read from the stream.
 				def each
 					while message = self.read
 						yield message
 					end
 				end
 
+				# Make a synchronous call and wait for a single response.
 				def call(...)
 					Call.call(self, ...)
 				end
 
+				# Run the connection, processing incoming messages.
+				#
+				# Dispatches incoming calls to the target and routes responses to waiting calls.
+				#
+				# @parameter target [Dispatchable] The target to dispatch calls to.
 				def run(target)
 					self.each do |message|
 						if id = message.delete(:id)
@@ -219,12 +307,20 @@ def run(target)
 					end
 				end
 
+				# Run the connection in a background task.
+				#
+				# @parameter target [Dispatchable] The target to dispatch calls to.
+				# @parameter parent [Async::Task] The parent task.
+				# @returns [Async::Task] The background reader task.
 				def run_in_background(target, parent: Task.current)
 					@reader ||= parent.async do
 						self.run(target)
 					end
 				end
 
+				# Close the connection and clean up resources.
+				#
+				# Stops the background reader, closes the stream, and closes all pending calls.
 				def close
 					if @reader
 						@reader.stop

lib/async/container/supervisor/dispatchable.rb

@@ -9,7 +9,15 @@
 module Async
 	module Container
 		module Supervisor
+			# A mixin for objects that can dispatch calls.
+			#
+			# Provides automatic method dispatch based on the call's `:do` parameter.
 			module Dispatchable
+				# Dispatch a call to the appropriate method.
+				#
+				# Routes calls to methods named `do_#{operation}` based on the call's `:do` parameter.
+				#
+				# @parameter call [Connection::Call] The call to dispatch.
 				def dispatch(call)
 					method_name = "do_#{call.message[:do]}"
 					self.public_send(method_name, call)

lib/async/container/supervisor/endpoint.rb

@@ -8,6 +8,10 @@
 module Async
 	module Container
 		module Supervisor
+			# Get the supervisor IPC endpoint.
+			#
+			# @parameter path [String] The path for the Unix socket (default: "supervisor.ipc").
+			# @returns [IO::Endpoint] The Unix socket endpoint.
 			def self.endpoint(path = "supervisor.ipc")
 				::IO::Endpoint.unix(path)
 			end

lib/async/container/supervisor/environment.rb

@@ -10,6 +10,9 @@
 module Async
 	module Container
 		module Supervisor
+			# An environment mixin for supervisor services.
+			#
+			# Provides configuration and setup for supervisor processes that monitor workers.
 			module Environment
 				# The service class to use for the supervisor.
 				# @returns [Class]
@@ -40,10 +43,18 @@ def container_options
 					{restart: true, count: 1, health_check_timeout: 30}
 				end
 
+				# Get the list of monitors to run in the supervisor.
+				#
+				# Override this method to provide custom monitors.
+				#
+				# @returns [Array] The list of monitor instances.
 				def monitors
 					[]
 				end
 
+				# Create the supervisor server instance.
+				#
+				# @returns [Server] The supervisor server.
 				def make_server(endpoint)
 					Server.new(endpoint: endpoint, monitors: self.monitors)
 				end

lib/async/container/supervisor/memory_monitor.rb

@@ -9,6 +9,9 @@
 module Async
 	module Container
 		module Supervisor
+			# Monitors worker memory usage and restarts workers that exceed limits.
+			#
+			# Uses the `memory` gem to track process memory and detect leaks.
 			class MemoryMonitor
 				MEMORY_SAMPLE = {duration: 60, timeout: 60+20}