Deduplicate shared listener endpoints

context/getting-started.md

@@ -41,16 +41,18 @@ state = {
 
 Workers without endpoint state are ignored by the Envoy monitor.
 
-Falcon cluster workers can register their concrete post-bind listener automatically:
+Falcon server workers can register their concrete bound listener automatically:
 
 ``` ruby
 service "application" do
-	include Falcon::Environment::Cluster
+	include Falcon::Environment::Server
 	include Async::Service::Supervisor::Envoy::Supervised
 end
 ```
 
-Falcon describes its bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint.
+`Falcon::Environment::Server` reports one listener shared by all of its workers, while `Falcon::Environment::Cluster` reports the listener bound independently by each worker. Falcon describes either bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address.
+
+Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. When several workers report the same shared listener, the monitor publishes it once and keeps it available while any reporting worker is healthy.
 
 ## Monitor Usage
 

guides/getting-started/readme.md

@@ -41,16 +41,18 @@ state = {
 
 Workers without endpoint state are ignored by the Envoy monitor.
 
-Falcon cluster workers can register their concrete post-bind listener automatically:
+Falcon server workers can register their concrete bound listener automatically:
 
 ``` ruby
 service "application" do
-	include Falcon::Environment::Cluster
+	include Falcon::Environment::Server
 	include Async::Service::Supervisor::Envoy::Supervised
 end
 ```
 
-Falcon describes its bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint.
+`Falcon::Environment::Server` reports one listener shared by all of its workers, while `Falcon::Environment::Cluster` reports the listener bound independently by each worker. Falcon describes either bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address.
+
+Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. When several workers report the same shared listener, the monitor publishes it once and keeps it available while any reporting worker is healthy.
 
 ## Monitor Usage
 

lib/async/service/supervisor/envoy/endpoint.rb

@@ -7,8 +7,23 @@ module Async
 	module Service
 		module Supervisor
 			module Envoy
-				# Represents one upstream endpoint published to Envoy EDS.
+				# Represents an upstream endpoint published to Envoy EDS.
 				class Endpoint
+					# Build an immutable endpoint.
+					# @parameter name [String] The upstream cluster name.
+					# @parameter scheme [String | Symbol] The upstream application scheme.
+					# @parameter protocols [Array(String)] The supported upstream HTTP protocol names.
+					# @parameter addresses [Array(Hash)] The grouped concrete addresses.
+					# @returns [Endpoint] The immutable endpoint.
+					def self.build(name:, scheme:, protocols:, addresses:)
+						new(
+							name.to_s,
+							scheme.to_sym,
+							protocols.map(&:to_s).uniq,
+							addresses,
+						).tap(&:freeze)
+					end
+
 					# Wrap serialized endpoint state.
 					# @parameter value [Endpoint | Hash] The value to wrap.
 					# @returns [Endpoint] The endpoint value.
@@ -17,39 +32,26 @@ def self.wrap(value)
 						when self
 							value
 						when Hash
-							new(**value)
+							build(**value)
 						else
 							raise ArgumentError, "Invalid Envoy endpoint: #{value.inspect}"
 						end
 					end
 
-					# Normalize a concrete endpoint address.
-					# @parameter value [Hash] The address value.
-					# @returns [Hash] A normalized IP or Unix address.
-					def self.normalize_address(value)
-						if path = value[:path]
-							raise ArgumentError, "A Unix endpoint cannot specify an IP address or port!" if value[:address] || value[:port]
-
-							{path: path.to_s}.freeze
-						elsif value[:address] && value[:port]
-							{address: value[:address].to_s, port: Integer(value[:port])}.freeze
-						else
-							raise ArgumentError, "An endpoint address requires either path, or address and port: #{value.inspect}"
-						end
-					end
-
 					# Initialize an endpoint.
 					# @parameter name [String] The upstream cluster name.
-					# @parameter scheme [String | Symbol] The upstream application scheme.
+					# @parameter scheme [Symbol] The upstream application scheme.
 					# @parameter protocols [Array(String)] The supported upstream HTTP protocol names.
 					# @parameter addresses [Array(Hash)] The grouped concrete addresses.
-					def initialize(name:, scheme:, protocols:, addresses:)
-						@name = name.to_s
-						@scheme = scheme.to_sym
-						@protocols = protocols.map(&:to_s).uniq.freeze
+					def initialize(name, scheme, protocols, addresses)
+						@name = name
+						@scheme = scheme
+						@protocols = protocols
 						raise ArgumentError, "An endpoint requires at least one protocol!" if @protocols.empty?
-						@addresses = addresses.map{|value| self.class.normalize_address(value)}.freeze
+						@addresses = addresses
 						raise ArgumentError, "An endpoint requires at least one address!" if @addresses.empty?
+
+						@hash = nil
 					end
 
 					# @attribute [String] The upstream cluster name.
@@ -64,6 +66,39 @@ def initialize(name:, scheme:, protocols:, addresses:)
 					# @attribute [Array(Hash)] The grouped concrete addresses.
 					attr :addresses
 
+					# Freeze this endpoint and cache its value hash.
+					def freeze
+						return self if frozen?
+
+						@name.freeze
+						@protocols.each(&:freeze).freeze
+						@addresses.each do |address|
+							address.each_value(&:freeze)
+							address.freeze
+						end.freeze
+						@hash = self.hash
+
+						super
+					end
+
+					# Compare this endpoint with another endpoint by value.
+					# @parameter other [Object] The object to compare.
+					# @returns [Boolean] Whether both endpoints have identical values.
+					def eql?(other)
+						other.instance_of?(self.class) &&
+							@name == other.name &&
+							@scheme == other.scheme &&
+							@protocols == other.protocols &&
+							@addresses == other.addresses
+					end
+
+					alias == eql?
+
+					# Compute the value hash used when grouping endpoints.
+					# @returns [Integer] The endpoint value hash.
+					def hash
+						@hash || [self.class, @name, @scheme, @protocols, @addresses].hash
+					end
 				end
 			end
 		end

lib/async/service/supervisor/envoy/endpoint_group.rb

@@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+module Async
+	module Service
+		module Supervisor
+			module Envoy
+				# Groups the workers reporting a single upstream endpoint.
+				class EndpointGroup
+					# Initialize an endpoint group.
+					# @parameter endpoint [Endpoint] The endpoint shared by the workers.
+					def initialize(endpoint)
+						@endpoint = endpoint
+						@workers = {}
+					end
+
+					# @attribute [Endpoint] The endpoint shared by the workers.
+					attr :endpoint
+
+					# Add or update a worker report.
+					# @parameter worker [SupervisorController] The worker reporting the endpoint.
+					# @parameter healthy [Boolean] Whether the worker considers the endpoint healthy.
+					def add(worker, healthy:)
+						@workers[worker.id] = healthy
+					end
+
+					# Remove a worker report.
+					# @parameter worker [SupervisorController] The worker to remove.
+					# @returns [Boolean | Nil] The removed health value, if present.
+					def remove(worker)
+						@workers.delete(worker.id)
+					end
+
+					# Count the workers reporting this endpoint.
+					# @returns [Integer] The number of reporting workers.
+					def size
+						@workers.size
+					end
+
+					# Determine whether any workers report this endpoint.
+					# @returns [Boolean] Whether the group has no workers.
+					def empty?
+						@workers.empty?
+					end
+
+					# Determine the aggregate endpoint health.
+					# @returns [Boolean] Whether any reporting worker is healthy.
+					def healthy?
+						@workers.each_value.any?
+					end
+
+					# Convert the group to an xDS endpoint description.
+					# @returns [Hash] The endpoint addresses and aggregate health.
+					def as_json
+						{addresses: @endpoint.addresses, healthy: healthy?}
+					end
+				end
+			end
+		end
+	end
+end

lib/async/service/supervisor/envoy/monitor.rb

@@ -10,6 +10,7 @@
 
 require_relative "delegate"
 require_relative "endpoint"
+require_relative "endpoint_group"
 
 module Async
 	module Service
@@ -108,6 +109,7 @@ def build_record(supervisor_controller, endpoint)
 						{
 							cluster: cluster.to_s,
 							endpoint: endpoint,
+							worker: supervisor_controller,
 							healthy: @delegate.healthy?(supervisor_controller, endpoint),
 						}
 					end
@@ -146,9 +148,14 @@ def build_records_by_cluster
 
 					def build_clusters(records_by_cluster = build_records_by_cluster)
 						records_by_cluster.transform_values do |records|
-							records.map do |record|
-								{addresses: record[:endpoint].addresses, healthy: record[:healthy]}
+							groups = {}
+
+							records.each do |record|
+								group = groups[record[:endpoint]] ||= EndpointGroup.new(record[:endpoint])
+								group.add(record[:worker], healthy: record[:healthy])
 							end
+
+							groups.each_value.map(&:as_json)
 						end
 					end
 

lib/async/service/supervisor/envoy/supervised.rb

@@ -9,7 +9,7 @@ module Async
 	module Service
 		module Supervisor
 			module Envoy
-				# Registers post-bind cluster listeners with the supervisor.
+				# Registers bound Falcon listeners with the supervisor.
 				module Supervised
 					include Async::Service::Supervisor::Supervised
 

readme.md

@@ -24,6 +24,10 @@ Please see the [project documentation](https://socketry.github.io/async-service-
 
 Please see the [project releases](https://socketry.github.io/async-service-supervisor-envoy/releases/index) for all releases.
 
+### Unreleased
+
+  - Deduplicate shared listener endpoints reported by multiple supervised workers.
+
 ### v0.0.1
 
   - Register concrete Falcon cluster listeners as Envoy upstream endpoint state after binding.

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Deduplicate immutable endpoint values reported by multiple supervised workers and aggregate their health.
+
 ## v0.0.1
 
   - Register concrete Falcon cluster listeners as Envoy upstream endpoint state after binding.

test/async/service/supervisor/envoy/endpoint.rb

@@ -0,0 +1,75 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/service/supervisor/envoy/endpoint"
+
+describe Async::Service::Supervisor::Envoy::Endpoint do
+	let(:attributes) do
+		{
+			name: "api",
+			scheme: "http",
+			protocols: ["h2"],
+			addresses: [{path: "/tmp/api.ipc"}],
+		}
+	end
+
+	it "wraps endpoint values" do
+		endpoint = subject.wrap(attributes)
+
+		expect(endpoint.name).to be == "api"
+		expect(endpoint.scheme).to be == :http
+		expect(endpoint.protocols).to be == ["h2"]
+		expect(endpoint.addresses).to be == [{path: "/tmp/api.ipc"}]
+	end
+
+	it "returns endpoint instances unchanged" do
+		endpoint = subject.build(**attributes)
+
+		expect(subject.wrap(endpoint)).to be_equal(endpoint)
+	end
+
+	it "has immutable value semantics" do
+		endpoint = subject.build(**attributes)
+		equivalent = subject.build(**attributes)
+
+		expect(endpoint).to be == equivalent
+		expect(endpoint.eql?(equivalent)).to be == true
+		expect(endpoint.hash).to be == equivalent.hash
+		expect({endpoint => true}[equivalent]).to be == true
+
+		expect(endpoint.frozen?).to be == true
+		expect(endpoint.name.frozen?).to be == true
+		expect(endpoint.protocols.frozen?).to be == true
+		expect(endpoint.protocols.first.frozen?).to be == true
+		expect(endpoint.addresses.frozen?).to be == true
+		expect(endpoint.addresses.first.frozen?).to be == true
+		expect(endpoint.addresses.first[:path].frozen?).to be == true
+		expect(endpoint.freeze).to be_equal(endpoint)
+	end
+
+	it "preserves address order in endpoint identity" do
+		addresses = [
+			{address: "127.0.0.1", port: 9292},
+			{path: "/tmp/api.ipc"},
+		]
+
+		endpoint = subject.build(**attributes, addresses: addresses)
+		reordered = subject.build(**attributes, addresses: addresses.reverse)
+
+		expect(endpoint).not.to be == reordered
+	end
+
+	it "rejects invalid endpoint objects" do
+		expect do
+			subject.wrap(Object.new)
+		end.to raise_exception(ArgumentError)
+	end
+
+	it "rejects endpoints without protocols" do
+		expect do
+			subject.build(**attributes, protocols: [])
+		end.to raise_exception(ArgumentError)
+	end
+end