diff --git a/examples/cluster/Dockerfile b/examples/cluster/Dockerfile new file mode 100644 index 00000000..047886d7 --- /dev/null +++ b/examples/cluster/Dockerfile @@ -0,0 +1,12 @@ +ARG RUBY_VERSION=4.0 +FROM ruby:${RUBY_VERSION} + +WORKDIR /code + +ENV BUNDLE_GEMFILE=/code/gems.rb + +COPY . . + +RUN bundle install + +CMD ["bundle", "exec", "async-service", "falcon.rb"] diff --git a/examples/cluster/client.rb b/examples/cluster/client.rb new file mode 100644 index 00000000..08e89b8f --- /dev/null +++ b/examples/cluster/client.rb @@ -0,0 +1,32 @@ +#!/usr/bin/env ruby +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/http/internet/instance" + +url = ENV.fetch("ENVOY_URI", "http://127.0.0.1:10000") +deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 20 +workers = {} + +Sync do + until workers.size == 2 || Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline + begin + Async::HTTP::Internet.get(url) do |response| + if response.success? + worker_id = response.headers["x-worker-id"] + workers[worker_id] ||= response.read + end + end + rescue Errno::ECONNREFUSED, EOFError + # Envoy may still be connecting to the xDS control plane. + end + + sleep(0.1) unless workers.size == 2 + end +end + +abort "Envoy did not route requests to both workers." unless workers.size == 2 + +workers.each_value{|body| puts(body)} diff --git a/examples/cluster/compose.yaml b/examples/cluster/compose.yaml new file mode 100644 index 00000000..8c1ee2bb --- /dev/null +++ b/examples/cluster/compose.yaml @@ -0,0 +1,31 @@ +services: + falcon: + image: cluster-falcon + build: + context: . + environment: + CONSOLE_OUTPUT: XTerm + ports: + - "10000:10000" + + envoy: + image: envoyproxy/envoy:v1.32-latest + command: ["envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "info"] + network_mode: "service:falcon" + volumes: + - ./envoy.yaml:/etc/envoy/envoy.yaml:ro + depends_on: + falcon: + condition: service_started + + client: + image: cluster-falcon + command: ["bundle", "exec", "ruby", "client.rb"] + environment: + ENVOY_URI: http://127.0.0.1:10000 + network_mode: "service:falcon" + depends_on: + envoy: + condition: service_started + profiles: + - client diff --git a/examples/cluster/envoy.yaml b/examples/cluster/envoy.yaml new file mode 100644 index 00000000..fbdd9bb5 --- /dev/null +++ b/examples/cluster/envoy.yaml @@ -0,0 +1,68 @@ +node: + id: falcon-cluster-example + cluster: falcon-cluster-example + +dynamic_resources: + ads_config: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + listeners: + - name: listener_http + address: + socket_address: + address: 0.0.0.0 + port_value: 10000 + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + route_config: + name: local_route + virtual_hosts: + - name: falcon + domains: ["*"] + routes: + - match: + prefix: "/" + route: + cluster: cluster + http_filters: + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + - name: cluster + connect_timeout: 1s + type: EDS + lb_policy: ROUND_ROBIN + eds_cluster_config: + service_name: cluster + eds_config: + ads: {} + resource_api_version: V3 + + - name: xds_cluster + connect_timeout: 1s + type: STATIC + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} diff --git a/examples/cluster/falcon.rb b/examples/cluster/falcon.rb new file mode 100644 index 00000000..f309abf2 --- /dev/null +++ b/examples/cluster/falcon.rb @@ -0,0 +1,51 @@ +#!/usr/bin/env async-service +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/service/supervisor" +require "async/service/supervisor/envoy" +require "falcon/environment/cluster" + +service "cluster" do + include Falcon::Environment::Cluster + include Async::Service::Supervisor::Envoy::Supervised + + count 2 + + def url + "http://localhost:0" + end + + middleware do + rack_application = proc do |_env| + worker_id = Process.pid.to_s + body = "Hello from worker #{worker_id}!\n" + + [ + 200, + { + "content-type" => "text/plain", + "content-length" => body.bytesize.to_s, + "x-worker-id" => worker_id, + }, + [body], + ] + end + + Falcon::Server.middleware(rack_application, cache: false) + end +end + +service "supervisor" do + include Async::Service::Supervisor::Environment + + monitors do + [ + Async::Service::Supervisor::Envoy::Monitor.new( + bind: "http://127.0.0.1:18000", + ), + ] + end +end diff --git a/examples/cluster/gems.rb b/examples/cluster/gems.rb new file mode 100644 index 00000000..99856c34 --- /dev/null +++ b/examples/cluster/gems.rb @@ -0,0 +1,9 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +source "https://rubygems.org" + +gem "falcon", "~> 0.56.0" +gem "async-service-supervisor-envoy", "~> 0.2.0" diff --git a/examples/cluster/readme.md b/examples/cluster/readme.md new file mode 100644 index 00000000..def1761a --- /dev/null +++ b/examples/cluster/readme.md @@ -0,0 +1,33 @@ +# Cluster with Envoy + +This example runs a two-worker Falcon cluster behind Envoy using Docker Compose. Falcon publishes each worker's dynamically bound endpoint to Envoy through the supervisor's xDS control plane. + +See the [Dynamic Clusters with Envoy](../../guides/cluster-deployment/readme.md) guide for the architecture, endpoint registration lifecycle, and deployment considerations. + +## Requirements + +- Docker with Compose support. + +## Usage + +From this directory, build and start Falcon and Envoy: + +```shell +$ docker compose up --build --detach +``` + +Run the client through Compose: + +```shell +$ docker compose run --rm client +Hello from worker 12! +Hello from worker 13! +``` + +The client waits for Envoy and confirms that requests reach both workers. + +Stop and remove the containers: + +```shell +$ docker compose down +``` diff --git a/falcon.gemspec b/falcon.gemspec index a19cce02..6713f3b9 100644 --- a/falcon.gemspec +++ b/falcon.gemspec @@ -28,7 +28,7 @@ Gem::Specification.new do |spec| spec.add_dependency "async" spec.add_dependency "async-container", "~> 0.20" - spec.add_dependency "async-http", "~> 0.75" + spec.add_dependency "async-http", "~> 0.97" spec.add_dependency "async-http-cache", "~> 0.4" spec.add_dependency "async-service", "~> 0.19" spec.add_dependency "async-utilization", "~> 0.3" diff --git a/gems.rb b/gems.rb index 3b465060..c0c71ffa 100644 --- a/gems.rb +++ b/gems.rb @@ -25,7 +25,7 @@ # gem "fiber-profiler" -gem "async-service-supervisor" +gem "async-service-supervisor", "~> 0.18" group :maintenance, optional: true do gem "bake-modernize" diff --git a/guides/cluster-deployment/readme.md b/guides/cluster-deployment/readme.md new file mode 100644 index 00000000..d085aad8 --- /dev/null +++ b/guides/cluster-deployment/readme.md @@ -0,0 +1,197 @@ +# Dynamic Clusters with Envoy + +This guide explains how to run Falcon workers with independently bound endpoints and publish them dynamically to Envoy using xDS. + +## When to Use a Cluster + +A regular {ruby Falcon::Service::Server} binds one listener and shares it with every worker. This is the simplest design when Falcon accepts connections directly or sits behind a load balancer that targets one stable address. + +{ruby Falcon::Service::Cluster} instead gives each worker its own listener. Use it when an external load balancer needs to address, monitor, and remove workers individually. Because workers may bind ephemeral ports or Unix-domain sockets, the load balancer needs a dynamic source of viable endpoints rather than a static address list. + +| Service | Listener ownership | Upstream discovery | +| --- | --- | --- | +| `Falcon::Service::Server` | One listener shared by all workers | One stable address | +| `Falcon::Service::Cluster` | One listener per worker | Dynamic worker endpoints | + +## Architecture + +Each cluster worker can bind to `localhost` with port `0`, allowing the operating system to assign an available port. Falcon describes the bound resource with a {ruby Falcon::Listener}, including its name, scheme, supported protocols, and concrete addresses. + +The worker registers that listener with `async-service-supervisor-envoy`. The supervisor publishes the current workers through an xDS control plane, and Envoy uses Endpoint Discovery Service (EDS) updates to maintain the upstream cluster. + +Requests arrive at Envoy's stable listener. Envoy selects one of the discovered worker endpoints and forwards the request to it: + +```mermaid +flowchart LR + Client[Client] -->|HTTP on port 10000| Envoy + + subgraph Network[Shared network namespace] + Envoy[Envoy] + + subgraph Falcon[Falcon container] + Supervisor[Supervisor and xDS control plane] + Worker1[Falcon worker 1] + Worker2[Falcon worker 2] + end + + Worker1 -.->|Register endpoint| Supervisor + Worker2 -.->|Register endpoint| Supervisor + Supervisor -.->|EDS over ADS on port 18000| Envoy + Envoy -->|HTTP on dynamic port| Worker1 + Envoy -->|HTTP on dynamic port| Worker2 + end +``` + +## Configuration + +Add Falcon and the Envoy supervisor integration to your `gems.rb`: + +```ruby +gem "falcon", "~> 0.56.0" +gem "async-service-supervisor-envoy", "~> 0.2" +``` + +Define a Falcon cluster service and an accompanying supervisor in `falcon.rb`: + +```ruby +#!/usr/bin/env async-service +# frozen_string_literal: true + +require "async/service/supervisor" +require "async/service/supervisor/envoy" +require "falcon/environment/cluster" + +service "cluster" do + include Falcon::Environment::Cluster + include Async::Service::Supervisor::Envoy::Supervised + + count 2 + + def url + "http://localhost:0" + end + + middleware do + application = proc do |_env| + body = "Hello from worker #{Process.pid}!\n" + + [200, { + "content-type" => "text/plain", + "content-length" => body.bytesize.to_s, + }, [body]] + end + + Falcon::Server.middleware(application, cache: false) + end +end + +service "supervisor" do + include Async::Service::Supervisor::Environment + + monitors do + [ + Async::Service::Supervisor::Envoy::Monitor.new( + bind: "http://127.0.0.1:18000", + ), + ] + end +end +``` + +The Falcon service name becomes the listener name, so the corresponding Envoy EDS cluster uses `cluster` as its service name. Configure Envoy to receive aggregated discovery updates from the supervisor: + +```yaml +node: + id: falcon-cluster + cluster: falcon-cluster + +dynamic_resources: + ads_config: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + listeners: + - name: listener_http + address: + socket_address: + address: 0.0.0.0 + port_value: 10000 + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + route_config: + name: local_route + virtual_hosts: + - name: falcon + domains: ["*"] + routes: + - match: + prefix: "/" + route: + cluster: cluster + http_filters: + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + - name: cluster + connect_timeout: 1s + type: EDS + lb_policy: ROUND_ROBIN + eds_cluster_config: + service_name: cluster + eds_config: + ads: {} + resource_api_version: V3 + + - name: xds_cluster + connect_timeout: 1s + type: STATIC + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} +``` + +The `xds_cluster` connection uses HTTP/2 because ADS is served over gRPC. + +## Worker Registration + +When each worker starts: + +1. Falcon binds the worker to an available loopback port. +2. The worker registers its concrete addresses and supported protocols with the supervisor. +3. The supervisor's Envoy monitor publishes the current worker endpoints as an EDS resource. +4. Envoy receives the resource over its Aggregated Discovery Service (ADS) connection and updates its upstream cluster. + +The listener preserves all addresses returned by the bound endpoint. This allows the same interface to describe IP sockets, Unix-domain sockets, and endpoints with additional addresses. + +## Worker Restarts + +If a worker exits, its supervisor connection closes and the monitor publishes an EDS update without that endpoint. Falcon restarts the worker, which binds a new available port and registers it. The monitor then publishes another update, and Envoy receives both changes over its existing ADS stream without polling or restarting. + +This lifecycle is important when ports are ephemeral or a directory may contain stale Unix-domain socket paths: consumers should use the supervisor's current endpoint state as the source of truth. + +## Network Topology + +Falcon and Envoy can run in the same network namespace, allowing workers to bind to loopback addresses while remaining reachable by Envoy. With Docker Compose, `network_mode: service:falcon` gives the Envoy service access to Falcon's network namespace, so `127.0.0.1` and `localhost` refer to the same loopback interface for both processes. + +Without a shared network namespace, Envoy cannot connect to worker endpoints bound to Falcon's loopback interface. In a different deployment topology, bind workers to an interface that Envoy can reach and apply the appropriate network access controls. diff --git a/guides/links.yaml b/guides/links.yaml index e5beacd7..07d8ce32 100644 --- a/guides/links.yaml +++ b/guides/links.yaml @@ -4,11 +4,13 @@ rails-integration: order: 2 deployment: order: 3 -performance-tuning: +cluster-deployment: order: 4 -websockets: +performance-tuning: order: 5 -interim-responses: +websockets: order: 6 +interim-responses: + order: 7 how-it-works: order: 10 diff --git a/lib/falcon.rb b/lib/falcon.rb index 44ca3579..0b7b9e16 100644 --- a/lib/falcon.rb +++ b/lib/falcon.rb @@ -4,6 +4,7 @@ # Copyright, 2017-2026, by Samuel Williams. require_relative "falcon/version" +require_relative "falcon/listener" require_relative "falcon/server" require_relative "falcon/composite_server" diff --git a/lib/falcon/environment/cluster.rb b/lib/falcon/environment/cluster.rb new file mode 100644 index 00000000..377bb932 --- /dev/null +++ b/lib/falcon/environment/cluster.rb @@ -0,0 +1,27 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require_relative "server" +require_relative "../service/cluster" + +module Falcon + module Environment + # Provides an environment for hosting a cluster of Falcon server workers, where each worker binds its own endpoint. + module Cluster + include Server + + # The service class to use for the cluster. + # @returns [Class] + def service_class + Service::Cluster + end + + # The host that this server will receive connections for. + def url + "http://[::]:0" + end + end + end +end diff --git a/lib/falcon/environment/server.rb b/lib/falcon/environment/server.rb index 628a8393..a4640bdf 100644 --- a/lib/falcon/environment/server.rb +++ b/lib/falcon/environment/server.rb @@ -67,6 +67,14 @@ def client_endpoint ::Async::HTTP::Endpoint.parse(url) end + # Prepare a server worker after its listener has been bound. + # + # @parameter instance [Object] The container instance. + # @parameter listener [Falcon::Listener] The worker's bound listener. + def prepare_worker!(instance, listener) + prepare!(instance) + end + # Make a server instance using the given endpoint. The endpoint may be a bound endpoint, so we take care to specify the protocol and scheme as per the original endpoint. # # @parameter endpoint [IO::Endpoint] The endpoint to bind to. diff --git a/lib/falcon/listener.rb b/lib/falcon/listener.rb new file mode 100644 index 00000000..12dfdc8a --- /dev/null +++ b/lib/falcon/listener.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +module Falcon + # Describes a bound listener for a Falcon server. + class Listener + # Initialize a bound listener. + # @parameter name [String] The logical listener name. + # @parameter scheme [String] The application protocol scheme. + # @parameter protocols [Array(String)] The supported application protocol names. + # @parameter endpoint [IO::Endpoint::BoundEndpoint] The bound endpoint. + def initialize(name:, scheme:, protocols:, endpoint:) + @name = name + @scheme = scheme + @protocols = protocols.map(&:to_s).freeze + @endpoint = endpoint + @addresses = endpoint.sockets.map{|socket| socket.to_io.local_address}.freeze + freeze + end + + # @attribute [String] The logical listener name. + attr_reader :name + + # @attribute [String] The application protocol scheme. + attr_reader :scheme + + # @attribute [Array(String)] The supported application protocol names. + attr_reader :protocols + + # @attribute [IO::Endpoint::BoundEndpoint] The bound endpoint. + attr_reader :endpoint + + # @attribute [Array(Addrinfo)] The bound addresses. + attr_reader :addresses + + # Close the bound endpoint. + def close + @endpoint.close + end + end +end diff --git a/lib/falcon/service/cluster.rb b/lib/falcon/service/cluster.rb new file mode 100644 index 00000000..52dc4318 --- /dev/null +++ b/lib/falcon/service/cluster.rb @@ -0,0 +1,32 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require_relative "server" + +module Falcon + # @namespace + module Service + # A managed service for running Falcon workers with independently bound endpoints. + class Cluster < Server + # Cluster workers bind independently in their own process. + def bind_endpoint + end + + # Bind and yield a listener owned by a cluster worker. + # @parameter evaluator [Environment::Evaluator] The environment evaluator. + # @yields {|listener| ...} The listener owned by the worker. + # @parameter listener [Falcon::Listener] The bound listener. + def with_listener(evaluator) + endpoint = evaluator.endpoint + bound_endpoint = endpoint.bound + listener = make_listener(evaluator, endpoint, bound_endpoint) + + yield listener + ensure + bound_endpoint&.close + end + end + end +end diff --git a/lib/falcon/service/server.rb b/lib/falcon/service/server.rb index c3061bc6..f9179c51 100644 --- a/lib/falcon/service/server.rb +++ b/lib/falcon/service/server.rb @@ -7,6 +7,7 @@ require "async/service/managed/service" require "async/http/endpoint" +require_relative "../listener" require_relative "../server" module Falcon @@ -18,35 +19,99 @@ class Server < Async::Service::Managed::Service def initialize(...) super - @bound_endpoint = nil + @listener = nil end - # Prepare the bound endpoint for the server. - def start + # Build a listener from a configured and bound endpoint. + # @parameter evaluator [Environment::Evaluator] The environment evaluator. + # @parameter endpoint [Async::HTTP::Endpoint] The configured endpoint. + # @parameter bound_endpoint [IO::Endpoint::BoundEndpoint] The bound endpoint. + # @returns [Falcon::Listener] The bound listener. + def make_listener(evaluator, endpoint, bound_endpoint) + Listener.new( + name: evaluator.name, + scheme: endpoint.scheme, + protocols: endpoint.protocol.names, + endpoint: bound_endpoint, + ) + end + + # Bind the endpoint used by each server worker. + def bind_endpoint @endpoint = @evaluator.endpoint Sync do - @bound_endpoint = @endpoint.bound + bound_endpoint = @endpoint.bound + @listener = make_listener(@evaluator, @endpoint, bound_endpoint) end Console.info(self){"Starting #{self.name} on #{@endpoint}"} + end + + # Prepare the bound endpoint for the server. + def start + bind_endpoint super end + # Yield the listener used by a server worker. + # @parameter evaluator [Environment::Evaluator] The environment evaluator. + # @yields {|listener| ...} The listener used by the worker. + # @parameter listener [Falcon::Listener] The bound listener. + def with_listener(evaluator) + yield @listener + end + + # Setup the service into the specified container. + # @parameter container [Async::Container] The container to configure. + def setup(container) + container_options = @evaluator.container_options + health_check_timeout = container_options[:health_check_timeout] + + container.run(**container_options) do |instance| + clock = Async::Clock.start + evaluator = self.environment.evaluator + + with_listener(evaluator) do |listener| + Async do + server = nil + + health_checker(instance, health_check_timeout) do + if server + instance.name = format_title(evaluator, server) + end + end + + instance.status!("Preparing...") + evaluator.prepare_worker!(instance, listener) + emit_prepared(instance, clock) + + instance.status!("Running...") + server = run(instance, evaluator, listener) + instance.name = format_title(evaluator, server) + emit_running(instance, clock) + + instance.ready! + end + end + end + end + # Run the service logic. # # @parameter instance [Object] The container instance. # @parameter evaluator [Environment::Evaluator] The environment evaluator. + # @parameter listener [Falcon::Listener] The listener used by this worker. # @returns [Falcon::Server] The server instance. - def run(instance, evaluator) + def run(instance, evaluator, listener = @listener) if evaluator.respond_to?(:make_supervised_worker) Console.warn(self, "Async::Container::Supervisor is replaced by Async::Service::Supervisor, please update your service definition.") evaluator.make_supervised_worker(instance).run end - server = evaluator.make_server(@bound_endpoint) + server = evaluator.make_server(listener.endpoint) Async do |task| server.run @@ -69,9 +134,9 @@ def run(instance, evaluator) # Close the bound endpoint. def stop(...) - if @bound_endpoint - @bound_endpoint.close - @bound_endpoint = nil + if @listener + @listener.close + @listener = nil end @endpoint = nil diff --git a/readme.md b/readme.md index 239d9353..ad0f4dec 100644 --- a/readme.md +++ b/readme.md @@ -47,6 +47,11 @@ Please see the [project documentation](https://socketry.github.io/falcon/) for m Please see the [project releases](https://socketry.github.io/falcon/releases/index) for all releases. +### Unreleased + + - Add `Falcon::Environment::Cluster` and `Falcon::Service::Cluster` for running workers with independently bound endpoints. + - Add `Falcon::Listener` to describe bound listeners shared by regular server workers or owned by cluster workers. + ### v0.55.6 - Move Falcon middleware trace providers to `traces/provider/falcon/middleware`. @@ -87,10 +92,6 @@ Please see the [project releases](https://socketry.github.io/falcon/releases/ind - Introduce `Falcon::CompositeServer` for hosting multiple server instances in a single worker. -### v0.52.4 - - - Relax dependency on `async-container-supervisor` to allow `~> 0.6`. - ## Contributing We welcome contributions to this project. diff --git a/releases.md b/releases.md index f77bd436..d8941252 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,10 @@ # Releases +## Unreleased + + - Add `Falcon::Environment::Cluster` and `Falcon::Service::Cluster` for running workers with independently bound endpoints. + - Add `Falcon::Listener` to describe bound listeners shared by regular server workers or owned by cluster workers. + ## v0.55.6 - Move Falcon middleware trace providers to `traces/provider/falcon/middleware`. diff --git a/test/falcon/environment/cluster.rb b/test/falcon/environment/cluster.rb new file mode 100644 index 00000000..95afe789 --- /dev/null +++ b/test/falcon/environment/cluster.rb @@ -0,0 +1,30 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "falcon/environment/cluster" +require "async/service/environment" + +describe Falcon::Environment::Cluster do + let(:evaluator) do + Async::Service::Environment.build(subject, name: "localhost").evaluator + end + + it "provides default cluster settings" do + expect(evaluator).to have_attributes( + service_class: be == Falcon::Service::Cluster, + url: be == "http://[::]:0", + authority: be == "localhost", + ) + end + + it "prepares workers using the existing preparation hook" do + instance = Object.new + listener = Object.new + + expect(evaluator).to receive(:prepare!).with(instance) + + evaluator.prepare_worker!(instance, listener) + end +end diff --git a/test/falcon/environment/server.rb b/test/falcon/environment/server.rb index 07d7b9b1..fb83fb4d 100644 --- a/test/falcon/environment/server.rb +++ b/test/falcon/environment/server.rb @@ -21,4 +21,13 @@ ) expect(evaluator.client_endpoint).to be_a(Async::HTTP::Endpoint) end + + it "prepares workers using the existing preparation hook" do + instance = Object.new + listener = Object.new + + expect(evaluator).to receive(:prepare!).with(instance) + + evaluator.prepare_worker!(instance, listener) + end end diff --git a/test/falcon/listener.rb b/test/falcon/listener.rb new file mode 100644 index 00000000..13bfdf55 --- /dev/null +++ b/test/falcon/listener.rb @@ -0,0 +1,48 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "falcon/listener" + +describe Falcon::Listener do + def make_listener(*addresses, name: "hello", scheme: "http", protocols: ["http/1.1", "http/1.0"]) + sockets = addresses.map do |address| + io = Struct.new(:local_address).new(address) + Struct.new(:to_io).new(io) + end + + endpoint = Struct.new(:sockets) do + attr_reader :closed + + def close + @closed = true + end + end.new(sockets) + subject.new(name: name, scheme: scheme, protocols: protocols, endpoint: endpoint) + end + + it "describes a bound listener" do + ip_address = Addrinfo.tcp("127.0.0.1", 9292) + unix_address = Addrinfo.unix("/tmp/falcon.sock") + listener = make_listener(ip_address, unix_address) + + expect(listener).to have_attributes( + name: be == "hello", + scheme: be == "http", + protocols: be == ["http/1.1", "http/1.0"], + addresses: be == [ip_address, unix_address], + frozen?: be == true, + ) + expect(listener.addresses.frozen?).to be == true + expect(listener.protocols.frozen?).to be == true + end + + it "closes the bound endpoint" do + listener = make_listener(Addrinfo.tcp("127.0.0.1", 9292)) + + listener.close + + expect(listener.endpoint.closed).to be == true + end +end diff --git a/test/falcon/service/cluster.rb b/test/falcon/service/cluster.rb new file mode 100644 index 00000000..6fd5d482 --- /dev/null +++ b/test/falcon/service/cluster.rb @@ -0,0 +1,92 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "falcon/service/cluster" +require "falcon/environment/cluster" +require "async/container" +require "async/service/environment" +require "fileutils" +require "net/http" +require "protocol/http/middleware" + +describe Falcon::Service::Cluster do + let(:ports_path) {File.expand_path(".cluster/ports.txt", __dir__)} + + let(:recorder) do + path = ports_path + + Module.new do + def middleware + Protocol::HTTP::Middleware::HelloWorld + end + + def container_options + super.merge(restart: false) + end + + define_method(:prepare_worker!) do |instance, listener| + super(instance, listener) + + File.open(path, "a") do |file| + listener.addresses.each do |address| + file.puts(address.ip_port) if address.ip? + end + end + end + end + end + + let(:environment) do + Async::Service::Environment.new(Falcon::Environment::Cluster).with( + recorder, + name: "hello", + root: File.expand_path(".cluster/hello", __dir__), + url: "http://127.0.0.1:0", + count: 2, + health_check_timeout: 0.01, + ) + end + + let(:server) do + subject.new(environment) + end + + before do + FileUtils.rm_rf(File.dirname(ports_path)) + FileUtils.mkdir_p(File.dirname(ports_path)) + end + + after do + FileUtils.rm_rf(File.dirname(ports_path)) + end + + it "binds a unique port for each worker before preparing the instance" do + container = Async::Container.new + + server.start + server.setup(container) + container.wait_until_ready + + ports = File.readlines(ports_path, chomp: true).map(&:to_i) + + expect(ports).to have_attributes(size: be == 2) + expect(ports).to have_value(be > 0) + expect(ports.uniq).to be == ports + + ports.each do |port| + response = Net::HTTP.get_response(URI("http://127.0.0.1:#{port}/")) + + expect(response).to have_attributes(code: be == "200") + end + + sleep(0.01) + + container.stop + expect(container.failed?).to be_falsey + ensure + server.stop + container&.stop unless container&.stopping? + end +end diff --git a/test/falcon/service/server.rb b/test/falcon/service/server.rb index c04f6aa8..18d72f25 100644 --- a/test/falcon/service/server.rb +++ b/test/falcon/service/server.rb @@ -28,6 +28,22 @@ expect(server).to be_a subject end + it "binds a listener shared by its workers" do + server.start + + server.with_listener(environment.evaluator) do |listener| + expect(listener).to be_a(Falcon::Listener) + expect(listener).to have_attributes( + name: be == "hello", + scheme: be == "http", + protocols: be(:include?, "http/1.1"), + ) + expect(listener.addresses.first.ip?).to be == true + end + ensure + server.stop + end + it "can start and stop server" do container = Async::Container.new @@ -110,4 +126,32 @@ def server.run server.stop end end + + it "propagates server IO errors" do + evaluator = Object.new + bound_endpoint = Object.new + listener = Struct.new(:endpoint).new(bound_endpoint) + failing_server = Object.new + condition = Async::Condition.new + + failing_server.define_singleton_method(:run) do + Async do + condition.wait + raise IOError, "application failure" + end + end + + evaluator.define_singleton_method(:make_server) do |endpoint| + raise ArgumentError, "Unexpected endpoint!" unless endpoint.equal?(bound_endpoint) + failing_server + end + + expect do + Async do |task| + server.run(nil, evaluator, listener) + condition.signal + task.children.each(&:wait) + end.wait + end.to raise_exception(IOError, message: be == "application failure") + end end