From c2d38d16c32fc29a6e364baf22db2cf6b72db5e9 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Sat, 29 Aug 2026 13:08:07 +1200
Subject: [PATCH 1/8] Add configurable request queues

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 context/getting-started.md       |  52 ++++++-
 guides/getting-started/readme.md |  52 ++++++-
 lib/fantail.rb                   |   3 +
 lib/fantail/backend.rb           |  64 +++++----
 lib/fantail/balance.rb           |  45 ++++++
 lib/fantail/configuration.rb     | 185 +++++++++++++++++++++++++
 lib/fantail/proxy.rb             |  25 ++--
 lib/fantail/registry.rb          |  62 ++++++---
 lib/fantail/scheduler.rb         | 176 +++++++++++++++++++++++
 lib/fantail/server.rb            |  13 +-
 readme.md                        |   2 +-
 releases.md                      |   4 +
 test/fantail/configuration.rb    |  65 +++++++++
 test/fantail/fixtures.rb         |   8 +-
 test/fantail/registry.rb         |  17 +++
 test/fantail/scheduler.rb        | 231 +++++++++++++++++++++++++++++++
 16 files changed, 942 insertions(+), 62 deletions(-)
 create mode 100644 lib/fantail/balance.rb
 create mode 100644 lib/fantail/configuration.rb
 create mode 100644 lib/fantail/scheduler.rb
 create mode 100644 test/fantail/configuration.rb
 create mode 100644 test/fantail/scheduler.rb

diff --git a/context/getting-started.md b/context/getting-started.md
index 34581c2..0bfe9f7 100644
--- a/context/getting-started.md
+++ b/context/getting-started.md
@@ -21,9 +21,12 @@ require "fantail"
 require "io/endpoint"
 
 Sync do
+	configuration = Fantail::Configuration.load("config/fantail.rb")
+	
 	server = Fantail::Server.new(
 		Async::HTTP::Endpoint.parse("http://0.0.0.0:9292"),
 		IO::Endpoint.tcp("0.0.0.0", 9293),
+		configuration: configuration,
 	)
 	
 	server.run.wait
@@ -54,6 +57,53 @@ After connecting, the monitor performs a complete replacement. It then publishes
 
 ## Admission Semantics
 
-Each backend has one request-processing slot and a configurable number of response exchanges. The processing slot is released as soon as upstream response headers arrive. The exchange remains reserved until the response body closes.
+Each backend has a configurable number of request-processing permits and response exchanges. A processing permit is released as soon as upstream response headers arrive. The exchange remains reserved until the response body closes.
 
 This allows a worker to begin another request while an earlier response streams, without allowing an unbounded number of streaming responses to accumulate.
+
+The scheduler owns all permits. Request queues can decide which workers are eligible and express a soft preference between them, but cannot reserve capacity independently. If the preferred worker is unavailable, the scheduler remains work-conserving and uses another eligible worker.
+
+## Request Queues
+
+Fantail configuration is trusted application Ruby. The file's final expression must be an immutable `Fantail::Configuration`:
+
+~~~ ruby
+# config/fantail.rb
+Fantail::Configuration.define do |config|
+	config.queue :liquid do |queue|
+		queue.match{|request| request.path.start_with?("/render")}
+		queue.balance :spread
+		queue.depth_limit 500
+		queue.wait_limit 0.25
+		queue.shed status: 429, retry_after: 1
+	end
+	
+	config.queue :grpc do |queue|
+		queue.match do |request|
+			request.headers["content-type"]&.start_with?("application/grpc")
+		end
+		
+		queue.balance :pack, affinity: :grpc
+	end
+	
+	config.default_queue :liquid
+	config.pending_limit 1_000
+	config.permit_limit 1
+end
+~~~
+
+Matchers are evaluated in definition order, followed by the default queue. Across queues, the oldest eligible head request is dispatched first. If that request has no eligible worker, another queue can use the available permit.
+
+The built-in `:spread` policy prefers the least-active worker. The `:pack` policy prefers a worker already processing the specified affinity, while remaining bounded by its permits. An application can supply a policy object implementing `select(backends, queue:, request:)`, and can restrict hard eligibility with `queue.eligible`.
+
+## Load Shedding
+
+`depth_limit` bounds requests actually waiting in a queue; immediately dispatchable requests do not count against it. `pending_limit` provides a global bound across all queues. `wait_limit` bounds actual queue residence time in seconds. Rejected requests use the response configured by `shed`, which defaults to HTTP 429.
+
+Applications can add an admission policy with either a block or an object implementing `admit?(request, queue:, pending:)`:
+
+~~~ ruby
+queue.admit do |request, queue:, pending:|
+	pending < application_limit_for(queue.name)
+end
+~~~
diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md
index 34581c2..0bfe9f7 100644
--- a/guides/getting-started/readme.md
+++ b/guides/getting-started/readme.md
@@ -21,9 +21,12 @@ require "fantail"
 require "io/endpoint"
 
 Sync do
+	configuration = Fantail::Configuration.load("config/fantail.rb")
+	
 	server = Fantail::Server.new(
 		Async::HTTP::Endpoint.parse("http://0.0.0.0:9292"),
 		IO::Endpoint.tcp("0.0.0.0", 9293),
+		configuration: configuration,
 	)
 	
 	server.run.wait
@@ -54,6 +57,53 @@ After connecting, the monitor performs a complete replacement. It then publishes
 
 ## Admission Semantics
 
-Each backend has one request-processing slot and a configurable number of response exchanges. The processing slot is released as soon as upstream response headers arrive. The exchange remains reserved until the response body closes.
+Each backend has a configurable number of request-processing permits and response exchanges. A processing permit is released as soon as upstream response headers arrive. The exchange remains reserved until the response body closes.
 
 This allows a worker to begin another request while an earlier response streams, without allowing an unbounded number of streaming responses to accumulate.
+
+The scheduler owns all permits. Request queues can decide which workers are eligible and express a soft preference between them, but cannot reserve capacity independently. If the preferred worker is unavailable, the scheduler remains work-conserving and uses another eligible worker.
+
+## Request Queues
+
+Fantail configuration is trusted application Ruby. The file's final expression must be an immutable `Fantail::Configuration`:
+
+~~~ ruby
+# config/fantail.rb
+Fantail::Configuration.define do |config|
+	config.queue :liquid do |queue|
+		queue.match{|request| request.path.start_with?("/render")}
+		queue.balance :spread
+		queue.depth_limit 500
+		queue.wait_limit 0.25
+		queue.shed status: 429, retry_after: 1
+	end
+	
+	config.queue :grpc do |queue|
+		queue.match do |request|
+			request.headers["content-type"]&.start_with?("application/grpc")
+		end
+		
+		queue.balance :pack, affinity: :grpc
+	end
+	
+	config.default_queue :liquid
+	config.pending_limit 1_000
+	config.permit_limit 1
+end
+~~~
+
+Matchers are evaluated in definition order, followed by the default queue. Across queues, the oldest eligible head request is dispatched first. If that request has no eligible worker, another queue can use the available permit.
+
+The built-in `:spread` policy prefers the least-active worker. The `:pack` policy prefers a worker already processing the specified affinity, while remaining bounded by its permits. An application can supply a policy object implementing `select(backends, queue:, request:)`, and can restrict hard eligibility with `queue.eligible`.
+
+## Load Shedding
+
+`depth_limit` bounds requests actually waiting in a queue; immediately dispatchable requests do not count against it. `pending_limit` provides a global bound across all queues. `wait_limit` bounds actual queue residence time in seconds. Rejected requests use the response configured by `shed`, which defaults to HTTP 429.
+
+Applications can add an admission policy with either a block or an object implementing `admit?(request, queue:, pending:)`:
+
+~~~ ruby
+queue.admit do |request, queue:, pending:|
+	pending < application_limit_for(queue.name)
+end
+~~~
diff --git a/lib/fantail.rb b/lib/fantail.rb
index 578f0ee..5becbb1 100644
--- a/lib/fantail.rb
+++ b/lib/fantail.rb
@@ -4,10 +4,13 @@
 # Copyright, 2026, by Samuel Williams.
 
 require_relative "fantail/version"
+require_relative "fantail/balance"
+require_relative "fantail/configuration"
 require_relative "fantail/endpoint"
 require_relative "fantail/backend"
 require_relative "fantail/response_body"
 require_relative "fantail/registry"
+require_relative "fantail/scheduler"
 require_relative "fantail/proxy"
 require_relative "fantail/control"
 require_relative "fantail/monitor"
diff --git a/lib/fantail/backend.rb b/lib/fantail/backend.rb
index c783043..f280044 100644
--- a/lib/fantail/backend.rb
+++ b/lib/fantail/backend.rb
@@ -10,20 +10,23 @@ class Backend
 		# @parameter endpoint [Endpoint] The endpoint served by this backend.
 		# @parameter client [Interface(:call, :close)] The HTTP client for the endpoint.
 		# @parameter exchange_limit [Integer] The maximum number of outstanding response exchanges.
+		# @parameter permit_limit [Integer] The maximum number of concurrent processing permits.
 		# @yields {|backend| ...} Invoked when the backend can accept another request.
-		def initialize(endpoint, client, exchange_limit:, &available)
+		def initialize(endpoint, client, exchange_limit:, permit_limit: 1, &available)
 			raise ArgumentError, "Exchange limit must be positive!" unless exchange_limit.positive?
+			raise ArgumentError, "Permit limit must be positive!" unless permit_limit.positive?
 			
 			@endpoint = endpoint
 			@client = client
 			@exchange_limit = exchange_limit
+			@permit_limit = permit_limit
 			@available = available
 			
 			@guard = Thread::Mutex.new
 			@active = true
-			@processing = false
+			@processing = 0
+			@processing_by_queue = Hash.new(0)
 			@exchanges = 0
-			@queued = false
 			@closed = false
 		end
 		
@@ -45,12 +48,11 @@ def start
 		
 		# Reserve the processing slot and one response exchange.
 		# @returns [Boolean] Whether the backend was successfully reserved.
-		def reserve
+		def reserve(queue_name = :default)
 			@guard.synchronize do
-				@queued = false
-				
-				if @active && !@processing && @exchanges < @exchange_limit
-					@processing = true
+				if @active && @processing < @permit_limit && @exchanges < @exchange_limit
+					@processing += 1
+					@processing_by_queue[queue_name] += 1
 					@exchanges += 1
 					return true
 				end
@@ -67,20 +69,18 @@ def call(request)
 		end
 		
 		# Release the request-processing slot after response headers arrive.
-		def processed
+		def processed(queue_name = :default)
 			@guard.synchronize do
-				raise RuntimeError, "Backend is not processing a request!" unless @processing
-				@processing = false
+				release_processing(queue_name)
 			end
 			
 			notify_available
 		end
 		
 		# Release both reservations when a request fails before response headers.
-		def failed
+		def failed(queue_name = :default)
 			close = @guard.synchronize do
-				raise RuntimeError, "Backend is not processing a request!" unless @processing
-				@processing = false
+				release_processing(queue_name)
 				@exchanges -= 1
 				should_close?
 			end
@@ -123,26 +123,40 @@ def exchanges
 		
 		# @returns [Boolean] Whether a request is waiting for response headers.
 		def processing?
+			@guard.synchronize{@processing.positive?}
+		end
+		
+		# @returns [Integer] The number of active processing permits.
+		def processing
 			@guard.synchronize{@processing}
 		end
 		
+		# @returns [Integer] The number of active permits for the given queue affinity.
+		def processing_for(queue_name)
+			@guard.synchronize{@processing_by_queue[queue_name]}
+		end
+		
+		# @returns [Boolean] Whether another request can be admitted.
+		def available?
+			@guard.synchronize{@active && @processing < @permit_limit && @exchanges < @exchange_limit}
+		end
+		
 		protected
 		
 		def notify_available
-			notify = @guard.synchronize do
-				if @active && !@processing && @exchanges < @exchange_limit && !@queued
-					@queued = true
-					true
-				else
-					false
-				end
-			end
-			
-			@available.call(self) if notify
+			@available.call(self) if available?
 		end
 		
 		def should_close?
-			!@active && !@processing && @exchanges.zero? && !@closed
+			!@active && @processing.zero? && @exchanges.zero? && !@closed
+		end
+		
+		def release_processing(queue_name)
+			raise RuntimeError, "Backend is not processing a request!" unless @processing.positive?
+			raise RuntimeError, "Backend is not processing queue #{queue_name.inspect}!" unless @processing_by_queue[queue_name].positive?
+			
+			@processing -= 1
+			@processing_by_queue[queue_name] -= 1
 		end
 		
 		def close_client
diff --git a/lib/fantail/balance.rb b/lib/fantail/balance.rb
new file mode 100644
index 0000000..803fd93
--- /dev/null
+++ b/lib/fantail/balance.rb
@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+module Fantail
+	# Built-in backend selection policies.
+	module Balance
+		# Prefer the backend with the fewest active requests.
+		class Spread
+			def select(backends, queue:, request:)
+				backends.min_by{|backend| [backend.processing, backend.name]}
+			end
+		end
+		
+		# Prefer a backend which is already processing the same class of work.
+		class Pack
+			def initialize(affinity: nil)
+				@affinity = affinity
+			end
+			
+			def select(backends, queue:, request:)
+				affinity = @affinity || queue.name
+				backends.min_by do |backend|
+					[-backend.processing_for(affinity), backend.processing, backend.name]
+				end
+			end
+		end
+		
+		def self.coerce(policy, **options)
+			case policy
+			when :spread
+				Spread.new(**options)
+			when :pack
+				Pack.new(**options)
+			else
+				unless policy.respond_to?(:select)
+					raise ArgumentError, "Balance policy must respond to #select!"
+				end
+				
+				policy
+			end
+		end
+	end
+end
diff --git a/lib/fantail/configuration.rb b/lib/fantail/configuration.rb
new file mode 100644
index 0000000..f3190ba
--- /dev/null
+++ b/lib/fantail/configuration.rb
@@ -0,0 +1,185 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "balance"
+
+module Fantail
+	# Immutable request queue and admission configuration.
+	class Configuration
+		UNDEFINED = Object.new.freeze
+		
+		# Configuration for one class of requests.
+		class Queue
+			def initialize(name)
+				@name = name.to_sym
+				@matcher = nil
+				@eligibility = nil
+				@admission = nil
+				@balance = Balance::Spread.new
+				@depth_limit = nil
+				@wait_limit = nil
+				@shed_status = 429
+				@shed_headers = {}
+			end
+			
+			attr :name
+			attr :balance_policy
+			attr :shed_status
+			attr :shed_headers
+			
+			def match(&block)
+				raise ArgumentError, "A matcher block is required!" unless block
+				@matcher = block
+			end
+			
+			def eligible(&block)
+				raise ArgumentError, "An eligibility block is required!" unless block
+				@eligibility = block
+			end
+			
+			def admit(policy = nil, &block)
+				@admission = policy || block
+				raise ArgumentError, "An admission policy is required!" unless @admission
+			end
+			
+			def balance(policy, **options)
+				@balance = Balance.coerce(policy, **options)
+			end
+			
+			def depth_limit(value = UNDEFINED)
+				return @depth_limit if value.equal?(UNDEFINED)
+				
+				value = Integer(value)
+				raise ArgumentError, "Depth limit must not be negative!" if value.negative?
+				@depth_limit = value
+			end
+			
+			def wait_limit(value = UNDEFINED)
+				return @wait_limit if value.equal?(UNDEFINED)
+				
+				value = Float(value)
+				raise ArgumentError, "Wait limit must be positive!" unless value.positive?
+				@wait_limit = value
+			end
+			
+			def shed(status: 429, retry_after: nil, headers: {})
+				@shed_status = Integer(status)
+				@shed_headers = headers.transform_keys(&:to_s)
+				@shed_headers["retry-after"] = retry_after.to_s if retry_after
+			end
+			
+			def match?(request)
+				@matcher&.call(request)
+			end
+			
+			def eligible?(backend, request)
+				!@eligibility || @eligibility.call(backend, request)
+			end
+			
+			def admit?(request, pending:)
+				return true unless @admission
+				
+				if @admission.respond_to?(:admit?)
+					@admission.admit?(request, queue: self, pending: pending)
+				else
+					@admission.call(request, queue: self, pending: pending)
+				end
+			end
+			
+			def finalize
+				@balance_policy = @balance
+				@shed_headers.freeze
+				freeze
+			end
+		end
+		
+		def self.define
+			configuration = new
+			yield configuration if block_given?
+			configuration.finalize
+		end
+		
+		def self.default
+			@default ||= define do |configuration|
+				configuration.queue(:default)
+				configuration.default_queue(:default)
+			end
+		end
+		
+		# Load trusted application configuration. The final expression must be a Configuration.
+		def self.load(path)
+			path = File.expand_path(path)
+			configuration = TOPLEVEL_BINDING.eval(File.read(path), path)
+			
+			unless configuration.is_a?(self)
+				raise TypeError, "#{path} must return a Fantail::Configuration!"
+			end
+			
+			configuration
+		end
+		
+		def initialize
+			@queues = {}
+			@default_queue = nil
+			@pending_limit = nil
+			@permit_limit = 1
+		end
+		
+		attr :queues
+		attr :default_queue_name
+		
+		# Define a named request queue. Matchers are evaluated in definition order.
+		def queue(name)
+			name = name.to_sym
+			raise ArgumentError, "Queue #{name.inspect} is already defined!" if @queues.key?(name)
+			
+			queue = Queue.new(name)
+			yield queue if block_given?
+			@queues[name] = queue
+			queue
+		end
+		
+		def default_queue(name)
+			@default_queue = name.to_sym
+		end
+		
+		# Set or get the global pending request limit.
+		def pending_limit(value = UNDEFINED)
+			return @pending_limit if value.equal?(UNDEFINED)
+			
+			value = Integer(value)
+			raise ArgumentError, "Pending limit must not be negative!" if value.negative?
+			@pending_limit = value
+		end
+		
+		# Set or get the number of processing permits provided by each worker.
+		def permit_limit(value = UNDEFINED)
+			return @permit_limit if value.equal?(UNDEFINED)
+			
+			value = Integer(value)
+			raise ArgumentError, "Permit limit must be positive!" unless value.positive?
+			@permit_limit = value
+		end
+		
+		def classify(request)
+			@queues.each_value do |queue|
+				return queue if queue.match?(request)
+			end
+			
+			@queues.fetch(@default_queue)
+		end
+		
+		def finalize
+			raise ArgumentError, "At least one queue must be defined!" if @queues.empty?
+			@default_queue ||= @queues.keys.first
+			raise ArgumentError, "Default queue #{@default_queue.inspect} is not defined!" unless @queues.key?(@default_queue)
+			
+			@queues.each_value(&:finalize)
+			@queues.freeze
+			@default_queue_name = @default_queue
+			freeze
+		end
+	end
+end
diff --git a/lib/fantail/proxy.rb b/lib/fantail/proxy.rb
index 76a6c42..d4ffb24 100644
--- a/lib/fantail/proxy.rb
+++ b/lib/fantail/proxy.rb
@@ -7,34 +7,41 @@
 require "protocol/http/response"
 
 require_relative "response_body"
+require_relative "scheduler"
 
 module Fantail
-	# Routes HTTP requests through the registry's global admission queue.
+	# Routes HTTP requests through the configured admission queues.
 	class Proxy
 		# Initialize an HTTP proxy.
 		# @parameter registry [Registry] The backend registry.
-		def initialize(registry)
-			@registry = registry
+		# @parameter configuration [Configuration] Request classification and scheduling policy.
+		def initialize(registry, configuration: Configuration.default)
+			@scheduler = Scheduler.new(registry, configuration)
 		end
 		
+		attr :scheduler
+		
 		# Route a request to the next available backend.
 		# @parameter request [Protocol::HTTP::Request] The downstream request.
 		# @returns [Protocol::HTTP::Response] The upstream or generated response.
 		def call(request)
-			unless backend = @registry.acquire
+			unless backend_reservation = @scheduler.acquire(request)
 				return Protocol::HTTP::Response[503, {"content-type" => "text/plain"}, ["No backends available.\n"]]
 			end
 			
+			return backend_reservation.response if backend_reservation.is_a?(Scheduler::Rejection)
+			
 			reservation = :processing
+			backend = backend_reservation.backend
 			upstream_request = build_request(request)
 			response = backend.call(upstream_request)
-			backend.processed
+			backend_reservation.processed
 			reservation = :exchange
 			
 			if body = response.body
-				response.body = ResponseBody.new(body){backend.release}
+				response.body = ResponseBody.new(body){backend_reservation.release}
 			else
-				backend.release
+				backend_reservation.release
 			end
 			reservation = nil
 			
@@ -42,9 +49,9 @@ def call(request)
 		rescue => error
 			case reservation
 			when :processing
-				backend.failed
+				backend_reservation.failed
 			when :exchange
-				backend.release
+				backend_reservation.release
 			end
 			
 			return Protocol::HTTP::Response[502, {"content-type" => "text/plain"}, ["Bad Gateway: #{error.class}\n"]]
diff --git a/lib/fantail/registry.rb b/lib/fantail/registry.rb
index 8e4d24c..4dce9ca 100644
--- a/lib/fantail/registry.rb
+++ b/lib/fantail/registry.rb
@@ -10,20 +10,22 @@
 require_relative "backend"
 
 module Fantail
-	# Maintains live backends and a global queue of available processing slots.
+	# Maintains live backends and notifies the scheduler when capacity changes.
 	class Registry < Async::Bus::Controller
-		WAKE = Object.new.freeze
-		
 		# Initialize an endpoint registry.
 		# @parameter exchange_limit [Integer] The maximum outstanding responses per backend.
-		# @parameter backend_factory [Proc | Nil] An optional backend construction strategy.
-		def initialize(exchange_limit: 8, backend_factory: nil)
+		# @parameter permit_limit [Integer] The maximum active processing permits per backend.
+		# @parameter backend_factory [#call(endpoint, exchange_limit, permit_limit, available) | Nil] An optional backend construction strategy.
+		def initialize(exchange_limit: 8, permit_limit: 1, backend_factory: nil)
 			@exchange_limit = exchange_limit
+			@permit_limit = permit_limit
 			@backend_factory = backend_factory || self.method(:make_backend)
 			
 			@guard = Thread::Mutex.new
 			@backends = {}
-			@available = Async::Queue.new
+			@notifications = Async::Queue.new
+			@waiting = 0
+			@available = nil
 			@closed = false
 		end
 		
@@ -62,7 +64,7 @@ def update(upserted, removed)
 					next if current&.endpoint == endpoint
 					
 					retired << current if current
-					backend = @backend_factory.call(endpoint, @exchange_limit, self.method(:offer))
+					backend = @backend_factory.call(endpoint, @exchange_limit, @permit_limit, self.method(:offer))
 					@backends[endpoint.name] = backend
 					started << backend
 				end
@@ -70,7 +72,7 @@ def update(upserted, removed)
 			
 			retired.each(&:retire)
 			started.each(&:start)
-			@available.enqueue(WAKE) unless retired.empty?
+			notify_available unless retired.empty?
 			
 			self.size
 		end
@@ -78,15 +80,33 @@ def update(upserted, removed)
 		# Acquire the next backend with processing capacity.
 		# @returns [Backend | Nil] An admitted backend, or nil if no endpoints exist.
 		def acquire
+			waiting = false
+			
 			loop do
 				return nil if self.empty?
+				backend = backends.find{|candidate| candidate.reserve}
+				return backend if backend
 				
-				candidate = @available.dequeue
-				return nil unless candidate
-				next if candidate.equal?(WAKE)
+				unless waiting
+					@guard.synchronize{@waiting += 1}
+					waiting = true
+					next
+				end
 				
-				return candidate if candidate.reserve
+				return nil unless @notifications.dequeue
 			end
+		ensure
+			@guard.synchronize{@waiting -= 1} if waiting
+		end
+		
+		# @returns [Array(Backend)] A snapshot of active backends.
+		def backends
+			@guard.synchronize{@backends.values.dup}
+		end
+		
+		# Register the central scheduler capacity callback.
+		def on_available(&block)
+			@guard.synchronize{@available = block}
 		end
 		
 		# @returns [Integer] The number of active endpoints.
@@ -120,21 +140,27 @@ def close
 				@backends.values.tap{@backends = {}}
 			end
 			
-			@available.close
+			@notifications.close
 			backends.each(&:retire)
 		end
 		
 		protected
 		
-		def offer(backend)
-			@available.enqueue(backend)
+		def offer(_backend)
+			notify_available
+		end
+		
+		def notify_available
+			available, waiting = @guard.synchronize{[@available, @waiting]}
+			@notifications.enqueue(true) if waiting.positive?
+			available&.call
 		rescue Async::Queue::ClosedError
-			# The registry is already shutting down:
+			# The registry is already shutting down.
 		end
 		
-		def make_backend(endpoint, exchange_limit, available)
+		def make_backend(endpoint, exchange_limit, permit_limit, available)
 			client = endpoint.make_client(exchange_limit: exchange_limit)
-			Backend.new(endpoint, client, exchange_limit: exchange_limit, &available)
+			Backend.new(endpoint, client, exchange_limit: exchange_limit, permit_limit: permit_limit, &available)
 		end
 	end
 end
diff --git a/lib/fantail/scheduler.rb b/lib/fantail/scheduler.rb
new file mode 100644
index 0000000..e8578b5
--- /dev/null
+++ b/lib/fantail/scheduler.rb
@@ -0,0 +1,176 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/queue"
+require "protocol/http/response"
+
+require_relative "configuration"
+
+module Fantail
+	# Matches pending requests to concrete backend permits.
+	class Scheduler
+		# A reserved processing permit and response exchange.
+		class Reservation
+			def initialize(backend, queue_name)
+				@backend = backend
+				@queue_name = queue_name
+			end
+			
+			attr :backend
+			
+			def processed
+				@backend.processed(@queue_name)
+			end
+			
+			def failed
+				@backend.failed(@queue_name)
+			end
+			
+			def release
+				@backend.release
+			end
+		end
+		
+		# A queue admission rejection.
+		class Rejection
+			def initialize(queue)
+				@queue = queue
+			end
+			
+			def response
+				headers = {"content-type" => "text/plain"}.merge(@queue.shed_headers)
+				Protocol::HTTP::Response[@queue.shed_status, headers, ["Request queue is full.\n"]]
+			end
+		end
+		
+		Entry = Struct.new(:request, :queue, :enqueued_at, :result, :pending, :assignment)
+		
+		def initialize(registry, configuration = Configuration.default)
+			@registry = registry
+			@configuration = configuration
+			@guard = Thread::Mutex.new
+			@pending = configuration.queues.to_h{|name, queue| [name, []]}
+			@pending_count = 0
+			
+			@registry.on_available{schedule}
+		end
+		
+		# Admit a request, wait for a matching permit, or return a rejection.
+		def acquire(request)
+			queue = @configuration.classify(request)
+			entry = nil
+			result = @guard.synchronize do
+				if reservation = reserve(queue, request)
+					reservation
+				elsif @registry.empty?
+					nil
+				elsif reject?(queue, request)
+					Rejection.new(queue)
+				else
+					entry = Entry.new(request, queue, now, Async::Queue.new, true, nil)
+					@pending.fetch(queue.name) << entry
+					@pending_count += 1
+					schedule_locked
+					entry.assignment
+				end
+			end
+			
+			return result if result
+			return nil unless entry
+			return entry.assignment if entry.assignment
+			
+			if wait_limit = queue.wait_limit
+				remaining = wait_limit - (now - entry.enqueued_at)
+				result = entry.result.dequeue(timeout: remaining) if remaining.positive?
+				return result if result
+				
+				return cancel(entry)
+			else
+				return entry.result.dequeue
+			end
+		end
+		
+		# Try to dispatch pending requests after capacity changes.
+		def schedule
+			@guard.synchronize{schedule_locked}
+		end
+		
+		# @parameter queue_name [Symbol | String | Nil] An optional queue to inspect.
+		# @returns [Integer] The number of requests waiting for a permit.
+		def pending_count(queue_name = nil)
+			@guard.synchronize do
+				if queue_name
+					@pending.fetch(queue_name.to_sym).size
+				else
+					@pending_count
+				end
+			end
+		end
+		
+		protected
+		
+		def now
+			Process.clock_gettime(Process::CLOCK_MONOTONIC)
+		end
+		
+		def reject?(queue, request)
+			return true if @configuration.pending_limit && @pending_count >= @configuration.pending_limit
+			pending = @pending.fetch(queue.name).size
+			return true if queue.depth_limit && pending >= queue.depth_limit
+			return true unless queue.admit?(request, pending: pending)
+			
+			false
+		end
+		
+		def cancel(entry)
+			@guard.synchronize do
+				return entry.assignment unless entry.pending
+				
+				@pending.fetch(entry.queue.name).delete(entry)
+				entry.pending = false
+				@pending_count -= 1
+				Rejection.new(entry.queue)
+			end
+		end
+		
+		def schedule_locked
+			loop do
+				entries = @pending.each_value.filter_map(&:first).sort_by(&:enqueued_at)
+				matched = false
+				
+				entries.each do |entry|
+					if reservation = reserve(entry.queue, entry.request)
+						@pending.fetch(entry.queue.name).shift
+						@pending_count -= 1
+						entry.pending = false
+						entry.assignment = reservation
+						entry.result.enqueue(reservation)
+						matched = true
+						break
+					end
+				end
+				
+				break unless matched
+			end
+		end
+		
+		def reserve(queue, request)
+			backends = @registry.backends.select do |backend|
+				backend.available? && queue.eligible?(backend, request)
+			end
+			
+			until backends.empty?
+				backend = queue.balance_policy.select(backends, queue: queue, request: request)
+				return nil unless backend
+				raise ArgumentError, "Balance policy selected an ineligible backend!" unless backends.include?(backend)
+				
+				return Reservation.new(backend, queue.name) if backend.reserve(queue.name)
+				backends.delete(backend)
+			end
+			
+			nil
+		end
+	end
+end
diff --git a/lib/fantail/server.rb b/lib/fantail/server.rb
index 9ea23f4..66bf4a7 100644
--- a/lib/fantail/server.rb
+++ b/lib/fantail/server.rb
@@ -8,6 +8,7 @@
 require_relative "registry"
 require_relative "proxy"
 require_relative "control"
+require_relative "configuration"
 
 module Fantail
 	# Runs the HTTP load balancer and endpoint-control server together.
@@ -16,9 +17,10 @@ class Server
 		# @parameter endpoint [Async::HTTP::Endpoint] The downstream HTTP endpoint.
 		# @parameter control_endpoint [IO::Endpoint] The async-bus control endpoint.
 		# @parameter exchange_limit [Integer] The maximum outstanding responses per backend.
-		def initialize(endpoint, control_endpoint, exchange_limit: 8)
-			@registry = Registry.new(exchange_limit: exchange_limit)
-			@proxy = Proxy.new(@registry)
+		# @parameter configuration [Configuration] Request classification and scheduling policy.
+		def initialize(endpoint, control_endpoint, exchange_limit: 8, configuration: Configuration.default)
+			@registry = Registry.new(exchange_limit: exchange_limit, permit_limit: configuration.permit_limit)
+			@proxy = Proxy.new(@registry, configuration: configuration)
 			@http_server = Async::HTTP::Server.new(@proxy, endpoint)
 			@control_server = Control.new(control_endpoint, @registry)
 		end
@@ -26,6 +28,11 @@ def initialize(endpoint, control_endpoint, exchange_limit: 8)
 		# @attribute [Registry] The server's endpoint registry.
 		attr :registry
 		
+		# @attribute [Scheduler] The central request scheduler.
+		def scheduler
+			@proxy.scheduler
+		end
+		
 		# Run the HTTP and control servers.
 		# @parameter parent [Interface(:async)] The parent task.
 		# @returns [Async::Task] The server task.
diff --git a/readme.md b/readme.md
index faf8110..65e1801 100644
--- a/readme.md
+++ b/readme.md
@@ -4,7 +4,7 @@ Worker-aware HTTP load balancing with a global admission queue.
 
 [![Development Status](https://github.com/socketry/fantail/workflows/Test/badge.svg)](https://github.com/socketry/fantail/actions?workflow=Test)
 
-Fantail routes each request to a worker which is ready to process it. It separates the short-lived request-processing reservation from the potentially longer response exchange, so another request can begin after response headers arrive while the previous response body is still streaming.
+Fantail routes each request to a worker which is ready to process it. Configurable request queues can express worker affinity and load-shedding policy while a central scheduler remains responsible for matching requests to worker permits. Fantail separates the short-lived request-processing reservation from the potentially longer response exchange, so another request can begin after response headers arrive while the previous response body is still streaming.
 
 ## Usage
 
diff --git a/releases.md b/releases.md
index 67fa769..98e4a2e 100644
--- a/releases.md
+++ b/releases.md
@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Add configurable request queues, worker affinity policies, central permit scheduling, and load shedding.
+
 ## v0.0.1
 
   - Initial implementation.
diff --git a/test/fantail/configuration.rb b/test/fantail/configuration.rb
new file mode 100644
index 0000000..a894a55
--- /dev/null
+++ b/test/fantail/configuration.rb
@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "fantail/configuration"
+require "protocol/http/request"
+require "tmpdir"
+
+describe Fantail::Configuration do
+	let(:configuration) do
+		subject.define do |config|
+			config.queue :grpc do |queue|
+				queue.match{|request| request.headers["content-type"]&.start_with?("application/grpc")}
+				queue.balance :pack
+			end
+			
+			config.queue :liquid do |queue|
+				queue.balance :spread
+				queue.depth_limit 500
+				queue.wait_limit 0.25
+				queue.shed status: 429, retry_after: 1
+			end
+			
+			config.default_queue :liquid
+			config.pending_limit 1_000
+			config.permit_limit 2
+		end
+	end
+	
+	it "classifies requests and freezes the result" do
+		grpc = Protocol::HTTP::Request["POST", "/rpc", {"content-type" => "application/grpc+proto"}]
+		liquid = Protocol::HTTP::Request["GET", "/render"]
+		
+		expect(configuration.classify(grpc).name).to be == :grpc
+		expect(configuration.classify(liquid).name).to be == :liquid
+		expect(configuration.pending_limit).to be == 1_000
+		expect(configuration.permit_limit).to be == 2
+		expect(configuration).to be(:frozen?)
+		expect(configuration.queues.fetch(:liquid)).to be(:frozen?)
+	end
+	
+	it "loads trusted application configuration" do
+		Dir.mktmpdir do |directory|
+			path = File.join(directory, "fantail.rb")
+			File.write(path, <<~RUBY)
+				Fantail::Configuration.define do |config|
+					config.queue :default
+				end
+			RUBY
+			
+			loaded = subject.load(path)
+			expect(loaded.default_queue_name).to be == :default
+		end
+	end
+	
+	it "rejects configuration files with the wrong result" do
+		Dir.mktmpdir do |directory|
+			path = File.join(directory, "fantail.rb")
+			File.write(path, "Object.new\n")
+			
+			expect{subject.load(path)}.to raise_exception(TypeError)
+		end
+	end
+end
diff --git a/test/fantail/fixtures.rb b/test/fantail/fixtures.rb
index 22f8add..3cfe615 100644
--- a/test/fantail/fixtures.rb
+++ b/test/fantail/fixtures.rb
@@ -30,17 +30,17 @@ def closed?
 			end
 		end
 		
-		def make_registry(exchange_limit: 8, &client_factory)
+		def make_registry(exchange_limit: 8, permit_limit: 1, &client_factory)
 			@clients = {}
 			
-			backend_factory = proc do |endpoint, limit, available|
+			backend_factory = proc do |endpoint, exchange_limit, backend_permit_limit, available|
 				client = client_factory&.call(endpoint) || Client.new
 				@clients[endpoint.name] = client
 				
-				Backend.new(endpoint, client, exchange_limit: limit, &available)
+				Backend.new(endpoint, client, exchange_limit: exchange_limit, permit_limit: backend_permit_limit, &available)
 			end
 			
-			Registry.new(exchange_limit: exchange_limit, backend_factory: backend_factory)
+			Registry.new(exchange_limit: exchange_limit, permit_limit: permit_limit, backend_factory: backend_factory)
 		end
 	end
 end
diff --git a/test/fantail/registry.rb b/test/fantail/registry.rb
index 093216a..ef5edc8 100644
--- a/test/fantail/registry.rb
+++ b/test/fantail/registry.rb
@@ -70,6 +70,23 @@
 		expect(backend).not.to be(:processing?)
 	end
 	
+	it "supports multiple processing permits" do
+		registry.close
+		registry = make_registry(permit_limit: 2)
+		registry.replace([{name: "worker-1", url: "http://127.0.0.1:9301"}])
+		backend = registry["worker-1"]
+		
+		expect(backend.reserve(:grpc)).to be_truthy
+		expect(backend.reserve(:grpc)).to be_truthy
+		expect(backend.reserve(:grpc)).to be_falsey
+		expect(backend.processing_for(:grpc)).to be == 2
+		
+		backend.failed(:grpc)
+		backend.failed(:grpc)
+	ensure
+		registry&.close
+	end
+	
 	it "returns nil when no endpoints exist" do
 		expect(registry.acquire).to be_nil
 	end
diff --git a/test/fantail/scheduler.rb b/test/fantail/scheduler.rb
new file mode 100644
index 0000000..896ffb8
--- /dev/null
+++ b/test/fantail/scheduler.rb
@@ -0,0 +1,231 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "fantail"
+require "sus/fixtures/async/reactor_context"
+require_relative "fixtures"
+
+describe Fantail::Scheduler do
+	include Sus::Fixtures::Async::ReactorContext
+	include Fantail::Fixtures
+	
+	def make_configuration(permit_limit: 1, pending_limit: nil, &block)
+		Fantail::Configuration.define do |configuration|
+			configuration.permit_limit permit_limit
+			configuration.pending_limit pending_limit if pending_limit
+			block.call(configuration)
+		end
+	end
+	
+	def add_workers(registry, count = 2)
+		registry.replace(Array.new(count) do |index|
+			{name: "worker-#{index + 1}", url: "http://127.0.0.1:#{9301 + index}"}
+		end)
+	end
+	
+	def finish(reservation)
+		reservation.processed
+		reservation.release
+	end
+	
+	it "spreads work across workers with spare permits" do
+		configuration = make_configuration(permit_limit: 2) do |config|
+			config.queue(:liquid){|queue| queue.balance :spread}
+		end
+		registry = make_registry(permit_limit: configuration.permit_limit)
+		add_workers(registry)
+		scheduler = subject.new(registry, configuration)
+		
+		first = scheduler.acquire(Protocol::HTTP::Request["GET", "/first"])
+		second = scheduler.acquire(Protocol::HTTP::Request["GET", "/second"])
+		
+		expect([first.backend.name, second.backend.name]).to be == ["worker-1", "worker-2"]
+	ensure
+		finish(first) if first
+		finish(second) if second
+		registry&.close
+	end
+	
+	it "packs affinity work onto an active worker" do
+		configuration = make_configuration(permit_limit: 2) do |config|
+			config.queue(:grpc){|queue| queue.balance :pack, affinity: :grpc}
+		end
+		registry = make_registry(permit_limit: configuration.permit_limit)
+		add_workers(registry)
+		scheduler = subject.new(registry, configuration)
+		
+		first = scheduler.acquire(Protocol::HTTP::Request["POST", "/first"])
+		second = scheduler.acquire(Protocol::HTTP::Request["POST", "/second"])
+		
+		expect([first.backend.name, second.backend.name]).to be == ["worker-1", "worker-1"]
+	ensure
+		finish(first) if first
+		finish(second) if second
+		registry&.close
+	end
+	
+	it "keeps affinity work-conserving" do
+		configuration = make_configuration(permit_limit: 2) do |config|
+			config.queue(:grpc){|queue| queue.balance :pack}
+		end
+		registry = make_registry(permit_limit: configuration.permit_limit)
+		add_workers(registry)
+		scheduler = subject.new(registry, configuration)
+		
+		first = scheduler.acquire(Protocol::HTTP::Request["POST", "/first"])
+		second = scheduler.acquire(Protocol::HTTP::Request["POST", "/second"])
+		third = scheduler.acquire(Protocol::HTTP::Request["POST", "/third"])
+		
+		expect([first.backend.name, second.backend.name, third.backend.name]).to be == ["worker-1", "worker-1", "worker-2"]
+	ensure
+		finish(first) if first
+		finish(second) if second
+		finish(third) if third
+		registry&.close
+	end
+	
+	it "supports application balance policies" do
+		policy = Object.new
+		policy.define_singleton_method(:select){|backends, **| backends.last}
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.balance policy}
+		end
+		registry = make_registry
+		add_workers(registry)
+		scheduler = subject.new(registry, configuration)
+		
+		reservation = scheduler.acquire(Protocol::HTTP::Request["GET", "/"])
+		expect(reservation.backend.name).to be == "worker-2"
+	ensure
+		finish(reservation) if reservation
+		registry&.close
+	end
+	
+	it "uses another queue when the oldest queue has no eligible worker" do
+		configuration = make_configuration do |config|
+			config.queue :special do |queue|
+				queue.match{|request| request.path == "/special"}
+				queue.eligible{|backend, _request| backend.name == "special-worker"}
+			end
+			config.queue :default
+			config.default_queue :default
+		end
+		registry = make_registry
+		registry.replace([{name: "default-worker", url: "http://127.0.0.1:9301"}])
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		
+		special_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/special"])}
+		Fiber.scheduler.yield
+		default_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/default"])}
+		Fiber.scheduler.yield
+		
+		finish(held)
+		held = nil
+		default = default_task.wait
+		expect(default.backend.name).to be == "default-worker"
+		expect(special_task).not.to be(:finished?)
+	ensure
+		finish(held) if held
+		finish(default) if default
+		special_task&.stop
+		default_task&.stop
+		registry&.close
+	end
+	
+	it "dispatches the oldest eligible queue head first" do
+		configuration = make_configuration do |config|
+			config.queue(:alpha){|queue| queue.match{|request| request.path == "/alpha"}}
+			config.queue :beta
+			config.default_queue :beta
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		
+		alpha_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/alpha"])}
+		Fiber.scheduler.yield
+		beta_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/beta"])}
+		Fiber.scheduler.yield
+		
+		finish(held)
+		held = nil
+		alpha = alpha_task.wait
+		expect(beta_task).not.to be(:finished?)
+		finish(alpha)
+		alpha = nil
+		beta = beta_task.wait
+		expect(beta.backend.name).to be == "worker-1"
+	ensure
+		finish(held) if held
+		finish(alpha) if alpha
+		finish(beta) if beta
+		alpha_task&.stop
+		beta_task&.stop
+		registry&.close
+	end
+	
+	it "sheds requests when the queue depth limit is reached" do
+		configuration = make_configuration do |config|
+			config.queue :default do |queue|
+				queue.depth_limit 1
+				queue.shed status: 429, retry_after: 2
+			end
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
+		Fiber.scheduler.yield
+		
+		rejection = scheduler.acquire(Protocol::HTTP::Request["GET", "/rejected"])
+		response = rejection.response
+		
+		expect(response.status).to be == 429
+		expect(response.headers["retry-after"]).to be == "2"
+	ensure
+		response&.close
+		finish(held) if held
+		waiting = waiting_task&.wait
+		finish(waiting) if waiting
+		waiting_task&.stop
+		registry&.close
+	end
+	
+	it "sheds requests which exceed their queue wait limit" do
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.wait_limit 0.01}
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		
+		rejection = scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])
+		expect(rejection).to be_a(Fantail::Scheduler::Rejection)
+		expect(scheduler.pending_count).to be == 0
+	ensure
+		finish(held) if held
+		registry&.close
+	end
+	
+	it "supports application admission policies" do
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.admit{|request, **| request.path != "/shed"}}
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		
+		rejection = scheduler.acquire(Protocol::HTTP::Request["GET", "/shed"])
+		expect(rejection).to be_a(Fantail::Scheduler::Rejection)
+	ensure
+		finish(held) if held
+		registry&.close
+	end
+end

From 6bf9c75654baae780b44d9dd3c25e577249ddb35 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Sat, 29 Aug 2026 13:27:22 +1200
Subject: [PATCH 2/8] Complete queue policy coverage

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 lib/fantail/balance.rb        | 15 ++++++++++++
 lib/fantail/configuration.rb  | 38 +++++++++++++++++++++++++++++
 lib/fantail/scheduler.rb      |  9 +++++++
 test/fantail/configuration.rb |  8 +++++++
 test/fantail/registry.rb      | 20 ++++++++++++++++
 test/fantail/scheduler.rb     | 45 +++++++++++++++++++++++++++++++++++
 test/fantail/server.rb        |  1 +
 7 files changed, 136 insertions(+)

diff --git a/lib/fantail/balance.rb b/lib/fantail/balance.rb
index 803fd93..8637aaf 100644
--- a/lib/fantail/balance.rb
+++ b/lib/fantail/balance.rb
@@ -8,6 +8,11 @@ module Fantail
 	module Balance
 		# Prefer the backend with the fewest active requests.
 		class Spread
+			# Select the least-active backend, using its name for deterministic ties.
+			# @parameter backends [Array(Backend)] Eligible backends with available permits.
+			# @parameter queue [Configuration::Queue] The request queue being scheduled.
+			# @parameter request [Protocol::HTTP::Request] The pending request.
+			# @returns [Backend | Nil] The preferred backend.
 			def select(backends, queue:, request:)
 				backends.min_by{|backend| [backend.processing, backend.name]}
 			end
@@ -15,10 +20,16 @@ def select(backends, queue:, request:)
 		
 		# Prefer a backend which is already processing the same class of work.
 		class Pack
+			# @parameter affinity [Symbol | Nil] The queue affinity to pack, or the current queue by default.
 			def initialize(affinity: nil)
 				@affinity = affinity
 			end
 			
+			# Select the backend with the most active work for the affinity.
+			# @parameter backends [Array(Backend)] Eligible backends with available permits.
+			# @parameter queue [Configuration::Queue] The request queue being scheduled.
+			# @parameter request [Protocol::HTTP::Request] The pending request.
+			# @returns [Backend | Nil] The preferred backend.
 			def select(backends, queue:, request:)
 				affinity = @affinity || queue.name
 				backends.min_by do |backend|
@@ -27,6 +38,10 @@ def select(backends, queue:, request:)
 			end
 		end
 		
+		# Resolve a built-in policy name or validate an application policy object.
+		# @parameter policy [Symbol | #select] The policy name or object.
+		# @parameter options [Hash] Options for a built-in policy.
+		# @returns [#select] The resolved balance policy.
 		def self.coerce(policy, **options)
 			case policy
 			when :spread
diff --git a/lib/fantail/configuration.rb b/lib/fantail/configuration.rb
index f3190ba..4db001c 100644
--- a/lib/fantail/configuration.rb
+++ b/lib/fantail/configuration.rb
@@ -12,6 +12,7 @@ class Configuration
 		
 		# Configuration for one class of requests.
 		class Queue
+			# @parameter name [Symbol | String] The stable queue name.
 			def initialize(name)
 				@name = name.to_sym
 				@matcher = nil
@@ -29,25 +30,38 @@ def initialize(name)
 			attr :shed_status
 			attr :shed_headers
 			
+			# Set the request classifier for this queue.
+			# @yields {|request| ...} Whether a request belongs to this queue.
 			def match(&block)
 				raise ArgumentError, "A matcher block is required!" unless block
 				@matcher = block
 			end
 			
+			# Restrict the backends which may serve this queue.
+			# @yields {|backend, request| ...} Whether the backend is eligible.
 			def eligible(&block)
 				raise ArgumentError, "An eligibility block is required!" unless block
 				@eligibility = block
 			end
 			
+			# Set an application admission policy.
+			# @parameter policy [#admit? | #call | Nil] The admission policy object.
+			# @yields {|request, queue:, pending:| ...} Whether the request can wait.
 			def admit(policy = nil, &block)
 				@admission = policy || block
 				raise ArgumentError, "An admission policy is required!" unless @admission
 			end
 			
+			# Set the soft backend balance policy.
+			# @parameter policy [Symbol | #select] A built-in name or application policy.
+			# @parameter options [Hash] Options for a built-in policy.
 			def balance(policy, **options)
 				@balance = Balance.coerce(policy, **options)
 			end
 			
+			# Set or get the maximum number of requests waiting in this queue.
+			# @parameter value [Integer] The new limit when given.
+			# @returns [Integer | Nil] The configured limit.
 			def depth_limit(value = UNDEFINED)
 				return @depth_limit if value.equal?(UNDEFINED)
 				
@@ -56,6 +70,9 @@ def depth_limit(value = UNDEFINED)
 				@depth_limit = value
 			end
 			
+			# Set or get the maximum time a request may wait for a permit.
+			# @parameter value [Numeric] The new limit in seconds when given.
+			# @returns [Float | Nil] The configured limit.
 			def wait_limit(value = UNDEFINED)
 				return @wait_limit if value.equal?(UNDEFINED)
 				
@@ -64,20 +81,28 @@ def wait_limit(value = UNDEFINED)
 				@wait_limit = value
 			end
 			
+			# Configure the response used when admission is rejected.
+			# @parameter status [Integer] The HTTP response status.
+			# @parameter retry_after [Numeric | String | Nil] An optional Retry-After value.
+			# @parameter headers [Hash] Additional response headers.
 			def shed(status: 429, retry_after: nil, headers: {})
 				@shed_status = Integer(status)
 				@shed_headers = headers.transform_keys(&:to_s)
 				@shed_headers["retry-after"] = retry_after.to_s if retry_after
 			end
 			
+			# @parameter request [Protocol::HTTP::Request] The request to classify.
+			# @returns [Boolean | Nil] Whether the request matches this queue.
 			def match?(request)
 				@matcher&.call(request)
 			end
 			
+			# @returns [Boolean] Whether a backend may serve the request.
 			def eligible?(backend, request)
 				!@eligibility || @eligibility.call(backend, request)
 			end
 			
+			# @returns [Boolean] Whether a request may enter the pending queue.
 			def admit?(request, pending:)
 				return true unless @admission
 				
@@ -88,6 +113,8 @@ def admit?(request, pending:)
 				end
 			end
 			
+			# Validate and freeze this queue definition.
+			# @returns [Queue] The finalized queue.
 			def finalize
 				@balance_policy = @balance
 				@shed_headers.freeze
@@ -95,12 +122,16 @@ def finalize
 			end
 		end
 		
+		# Build and finalize a configuration.
+		# @yields {|configuration| ...} The mutable configuration builder.
+		# @returns [Configuration] The immutable configuration.
 		def self.define
 			configuration = new
 			yield configuration if block_given?
 			configuration.finalize
 		end
 		
+		# @returns [Configuration] A single-queue, single-permit configuration.
 		def self.default
 			@default ||= define do |configuration|
 				configuration.queue(:default)
@@ -120,6 +151,7 @@ def self.load(path)
 			configuration
 		end
 		
+		# Initialize an empty mutable configuration builder.
 		def initialize
 			@queues = {}
 			@default_queue = nil
@@ -141,6 +173,8 @@ def queue(name)
 			queue
 		end
 		
+		# Select the fallback queue for unmatched requests.
+		# @parameter name [Symbol | String] A previously or subsequently defined queue.
 		def default_queue(name)
 			@default_queue = name.to_sym
 		end
@@ -163,6 +197,8 @@ def permit_limit(value = UNDEFINED)
 			@permit_limit = value
 		end
 		
+		# Classify a request using matchers in definition order.
+		# @returns [Queue] The matching or default queue.
 		def classify(request)
 			@queues.each_value do |queue|
 				return queue if queue.match?(request)
@@ -171,6 +207,8 @@ def classify(request)
 			@queues.fetch(@default_queue)
 		end
 		
+		# Validate and freeze the complete configuration.
+		# @returns [Configuration] The finalized configuration.
 		def finalize
 			raise ArgumentError, "At least one queue must be defined!" if @queues.empty?
 			@default_queue ||= @queues.keys.first
diff --git a/lib/fantail/scheduler.rb b/lib/fantail/scheduler.rb
index e8578b5..c57aeb1 100644
--- a/lib/fantail/scheduler.rb
+++ b/lib/fantail/scheduler.rb
@@ -13,6 +13,8 @@ module Fantail
 	class Scheduler
 		# A reserved processing permit and response exchange.
 		class Reservation
+			# @parameter backend [Backend] The reserved backend.
+			# @parameter queue_name [Symbol] The queue consuming the permit.
 			def initialize(backend, queue_name)
 				@backend = backend
 				@queue_name = queue_name
@@ -20,14 +22,17 @@ def initialize(backend, queue_name)
 			
 			attr :backend
 			
+			# Release the processing permit after response headers arrive.
 			def processed
 				@backend.processed(@queue_name)
 			end
 			
+			# Release the processing permit and exchange after an upstream failure.
 			def failed
 				@backend.failed(@queue_name)
 			end
 			
+			# Release the response exchange after its body closes.
 			def release
 				@backend.release
 			end
@@ -35,10 +40,12 @@ def release
 		
 		# A queue admission rejection.
 		class Rejection
+			# @parameter queue [Configuration::Queue] The queue which rejected admission.
 			def initialize(queue)
 				@queue = queue
 			end
 			
+			# @returns [Protocol::HTTP::Response] The configured shedding response.
 			def response
 				headers = {"content-type" => "text/plain"}.merge(@queue.shed_headers)
 				Protocol::HTTP::Response[@queue.shed_status, headers, ["Request queue is full.\n"]]
@@ -47,6 +54,8 @@ def response
 		
 		Entry = Struct.new(:request, :queue, :enqueued_at, :result, :pending, :assignment)
 		
+		# @parameter registry [Registry] The available backend registry.
+		# @parameter configuration [Configuration] Request and scheduling policy.
 		def initialize(registry, configuration = Configuration.default)
 			@registry = registry
 			@configuration = configuration
diff --git a/test/fantail/configuration.rb b/test/fantail/configuration.rb
index a894a55..26df204 100644
--- a/test/fantail/configuration.rb
+++ b/test/fantail/configuration.rb
@@ -62,4 +62,12 @@
 			expect{subject.load(path)}.to raise_exception(TypeError)
 		end
 	end
+	
+	it "rejects invalid balance policies" do
+		expect do
+			subject.define do |config|
+				config.queue(:default){|queue| queue.balance Object.new}
+			end
+		end.to raise_exception(ArgumentError)
+	end
 end
diff --git a/test/fantail/registry.rb b/test/fantail/registry.rb
index ef5edc8..19817f3 100644
--- a/test/fantail/registry.rb
+++ b/test/fantail/registry.rb
@@ -4,9 +4,11 @@
 # Copyright, 2026, by Samuel Williams.
 
 require "fantail"
+require "sus/fixtures/async/reactor_context"
 require_relative "fixtures"
 
 describe Fantail::Registry do
+	include Sus::Fixtures::Async::ReactorContext
 	include Fantail::Fixtures
 	
 	let(:registry) {make_registry}
@@ -70,6 +72,24 @@
 		expect(backend).not.to be(:processing?)
 	end
 	
+	it "wakes an acquisition when a permit is released" do
+		registry.replace([{name: "worker-1", url: "http://127.0.0.1:9301"}])
+		first = registry.acquire
+		second_task = Async{registry.acquire}
+		Fiber.scheduler.yield
+		
+		expect(second_task).not.to be(:finished?)
+		first.failed
+		first = nil
+		
+		second = second_task.wait
+		expect(second.name).to be == "worker-1"
+	ensure
+		first&.failed
+		second&.failed
+		second_task&.stop
+	end
+	
 	it "supports multiple processing permits" do
 		registry.close
 		registry = make_registry(permit_limit: 2)
diff --git a/test/fantail/scheduler.rb b/test/fantail/scheduler.rb
index 896ffb8..ed75299 100644
--- a/test/fantail/scheduler.rb
+++ b/test/fantail/scheduler.rb
@@ -103,6 +103,32 @@ def finish(reservation)
 		registry&.close
 	end
 	
+	it "retries selection when a permit is consumed concurrently" do
+		consumed = nil
+		policy = Object.new
+		policy.define_singleton_method(:select) do |backends, queue:, **|
+			unless consumed
+				consumed = backends.first
+				consumed.reserve(queue.name)
+			end
+			
+			backends.first
+		end
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.balance policy}
+		end
+		registry = make_registry
+		add_workers(registry)
+		scheduler = subject.new(registry, configuration)
+		
+		reservation = scheduler.acquire(Protocol::HTTP::Request["GET", "/"])
+		expect(reservation.backend.name).to be == "worker-2"
+	ensure
+		finish(reservation) if reservation
+		consumed&.failed(:default)
+		registry&.close
+	end
+	
 	it "uses another queue when the oldest queue has no eligible worker" do
 		configuration = make_configuration do |config|
 			config.queue :special do |queue|
@@ -181,6 +207,7 @@ def finish(reservation)
 		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
 		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
 		Fiber.scheduler.yield
+		expect(scheduler.pending_count(:default)).to be == 1
 		
 		rejection = scheduler.acquire(Protocol::HTTP::Request["GET", "/rejected"])
 		response = rejection.response
@@ -228,4 +255,22 @@ def finish(reservation)
 		finish(held) if held
 		registry&.close
 	end
+	
+	it "supports application admission policy objects" do
+		policy = Object.new
+		policy.define_singleton_method(:admit?){|request, **| request.path != "/shed"}
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.admit policy}
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		
+		rejection = scheduler.acquire(Protocol::HTTP::Request["GET", "/shed"])
+		expect(rejection).to be_a(Fantail::Scheduler::Rejection)
+	ensure
+		finish(held) if held
+		registry&.close
+	end
 end
diff --git a/test/fantail/server.rb b/test/fantail/server.rb
index 87715f0..4ef83e0 100644
--- a/test/fantail/server.rb
+++ b/test/fantail/server.rb
@@ -40,6 +40,7 @@ def wait_until
 			control_bound_endpoint = control_endpoint.bound
 			
 			server = subject.new(downstream_endpoint, control_bound_endpoint)
+			expect(server.scheduler).to be_a(Fantail::Scheduler)
 			server_task = server.run
 			
 			monitor = Fantail::Monitor.new(control_endpoint)

From 60dff328eeaa10d94c499fe993672dfb0264ac0f Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Wed, 2 Sep 2026 19:51:13 +1200
Subject: [PATCH 3/8] Handle queued request cancellation

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 lib/fantail/scheduler.rb  | 53 +++++++++++++++++++++++-----
 lib/fantail/server.rb     |  3 +-
 test/fantail/scheduler.rb | 73 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 120 insertions(+), 9 deletions(-)

diff --git a/lib/fantail/scheduler.rb b/lib/fantail/scheduler.rb
index c57aeb1..94652fb 100644
--- a/lib/fantail/scheduler.rb
+++ b/lib/fantail/scheduler.rb
@@ -62,6 +62,7 @@ def initialize(registry, configuration = Configuration.default)
 			@guard = Thread::Mutex.new
 			@pending = configuration.queues.to_h{|name, queue| [name, []]}
 			@pending_count = 0
+			@closed = false
 			
 			@registry.on_available{schedule}
 		end
@@ -71,7 +72,9 @@ def acquire(request)
 			queue = @configuration.classify(request)
 			entry = nil
 			result = @guard.synchronize do
-				if reservation = reserve(queue, request)
+				if @closed
+					nil
+				elsif reservation = reserve(queue, request)
 					reservation
 				elsif @registry.empty?
 					nil
@@ -88,22 +91,56 @@ def acquire(request)
 			
 			return result if result
 			return nil unless entry
-			return entry.assignment if entry.assignment
+			
+			if assignment = entry.assignment
+				entry = nil
+				return assignment
+			end
 			
 			if wait_limit = queue.wait_limit
 				remaining = wait_limit - (now - entry.enqueued_at)
 				result = entry.result.dequeue(timeout: remaining) if remaining.positive?
-				return result if result
+				if result
+					entry = nil
+					return result
+				end
 				
-				return cancel(entry)
+				result = cancel(entry)
+				entry = nil
+				return result
 			else
-				return entry.result.dequeue
+				result = entry.result.dequeue
+				entry = nil
+				return result
+			end
+		ensure
+			if entry && assignment = cancel(entry, rejection: false)
+				# The request was assigned concurrently but its waiting task was
+				# interrupted before receiving the reservation. No upstream request
+				# was started, so release both the permit and response exchange.
+				assignment.failed
 			end
 		end
 		
 		# Try to dispatch pending requests after capacity changes.
 		def schedule
-			@guard.synchronize{schedule_locked}
+			@guard.synchronize{schedule_locked unless @closed}
+		end
+		
+		# Stop accepting requests and wake all tasks waiting for a permit.
+		def close
+			entries = @guard.synchronize do
+				return if @closed
+				
+				@closed = true
+				entries = @pending.values.flatten(1)
+				@pending.each_value(&:clear)
+				@pending_count = 0
+				entries.each{|entry| entry.pending = false}
+				entries
+			end
+			
+			entries.each{|entry| entry.result.close}
 		end
 		
 		# @parameter queue_name [Symbol | String | Nil] An optional queue to inspect.
@@ -133,14 +170,14 @@ def reject?(queue, request)
 			false
 		end
 		
-		def cancel(entry)
+		def cancel(entry, rejection: true)
 			@guard.synchronize do
 				return entry.assignment unless entry.pending
 				
 				@pending.fetch(entry.queue.name).delete(entry)
 				entry.pending = false
 				@pending_count -= 1
-				Rejection.new(entry.queue)
+				Rejection.new(entry.queue) if rejection
 			end
 		end
 		
diff --git a/lib/fantail/server.rb b/lib/fantail/server.rb
index 66bf4a7..b915ab1 100644
--- a/lib/fantail/server.rb
+++ b/lib/fantail/server.rb
@@ -46,8 +46,9 @@ def run(parent: Async::Task.current)
 			end
 		end
 		
-		# Close the endpoint registry.
+		# Stop pending requests and close the endpoint registry.
 		def close
+			@proxy.scheduler.close
 			@registry.close
 		end
 	end
diff --git a/test/fantail/scheduler.rb b/test/fantail/scheduler.rb
index ed75299..0f9a4f7 100644
--- a/test/fantail/scheduler.rb
+++ b/test/fantail/scheduler.rb
@@ -240,6 +240,79 @@ def finish(reservation)
 		registry&.close
 	end
 	
+	it "withdraws an interrupted request waiting for a permit" do
+		configuration = make_configuration do |config|
+			config.queue :default
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
+		Fiber.scheduler.yield
+		
+		expect(scheduler.pending_count).to be == 1
+		waiting_task.stop
+		expect(scheduler.pending_count).to be == 0
+		
+		finish(held)
+		held = nil
+		replacement = scheduler.acquire(Protocol::HTTP::Request["GET", "/replacement"])
+		expect(replacement.backend.name).to be == "worker-1"
+	ensure
+		finish(held) if held
+		finish(replacement) if replacement
+		waiting_task&.stop
+		registry&.close
+	end
+	
+	it "releases an assignment when interruption races with dispatch" do
+		configuration = make_configuration do |config|
+			config.queue :default
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
+		Fiber.scheduler.yield
+		
+		finish(held)
+		held = nil
+		waiting_task.stop
+		
+		backend = registry["worker-1"]
+		expect(backend.processing).to be == 0
+		expect(backend.exchanges).to be == 0
+		expect(backend).to be(:available?)
+	ensure
+		finish(held) if held
+		waiting_task&.stop
+		registry&.close
+	end
+	
+	it "wakes pending requests when closed" do
+		configuration = make_configuration do |config|
+			config.queue :default
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
+		Fiber.scheduler.yield
+		
+		expect(scheduler.pending_count).to be == 1
+		scheduler.close
+		expect(waiting_task.wait).to be_nil
+		expect(scheduler.pending_count).to be == 0
+		expect(scheduler.acquire(Protocol::HTTP::Request["GET", "/closed"])).to be_nil
+	ensure
+		finish(held) if held
+		waiting_task&.stop
+		registry&.close
+	end
+	
 	it "supports application admission policies" do
 		configuration = make_configuration do |config|
 			config.queue(:default){|queue| queue.admit{|request, **| request.path != "/shed"}}

From 6ceacb15a7a4261ef524d71ba2f014fd19a32571 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Thu, 3 Sep 2026 09:37:17 +1200
Subject: [PATCH 4/8] Add scoped configuration loader

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 context/getting-started.md       | 44 ++++++++++---------
 guides/getting-started/readme.md | 44 ++++++++++---------
 lib/fantail/configuration.rb     | 48 +++++++++++++--------
 lib/fantail/loader.rb            | 73 ++++++++++++++++++++++++++++++++
 test/fantail/configuration.rb    | 49 ++++++++++++---------
 test/fantail/scheduler.rb        |  2 +-
 6 files changed, 179 insertions(+), 81 deletions(-)
 create mode 100644 lib/fantail/loader.rb

diff --git a/context/getting-started.md b/context/getting-started.md
index 0bfe9f7..f6bfeaf 100644
--- a/context/getting-started.md
+++ b/context/getting-started.md
@@ -65,33 +65,33 @@ The scheduler owns all permits. Request queues can decide which workers are elig
 
 ## Request Queues
 
-Fantail configuration is trusted application Ruby. The file's final expression must be an immutable `Fantail::Configuration`:
+Fantail configuration is trusted application Ruby evaluated using a scoped configuration loader:
 
 ~~~ ruby
 # config/fantail.rb
-Fantail::Configuration.define do |config|
-	config.queue :liquid do |queue|
-		queue.match{|request| request.path.start_with?("/render")}
-		queue.balance :spread
-		queue.depth_limit 500
-		queue.wait_limit 0.25
-		queue.shed status: 429, retry_after: 1
-	end
-	
-	config.queue :grpc do |queue|
-		queue.match do |request|
-			request.headers["content-type"]&.start_with?("application/grpc")
-		end
-		
-		queue.balance :pack, affinity: :grpc
+queue :liquid do
+	match{|request| request.path.start_with?("/render")}
+	balance :spread
+	depth_limit 500
+	wait_limit 0.25
+	shed status: 429, retry_after: 1
+end
+
+queue :grpc do
+	match do |request|
+		request.headers["content-type"]&.start_with?("application/grpc")
 	end
 	
-	config.default_queue :liquid
-	config.pending_limit 1_000
-	config.permit_limit 1
+	balance :pack, affinity: :grpc
 end
+
+default_queue :liquid
+pending_limit 1_000
+permit_limit 1
 ~~~
 
+Configuration can be split into files relative to the file being evaluated using `load_file "queues.rb"`.
+
 Matchers are evaluated in definition order, followed by the default queue. Across queues, the oldest eligible head request is dispatched first. If that request has no eligible worker, another queue can use the available permit.
 
 The built-in `:spread` policy prefers the least-active worker. The `:pack` policy prefers a worker already processing the specified affinity, while remaining bounded by its permits. An application can supply a policy object implementing `select(backends, queue:, request:)`, and can restrict hard eligibility with `queue.eligible`.
@@ -103,7 +103,9 @@ The built-in `:spread` policy prefers the least-active worker. The `:pack` polic
 Applications can add an admission policy with either a block or an object implementing `admit?(request, queue:, pending:)`:
 
 ~~~ ruby
-queue.admit do |request, queue:, pending:|
-	pending < application_limit_for(queue.name)
+queue :default do
+	admit do |request, queue:, pending:|
+		pending < application_limit_for(queue.name)
+	end
 end
 ~~~
diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md
index 0bfe9f7..f6bfeaf 100644
--- a/guides/getting-started/readme.md
+++ b/guides/getting-started/readme.md
@@ -65,33 +65,33 @@ The scheduler owns all permits. Request queues can decide which workers are elig
 
 ## Request Queues
 
-Fantail configuration is trusted application Ruby. The file's final expression must be an immutable `Fantail::Configuration`:
+Fantail configuration is trusted application Ruby evaluated using a scoped configuration loader:
 
 ~~~ ruby
 # config/fantail.rb
-Fantail::Configuration.define do |config|
-	config.queue :liquid do |queue|
-		queue.match{|request| request.path.start_with?("/render")}
-		queue.balance :spread
-		queue.depth_limit 500
-		queue.wait_limit 0.25
-		queue.shed status: 429, retry_after: 1
-	end
-	
-	config.queue :grpc do |queue|
-		queue.match do |request|
-			request.headers["content-type"]&.start_with?("application/grpc")
-		end
-		
-		queue.balance :pack, affinity: :grpc
+queue :liquid do
+	match{|request| request.path.start_with?("/render")}
+	balance :spread
+	depth_limit 500
+	wait_limit 0.25
+	shed status: 429, retry_after: 1
+end
+
+queue :grpc do
+	match do |request|
+		request.headers["content-type"]&.start_with?("application/grpc")
 	end
 	
-	config.default_queue :liquid
-	config.pending_limit 1_000
-	config.permit_limit 1
+	balance :pack, affinity: :grpc
 end
+
+default_queue :liquid
+pending_limit 1_000
+permit_limit 1
 ~~~
 
+Configuration can be split into files relative to the file being evaluated using `load_file "queues.rb"`.
+
 Matchers are evaluated in definition order, followed by the default queue. Across queues, the oldest eligible head request is dispatched first. If that request has no eligible worker, another queue can use the available permit.
 
 The built-in `:spread` policy prefers the least-active worker. The `:pack` policy prefers a worker already processing the specified affinity, while remaining bounded by its permits. An application can supply a policy object implementing `select(backends, queue:, request:)`, and can restrict hard eligibility with `queue.eligible`.
@@ -103,7 +103,9 @@ The built-in `:spread` policy prefers the least-active worker. The `:pack` polic
 Applications can add an admission policy with either a block or an object implementing `admit?(request, queue:, pending:)`:
 
 ~~~ ruby
-queue.admit do |request, queue:, pending:|
-	pending < application_limit_for(queue.name)
+queue :default do
+	admit do |request, queue:, pending:|
+		pending < application_limit_for(queue.name)
+	end
 end
 ~~~
diff --git a/lib/fantail/configuration.rb b/lib/fantail/configuration.rb
index 4db001c..4e0187d 100644
--- a/lib/fantail/configuration.rb
+++ b/lib/fantail/configuration.rb
@@ -4,6 +4,7 @@
 # Copyright, 2026, by Samuel Williams.
 
 require_relative "balance"
+require_relative "loader"
 
 module Fantail
 	# Immutable request queue and admission configuration.
@@ -122,33 +123,40 @@ def finalize
 			end
 		end
 		
-		# Build and finalize a configuration.
-		# @yields {|configuration| ...} The mutable configuration builder.
+		# Build and finalize a configuration using a scoped loader.
+		# @parameter root [String] The root for relative configuration files.
+		# @yields {|loader| ...} The configuration loader.
 		# @returns [Configuration] The immutable configuration.
-		def self.define
+		def self.build(root: Dir.pwd, &block)
 			configuration = new
-			yield configuration if block_given?
+			loader = Loader.new(configuration, root)
+			
+			if block
+				if block.arity.zero?
+					loader.instance_eval(&block)
+				else
+					block.call(loader)
+				end
+			end
+			
 			configuration.finalize
 		end
 		
 		# @returns [Configuration] A single-queue, single-permit configuration.
 		def self.default
-			@default ||= define do |configuration|
-				configuration.queue(:default)
-				configuration.default_queue(:default)
+			@default ||= build do
+				queue(:default)
+				default_queue(:default)
 			end
 		end
 		
-		# Load trusted application configuration. The final expression must be a Configuration.
-		def self.load(path)
-			path = File.expand_path(path)
-			configuration = TOPLEVEL_BINDING.eval(File.read(path), path)
-			
-			unless configuration.is_a?(self)
-				raise TypeError, "#{path} must return a Fantail::Configuration!"
-			end
-			
-			configuration
+		# Load and finalize trusted application configuration files.
+		# @parameter paths [String | Array(String)] The configuration files to load.
+		# @returns [Configuration] The immutable configuration.
+		def self.load(paths)
+			configuration = new
+			Array(paths).each{|path| configuration.load_file(path)}
+			configuration.finalize
 		end
 		
 		# Initialize an empty mutable configuration builder.
@@ -162,6 +170,12 @@ def initialize
 		attr :queues
 		attr :default_queue_name
 		
+		# Load a trusted configuration file into this configuration.
+		# @parameter path [String] The configuration file to load.
+		def load_file(path)
+			Loader.load_file(self, path)
+		end
+		
 		# Define a named request queue. Matchers are evaluated in definition order.
 		def queue(name)
 			name = name.to_sym
diff --git a/lib/fantail/loader.rb b/lib/fantail/loader.rb
new file mode 100644
index 0000000..f769c6a
--- /dev/null
+++ b/lib/fantail/loader.rb
@@ -0,0 +1,73 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+module Fantail
+	# The domain-specific language for building and loading configuration.
+	class Loader
+		# Initialize a loader attached to a configuration.
+		# @parameter configuration [Configuration] The configuration being built.
+		# @parameter root [String | Nil] The root for relative configuration files.
+		def initialize(configuration, root = nil)
+			@configuration = configuration
+			@root = root
+		end
+		
+		# @attribute [String | Nil] The root for relative configuration files.
+		attr :root
+		
+		# Evaluate a configuration file using a scoped loader.
+		# @parameter configuration [Configuration] The configuration being built.
+		# @parameter path [String] The configuration file to load.
+		def self.load_file(configuration, path)
+			realpath = File.realpath(path)
+			loader = new(configuration, File.dirname(realpath))
+			
+			if Module.method_defined?(:set_temporary_name)
+				loader.singleton_class.set_temporary_name("#{self}[#{path.inspect}]")
+			end
+			
+			loader.instance_eval(File.read(realpath), realpath)
+		end
+		
+		# Load another configuration file relative to this loader.
+		# @parameter path [String] The relative or absolute configuration path.
+		def load_file(path)
+			self.class.load_file(@configuration, File.expand_path(path, @root))
+		end
+		
+		# Define a named request queue.
+		# @parameter name [Symbol | String] The stable queue name.
+		# @yields {|queue| ...} The queue configuration, or evaluates the block as its DSL.
+		def queue(name, &block)
+			return @configuration.queue(name) unless block
+			
+			@configuration.queue(name) do |queue|
+				if block.arity.zero?
+					queue.instance_eval(&block)
+				else
+					block.call(queue)
+				end
+			end
+		end
+		
+		# Select the fallback queue for unmatched requests.
+		# @parameter name [Symbol | String] A defined queue name.
+		def default_queue(name)
+			@configuration.default_queue(name)
+		end
+		
+		# Set the global pending request limit.
+		# @parameter value [Integer] The maximum number of pending requests.
+		def pending_limit(value)
+			@configuration.pending_limit(value)
+		end
+		
+		# Set the number of processing permits provided by each worker.
+		# @parameter value [Integer] The number of permits per worker.
+		def permit_limit(value)
+			@configuration.permit_limit(value)
+		end
+	end
+end
diff --git a/test/fantail/configuration.rb b/test/fantail/configuration.rb
index 26df204..f1bcc63 100644
--- a/test/fantail/configuration.rb
+++ b/test/fantail/configuration.rb
@@ -9,22 +9,22 @@
 
 describe Fantail::Configuration do
 	let(:configuration) do
-		subject.define do |config|
-			config.queue :grpc do |queue|
-				queue.match{|request| request.headers["content-type"]&.start_with?("application/grpc")}
-				queue.balance :pack
+		subject.build do
+			queue :grpc do
+				match {|request| request.headers["content-type"]&.start_with?("application/grpc")}
+				balance :pack
 			end
 			
-			config.queue :liquid do |queue|
-				queue.balance :spread
-				queue.depth_limit 500
-				queue.wait_limit 0.25
-				queue.shed status: 429, retry_after: 1
+			queue :liquid do
+				balance :spread
+				depth_limit 500
+				wait_limit 0.25
+				shed status: 429, retry_after: 1
 			end
 			
-			config.default_queue :liquid
-			config.pending_limit 1_000
-			config.permit_limit 2
+			default_queue :liquid
+			pending_limit 1_000
+			permit_limit 2
 		end
 	end
 	
@@ -43,29 +43,36 @@
 	it "loads trusted application configuration" do
 		Dir.mktmpdir do |directory|
 			path = File.join(directory, "fantail.rb")
-			File.write(path, <<~RUBY)
-				Fantail::Configuration.define do |config|
-					config.queue :default
-				end
-			RUBY
+			File.write(path, "queue :default\n")
 			
 			loaded = subject.load(path)
 			expect(loaded.default_queue_name).to be == :default
 		end
 	end
 	
-	it "rejects configuration files with the wrong result" do
+	it "loads configuration files relative to the current file" do
 		Dir.mktmpdir do |directory|
+			queues_path = File.join(directory, "queues.rb")
 			path = File.join(directory, "fantail.rb")
-			File.write(path, "Object.new\n")
+			File.write(queues_path, "queue :default\n")
+			File.write(path, "load_file 'queues.rb'\n")
 			
-			expect{subject.load(path)}.to raise_exception(TypeError)
+			loaded = subject.load(path)
+			expect(loaded.default_queue_name).to be == :default
+		end
+	end
+	
+	it "builds configuration with an explicit loader" do
+		configuration = subject.build do |loader|
+			loader.queue(:default)
 		end
+		
+		expect(configuration.default_queue_name).to be == :default
 	end
 	
 	it "rejects invalid balance policies" do
 		expect do
-			subject.define do |config|
+			subject.build do |config|
 				config.queue(:default){|queue| queue.balance Object.new}
 			end
 		end.to raise_exception(ArgumentError)
diff --git a/test/fantail/scheduler.rb b/test/fantail/scheduler.rb
index 0f9a4f7..f2c377b 100644
--- a/test/fantail/scheduler.rb
+++ b/test/fantail/scheduler.rb
@@ -12,7 +12,7 @@
 	include Fantail::Fixtures
 	
 	def make_configuration(permit_limit: 1, pending_limit: nil, &block)
-		Fantail::Configuration.define do |configuration|
+		Fantail::Configuration.build do |configuration|
 			configuration.permit_limit permit_limit
 			configuration.pending_limit pending_limit if pending_limit
 			block.call(configuration)

From f26e0627de55582769217dab8e06a25b095d12d6 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Thu, 3 Sep 2026 09:42:12 +1200
Subject: [PATCH 5/8] Cover scheduler assignment races

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 lib/fantail/scheduler.rb  |  3 +--
 test/fantail/scheduler.rb | 44 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 45 insertions(+), 2 deletions(-)

diff --git a/lib/fantail/scheduler.rb b/lib/fantail/scheduler.rb
index 94652fb..d20a4cb 100644
--- a/lib/fantail/scheduler.rb
+++ b/lib/fantail/scheduler.rb
@@ -89,8 +89,7 @@ def acquire(request)
 				end
 			end
 			
-			return result if result
-			return nil unless entry
+			return result unless entry
 			
 			if assignment = entry.assignment
 				entry = nil
diff --git a/test/fantail/scheduler.rb b/test/fantail/scheduler.rb
index f2c377b..f796f25 100644
--- a/test/fantail/scheduler.rb
+++ b/test/fantail/scheduler.rb
@@ -129,6 +129,28 @@ def finish(reservation)
 		registry&.close
 	end
 	
+	it "accepts an assignment made while entering the pending queue" do
+		selections = 0
+		policy = Object.new
+		policy.define_singleton_method(:select) do |backends, **|
+			selections += 1
+			backends.first unless selections == 1
+		end
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.balance policy}
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		
+		reservation = scheduler.acquire(Protocol::HTTP::Request["GET", "/"])
+		expect(reservation.backend.name).to be == "worker-1"
+		expect(selections).to be == 2
+	ensure
+		finish(reservation) if reservation
+		registry&.close
+	end
+	
 	it "uses another queue when the oldest queue has no eligible worker" do
 		configuration = make_configuration do |config|
 			config.queue :special do |queue|
@@ -240,6 +262,28 @@ def finish(reservation)
 		registry&.close
 	end
 	
+	it "dispatches a waiting request before its wait limit" do
+		configuration = make_configuration do |config|
+			config.queue(:default){|queue| queue.wait_limit 1}
+		end
+		registry = make_registry
+		add_workers(registry, 1)
+		scheduler = subject.new(registry, configuration)
+		held = scheduler.acquire(Protocol::HTTP::Request["GET", "/held"])
+		waiting_task = Async{scheduler.acquire(Protocol::HTTP::Request["GET", "/waiting"])}
+		Fiber.scheduler.yield
+		
+		finish(held)
+		held = nil
+		reservation = waiting_task.wait
+		expect(reservation.backend.name).to be == "worker-1"
+	ensure
+		finish(held) if held
+		finish(reservation) if reservation
+		waiting_task&.stop
+		registry&.close
+	end
+	
 	it "withdraws an interrupted request waiting for a permit" do
 		configuration = make_configuration do |config|
 			config.queue :default

From d67398ecf843a0a055a097347788c7893cd6dd27 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Thu, 3 Sep 2026 10:46:14 +1200
Subject: [PATCH 6/8] Separate configuration builders from values

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 context/getting-started.md       |   2 +-
 guides/getting-started/readme.md |   2 +-
 lib/fantail.rb                   |   1 +
 lib/fantail/balance.rb           |   4 +-
 lib/fantail/configuration.rb     | 259 +++++++++++--------------------
 lib/fantail/loader.rb            |  73 ---------
 lib/fantail/queue.rb             | 149 ++++++++++++++++++
 lib/fantail/scheduler.rb         |   2 +-
 test/fantail/configuration.rb    |  13 +-
 9 files changed, 254 insertions(+), 251 deletions(-)
 delete mode 100644 lib/fantail/loader.rb
 create mode 100644 lib/fantail/queue.rb

diff --git a/context/getting-started.md b/context/getting-started.md
index f6bfeaf..809bc79 100644
--- a/context/getting-started.md
+++ b/context/getting-started.md
@@ -65,7 +65,7 @@ The scheduler owns all permits. Request queues can decide which workers are elig
 
 ## Request Queues
 
-Fantail configuration is trusted application Ruby evaluated using a scoped configuration loader:
+Fantail configuration is trusted application Ruby evaluated using a scoped configuration builder:
 
 ~~~ ruby
 # config/fantail.rb
diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md
index f6bfeaf..809bc79 100644
--- a/guides/getting-started/readme.md
+++ b/guides/getting-started/readme.md
@@ -65,7 +65,7 @@ The scheduler owns all permits. Request queues can decide which workers are elig
 
 ## Request Queues
 
-Fantail configuration is trusted application Ruby evaluated using a scoped configuration loader:
+Fantail configuration is trusted application Ruby evaluated using a scoped configuration builder:
 
 ~~~ ruby
 # config/fantail.rb
diff --git a/lib/fantail.rb b/lib/fantail.rb
index 5becbb1..b5da7aa 100644
--- a/lib/fantail.rb
+++ b/lib/fantail.rb
@@ -5,6 +5,7 @@
 
 require_relative "fantail/version"
 require_relative "fantail/balance"
+require_relative "fantail/queue"
 require_relative "fantail/configuration"
 require_relative "fantail/endpoint"
 require_relative "fantail/backend"
diff --git a/lib/fantail/balance.rb b/lib/fantail/balance.rb
index 8637aaf..6a688fa 100644
--- a/lib/fantail/balance.rb
+++ b/lib/fantail/balance.rb
@@ -10,7 +10,7 @@ module Balance
 		class Spread
 			# Select the least-active backend, using its name for deterministic ties.
 			# @parameter backends [Array(Backend)] Eligible backends with available permits.
-			# @parameter queue [Configuration::Queue] The request queue being scheduled.
+			# @parameter queue [Queue] The request queue being scheduled.
 			# @parameter request [Protocol::HTTP::Request] The pending request.
 			# @returns [Backend | Nil] The preferred backend.
 			def select(backends, queue:, request:)
@@ -27,7 +27,7 @@ def initialize(affinity: nil)
 			
 			# Select the backend with the most active work for the affinity.
 			# @parameter backends [Array(Backend)] Eligible backends with available permits.
-			# @parameter queue [Configuration::Queue] The request queue being scheduled.
+			# @parameter queue [Queue] The request queue being scheduled.
 			# @parameter request [Protocol::HTTP::Request] The pending request.
 			# @returns [Backend | Nil] The preferred backend.
 			def select(backends, queue:, request:)
diff --git a/lib/fantail/configuration.rb b/lib/fantail/configuration.rb
index 4e0187d..d2d8544 100644
--- a/lib/fantail/configuration.rb
+++ b/lib/fantail/configuration.rb
@@ -3,143 +3,110 @@
 # Released under the MIT License.
 # Copyright, 2026, by Samuel Williams.
 
-require_relative "balance"
-require_relative "loader"
+require_relative "queue"
 
 module Fantail
 	# Immutable request queue and admission configuration.
 	class Configuration
-		UNDEFINED = Object.new.freeze
-		
-		# Configuration for one class of requests.
-		class Queue
-			# @parameter name [Symbol | String] The stable queue name.
-			def initialize(name)
-				@name = name.to_sym
-				@matcher = nil
-				@eligibility = nil
-				@admission = nil
-				@balance = Balance::Spread.new
-				@depth_limit = nil
-				@wait_limit = nil
-				@shed_status = 429
-				@shed_headers = {}
-			end
-			
-			attr :name
-			attr :balance_policy
-			attr :shed_status
-			attr :shed_headers
-			
-			# Set the request classifier for this queue.
-			# @yields {|request| ...} Whether a request belongs to this queue.
-			def match(&block)
-				raise ArgumentError, "A matcher block is required!" unless block
-				@matcher = block
+		# Builds configurations using the application DSL.
+		class Builder
+			# @parameter root [String] The root for relative configuration files.
+			def initialize(root = Dir.pwd)
+				@root = File.expand_path(root)
+				@queues = {}
+				@default_queue_name = nil
+				@pending_limit = nil
+				@permit_limit = 1
 			end
 			
-			# Restrict the backends which may serve this queue.
-			# @yields {|backend, request| ...} Whether the backend is eligible.
-			def eligible(&block)
-				raise ArgumentError, "An eligibility block is required!" unless block
-				@eligibility = block
+			# @attribute [String] The root for relative configuration files.
+			attr :root
+			
+			# Evaluate a trusted configuration file using this builder.
+			# @parameter path [String] The relative or absolute configuration path.
+			def load_file(path)
+				realpath = File.realpath(File.expand_path(path, @root))
+				root = @root
+				@root = File.dirname(realpath)
+				instance_eval(File.read(realpath), realpath)
+			ensure
+				@root = root if root
 			end
 			
-			# Set an application admission policy.
-			# @parameter policy [#admit? | #call | Nil] The admission policy object.
-			# @yields {|request, queue:, pending:| ...} Whether the request can wait.
-			def admit(policy = nil, &block)
-				@admission = policy || block
-				raise ArgumentError, "An admission policy is required!" unless @admission
-			end
-			
-			# Set the soft backend balance policy.
-			# @parameter policy [Symbol | #select] A built-in name or application policy.
-			# @parameter options [Hash] Options for a built-in policy.
-			def balance(policy, **options)
-				@balance = Balance.coerce(policy, **options)
-			end
-			
-			# Set or get the maximum number of requests waiting in this queue.
-			# @parameter value [Integer] The new limit when given.
-			# @returns [Integer | Nil] The configured limit.
-			def depth_limit(value = UNDEFINED)
-				return @depth_limit if value.equal?(UNDEFINED)
+			# Define a named request queue. Matchers are evaluated in definition order.
+			# @parameter name [Symbol | String] The stable queue name.
+			# @yields {|queue| ...} The queue builder, or evaluates the block as its DSL.
+			# @returns [Queue] The configured queue.
+			def queue(name, &block)
+				name = name.to_sym
+				raise ArgumentError, "Queue #{name.inspect} is already defined!" if @queues.key?(name)
 				
-				value = Integer(value)
-				raise ArgumentError, "Depth limit must not be negative!" if value.negative?
-				@depth_limit = value
-			end
-			
-			# Set or get the maximum time a request may wait for a permit.
-			# @parameter value [Numeric] The new limit in seconds when given.
-			# @returns [Float | Nil] The configured limit.
-			def wait_limit(value = UNDEFINED)
-				return @wait_limit if value.equal?(UNDEFINED)
+				builder = Queue::Builder.new(name)
+				if block
+					if block.arity.zero?
+						builder.instance_eval(&block)
+					else
+						block.call(builder)
+					end
+				end
 				
-				value = Float(value)
-				raise ArgumentError, "Wait limit must be positive!" unless value.positive?
-				@wait_limit = value
+				@queues[name] = builder.build
 			end
 			
-			# Configure the response used when admission is rejected.
-			# @parameter status [Integer] The HTTP response status.
-			# @parameter retry_after [Numeric | String | Nil] An optional Retry-After value.
-			# @parameter headers [Hash] Additional response headers.
-			def shed(status: 429, retry_after: nil, headers: {})
-				@shed_status = Integer(status)
-				@shed_headers = headers.transform_keys(&:to_s)
-				@shed_headers["retry-after"] = retry_after.to_s if retry_after
+			# Select the fallback queue for unmatched requests.
+			# @parameter name [Symbol | String] A defined queue name.
+			def default_queue(name)
+				@default_queue_name = name.to_sym
 			end
 			
-			# @parameter request [Protocol::HTTP::Request] The request to classify.
-			# @returns [Boolean | Nil] Whether the request matches this queue.
-			def match?(request)
-				@matcher&.call(request)
+			# Set the global pending request limit.
+			# @parameter value [Integer] The maximum number of pending requests.
+			def pending_limit(value)
+				value = Integer(value)
+				raise ArgumentError, "Pending limit must not be negative!" if value.negative?
+				@pending_limit = value
 			end
 			
-			# @returns [Boolean] Whether a backend may serve the request.
-			def eligible?(backend, request)
-				!@eligibility || @eligibility.call(backend, request)
+			# Set the number of processing permits provided by each worker.
+			# @parameter value [Integer] The number of permits per worker.
+			def permit_limit(value)
+				value = Integer(value)
+				raise ArgumentError, "Permit limit must be positive!" unless value.positive?
+				@permit_limit = value
 			end
 			
-			# @returns [Boolean] Whether a request may enter the pending queue.
-			def admit?(request, pending:)
-				return true unless @admission
+			# Validate and build the immutable configuration.
+			# @returns [Configuration] The configured request queues.
+			def build
+				raise ArgumentError, "At least one queue must be defined!" if @queues.empty?
+				default_queue_name = @default_queue_name || @queues.keys.first
+				raise ArgumentError, "Default queue #{default_queue_name.inspect} is not defined!" unless @queues.key?(default_queue_name)
 				
-				if @admission.respond_to?(:admit?)
-					@admission.admit?(request, queue: self, pending: pending)
-				else
-					@admission.call(request, queue: self, pending: pending)
-				end
-			end
-			
-			# Validate and freeze this queue definition.
-			# @returns [Queue] The finalized queue.
-			def finalize
-				@balance_policy = @balance
-				@shed_headers.freeze
-				freeze
+				Configuration.new(
+					queues: @queues,
+					default_queue_name: default_queue_name,
+					pending_limit: @pending_limit,
+					permit_limit: @permit_limit,
+				)
 			end
 		end
 		
-		# Build and finalize a configuration using a scoped loader.
+		# Build a configuration using a scoped builder.
 		# @parameter root [String] The root for relative configuration files.
-		# @yields {|loader| ...} The configuration loader.
+		# @yields {|builder| ...} The configuration builder.
 		# @returns [Configuration] The immutable configuration.
 		def self.build(root: Dir.pwd, &block)
-			configuration = new
-			loader = Loader.new(configuration, root)
+			builder = Builder.new(root)
 			
 			if block
 				if block.arity.zero?
-					loader.instance_eval(&block)
+					builder.instance_eval(&block)
 				else
-					block.call(loader)
+					block.call(builder)
 				end
 			end
 			
-			configuration.finalize
+			builder.build
 		end
 		
 		# @returns [Configuration] A single-queue, single-permit configuration.
@@ -150,66 +117,31 @@ def self.default
 			end
 		end
 		
-		# Load and finalize trusted application configuration files.
+		# Load and build trusted application configuration files.
 		# @parameter paths [String | Array(String)] The configuration files to load.
 		# @returns [Configuration] The immutable configuration.
 		def self.load(paths)
-			configuration = new
-			Array(paths).each{|path| configuration.load_file(path)}
-			configuration.finalize
+			builder = Builder.new
+			Array(paths).each{|path| builder.load_file(path)}
+			builder.build
 		end
 		
-		# Initialize an empty mutable configuration builder.
-		def initialize
-			@queues = {}
-			@default_queue = nil
-			@pending_limit = nil
-			@permit_limit = 1
+		# @parameter queues [Hash(Symbol, Queue)] The configured queues.
+		# @parameter default_queue_name [Symbol] The fallback queue name.
+		# @parameter pending_limit [Integer | Nil] The global pending request limit.
+		# @parameter permit_limit [Integer] The processing permits per worker.
+		def initialize(queues:, default_queue_name:, pending_limit:, permit_limit:)
+			@queues = queues.dup.freeze
+			@default_queue_name = default_queue_name
+			@pending_limit = pending_limit
+			@permit_limit = permit_limit
+			freeze
 		end
 		
 		attr :queues
 		attr :default_queue_name
-		
-		# Load a trusted configuration file into this configuration.
-		# @parameter path [String] The configuration file to load.
-		def load_file(path)
-			Loader.load_file(self, path)
-		end
-		
-		# Define a named request queue. Matchers are evaluated in definition order.
-		def queue(name)
-			name = name.to_sym
-			raise ArgumentError, "Queue #{name.inspect} is already defined!" if @queues.key?(name)
-			
-			queue = Queue.new(name)
-			yield queue if block_given?
-			@queues[name] = queue
-			queue
-		end
-		
-		# Select the fallback queue for unmatched requests.
-		# @parameter name [Symbol | String] A previously or subsequently defined queue.
-		def default_queue(name)
-			@default_queue = name.to_sym
-		end
-		
-		# Set or get the global pending request limit.
-		def pending_limit(value = UNDEFINED)
-			return @pending_limit if value.equal?(UNDEFINED)
-			
-			value = Integer(value)
-			raise ArgumentError, "Pending limit must not be negative!" if value.negative?
-			@pending_limit = value
-		end
-		
-		# Set or get the number of processing permits provided by each worker.
-		def permit_limit(value = UNDEFINED)
-			return @permit_limit if value.equal?(UNDEFINED)
-			
-			value = Integer(value)
-			raise ArgumentError, "Permit limit must be positive!" unless value.positive?
-			@permit_limit = value
-		end
+		attr :pending_limit
+		attr :permit_limit
 		
 		# Classify a request using matchers in definition order.
 		# @returns [Queue] The matching or default queue.
@@ -218,20 +150,7 @@ def classify(request)
 				return queue if queue.match?(request)
 			end
 			
-			@queues.fetch(@default_queue)
-		end
-		
-		# Validate and freeze the complete configuration.
-		# @returns [Configuration] The finalized configuration.
-		def finalize
-			raise ArgumentError, "At least one queue must be defined!" if @queues.empty?
-			@default_queue ||= @queues.keys.first
-			raise ArgumentError, "Default queue #{@default_queue.inspect} is not defined!" unless @queues.key?(@default_queue)
-			
-			@queues.each_value(&:finalize)
-			@queues.freeze
-			@default_queue_name = @default_queue
-			freeze
+			@queues.fetch(@default_queue_name)
 		end
 	end
 end
diff --git a/lib/fantail/loader.rb b/lib/fantail/loader.rb
deleted file mode 100644
index f769c6a..0000000
--- a/lib/fantail/loader.rb
+++ /dev/null
@@ -1,73 +0,0 @@
-# frozen_string_literal: true
-
-# Released under the MIT License.
-# Copyright, 2026, by Samuel Williams.
-
-module Fantail
-	# The domain-specific language for building and loading configuration.
-	class Loader
-		# Initialize a loader attached to a configuration.
-		# @parameter configuration [Configuration] The configuration being built.
-		# @parameter root [String | Nil] The root for relative configuration files.
-		def initialize(configuration, root = nil)
-			@configuration = configuration
-			@root = root
-		end
-		
-		# @attribute [String | Nil] The root for relative configuration files.
-		attr :root
-		
-		# Evaluate a configuration file using a scoped loader.
-		# @parameter configuration [Configuration] The configuration being built.
-		# @parameter path [String] The configuration file to load.
-		def self.load_file(configuration, path)
-			realpath = File.realpath(path)
-			loader = new(configuration, File.dirname(realpath))
-			
-			if Module.method_defined?(:set_temporary_name)
-				loader.singleton_class.set_temporary_name("#{self}[#{path.inspect}]")
-			end
-			
-			loader.instance_eval(File.read(realpath), realpath)
-		end
-		
-		# Load another configuration file relative to this loader.
-		# @parameter path [String] The relative or absolute configuration path.
-		def load_file(path)
-			self.class.load_file(@configuration, File.expand_path(path, @root))
-		end
-		
-		# Define a named request queue.
-		# @parameter name [Symbol | String] The stable queue name.
-		# @yields {|queue| ...} The queue configuration, or evaluates the block as its DSL.
-		def queue(name, &block)
-			return @configuration.queue(name) unless block
-			
-			@configuration.queue(name) do |queue|
-				if block.arity.zero?
-					queue.instance_eval(&block)
-				else
-					block.call(queue)
-				end
-			end
-		end
-		
-		# Select the fallback queue for unmatched requests.
-		# @parameter name [Symbol | String] A defined queue name.
-		def default_queue(name)
-			@configuration.default_queue(name)
-		end
-		
-		# Set the global pending request limit.
-		# @parameter value [Integer] The maximum number of pending requests.
-		def pending_limit(value)
-			@configuration.pending_limit(value)
-		end
-		
-		# Set the number of processing permits provided by each worker.
-		# @parameter value [Integer] The number of permits per worker.
-		def permit_limit(value)
-			@configuration.permit_limit(value)
-		end
-	end
-end
diff --git a/lib/fantail/queue.rb b/lib/fantail/queue.rb
new file mode 100644
index 0000000..b18406f
--- /dev/null
+++ b/lib/fantail/queue.rb
@@ -0,0 +1,149 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "balance"
+
+module Fantail
+	# Immutable policy for one class of requests.
+	class Queue
+		# Builds a queue policy using the configuration DSL.
+		class Builder
+			# @parameter name [Symbol | String] The stable queue name.
+			def initialize(name)
+				@name = name
+				@matcher = nil
+				@eligibility = nil
+				@admission = nil
+				@balance_policy = Balance::Spread.new
+				@depth_limit = nil
+				@wait_limit = nil
+				@shed_status = 429
+				@shed_headers = {}
+			end
+			
+			# Set the request classifier for this queue.
+			# @yields {|request| ...} Whether a request belongs to this queue.
+			def match(&block)
+				raise ArgumentError, "A matcher block is required!" unless block
+				@matcher = block
+			end
+			
+			# Restrict the backends which may serve this queue.
+			# @yields {|backend, request| ...} Whether the backend is eligible.
+			def eligible(&block)
+				raise ArgumentError, "An eligibility block is required!" unless block
+				@eligibility = block
+			end
+			
+			# Set an application admission policy.
+			# @parameter policy [#admit? | #call | Nil] The admission policy object.
+			# @yields {|request, queue:, pending:| ...} Whether the request can wait.
+			def admit(policy = nil, &block)
+				@admission = policy || block
+				raise ArgumentError, "An admission policy is required!" unless @admission
+			end
+			
+			# Set the soft backend balance policy.
+			# @parameter policy [Symbol | #select] A built-in name or application policy.
+			# @parameter options [Hash] Options for a built-in policy.
+			def balance(policy, **options)
+				@balance_policy = Balance.coerce(policy, **options)
+			end
+			
+			# Set the maximum number of requests waiting in this queue.
+			# @parameter value [Integer] The maximum queue depth.
+			def depth_limit(value)
+				value = Integer(value)
+				raise ArgumentError, "Depth limit must not be negative!" if value.negative?
+				@depth_limit = value
+			end
+			
+			# Set the maximum time a request may wait for a permit.
+			# @parameter value [Numeric] The maximum wait in seconds.
+			def wait_limit(value)
+				value = Float(value)
+				raise ArgumentError, "Wait limit must be positive!" unless value.positive?
+				@wait_limit = value
+			end
+			
+			# Configure the response used when admission is rejected.
+			# @parameter status [Integer] The HTTP response status.
+			# @parameter retry_after [Numeric | String | Nil] An optional Retry-After value.
+			# @parameter headers [Hash] Additional response headers.
+			def shed(status: 429, retry_after: nil, headers: {})
+				@shed_status = Integer(status)
+				@shed_headers = headers.transform_keys(&:to_s)
+				@shed_headers["retry-after"] = retry_after.to_s if retry_after
+			end
+			
+			# Build the immutable queue policy.
+			# @returns [Queue] The configured queue.
+			def build
+				Queue.new(
+					@name,
+					matcher: @matcher,
+					eligibility: @eligibility,
+					admission: @admission,
+					balance_policy: @balance_policy,
+					depth_limit: @depth_limit,
+					wait_limit: @wait_limit,
+					shed_status: @shed_status,
+					shed_headers: @shed_headers,
+				)
+			end
+		end
+		
+		# @parameter name [Symbol | String] The stable queue name.
+		# @parameter matcher [Proc | Nil] The request classifier.
+		# @parameter eligibility [Proc | Nil] The backend eligibility policy.
+		# @parameter admission [#admit? | #call | Nil] The queue admission policy.
+		# @parameter balance_policy [#select] The backend balance policy.
+		# @parameter depth_limit [Integer | Nil] The maximum queue depth.
+		# @parameter wait_limit [Float | Nil] The maximum queue wait.
+		# @parameter shed_status [Integer] The rejection response status.
+		# @parameter shed_headers [Hash] The rejection response headers.
+		def initialize(name, matcher:, eligibility:, admission:, balance_policy:, depth_limit:, wait_limit:, shed_status:, shed_headers:)
+			@name = name.to_sym
+			@matcher = matcher
+			@eligibility = eligibility
+			@admission = admission
+			@balance_policy = balance_policy
+			@depth_limit = depth_limit
+			@wait_limit = wait_limit
+			@shed_status = shed_status
+			@shed_headers = shed_headers.freeze
+			freeze
+		end
+		
+		attr :name
+		attr :balance_policy
+		attr :depth_limit
+		attr :wait_limit
+		attr :shed_status
+		attr :shed_headers
+		
+		# @parameter request [Protocol::HTTP::Request] The request to classify.
+		# @returns [Boolean | Nil] Whether the request matches this queue.
+		def match?(request)
+			@matcher&.call(request)
+		end
+		
+		# @returns [Boolean] Whether a backend may serve the request.
+		def eligible?(backend, request)
+			!@eligibility || @eligibility.call(backend, request)
+		end
+		
+		# @returns [Boolean] Whether a request may enter the pending queue.
+		def admit?(request, pending:)
+			return true unless @admission
+			
+			if @admission.respond_to?(:admit?)
+				@admission.admit?(request, queue: self, pending: pending)
+			else
+				@admission.call(request, queue: self, pending: pending)
+			end
+		end
+	end
+end
diff --git a/lib/fantail/scheduler.rb b/lib/fantail/scheduler.rb
index d20a4cb..fe6810c 100644
--- a/lib/fantail/scheduler.rb
+++ b/lib/fantail/scheduler.rb
@@ -40,7 +40,7 @@ def release
 		
 		# A queue admission rejection.
 		class Rejection
-			# @parameter queue [Configuration::Queue] The queue which rejected admission.
+			# @parameter queue [Queue] The queue which rejected admission.
 			def initialize(queue)
 				@queue = queue
 			end
diff --git a/test/fantail/configuration.rb b/test/fantail/configuration.rb
index f1bcc63..78984bd 100644
--- a/test/fantail/configuration.rb
+++ b/test/fantail/configuration.rb
@@ -62,11 +62,18 @@
 		end
 	end
 	
-	it "builds configuration with an explicit loader" do
-		configuration = subject.build do |loader|
-			loader.queue(:default)
+	it "builds configuration with an explicit builder" do
+		configuration_builder = nil
+		queue_builder = nil
+		
+		configuration = subject.build do |builder|
+			configuration_builder = builder
+			builder.queue(:default){|builder| queue_builder = builder}
 		end
 		
+		expect(configuration_builder).to be_a(Fantail::Configuration::Builder)
+		expect(queue_builder).to be_a(Fantail::Queue::Builder)
+		expect(configuration.queues.fetch(:default)).to be_a(Fantail::Queue)
 		expect(configuration.default_queue_name).to be == :default
 	end
 	

From 2eb8e2e542ffe3080d5ff8b5f23a032f718ac747 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Thu, 3 Sep 2026 11:08:53 +1200
Subject: [PATCH 7/8] Finalize configuration in builders

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 lib/fantail/configuration.rb | 7 +++----
 lib/fantail/queue.rb         | 7 +++----
 2 files changed, 6 insertions(+), 8 deletions(-)

diff --git a/lib/fantail/configuration.rb b/lib/fantail/configuration.rb
index d2d8544..21456be 100644
--- a/lib/fantail/configuration.rb
+++ b/lib/fantail/configuration.rb
@@ -83,11 +83,11 @@ def build
 				raise ArgumentError, "Default queue #{default_queue_name.inspect} is not defined!" unless @queues.key?(default_queue_name)
 				
 				Configuration.new(
-					queues: @queues,
+					queues: @queues.dup.freeze,
 					default_queue_name: default_queue_name,
 					pending_limit: @pending_limit,
 					permit_limit: @permit_limit,
-				)
+				).freeze
 			end
 		end
 		
@@ -131,11 +131,10 @@ def self.load(paths)
 		# @parameter pending_limit [Integer | Nil] The global pending request limit.
 		# @parameter permit_limit [Integer] The processing permits per worker.
 		def initialize(queues:, default_queue_name:, pending_limit:, permit_limit:)
-			@queues = queues.dup.freeze
+			@queues = queues
 			@default_queue_name = default_queue_name
 			@pending_limit = pending_limit
 			@permit_limit = permit_limit
-			freeze
 		end
 		
 		attr :queues
diff --git a/lib/fantail/queue.rb b/lib/fantail/queue.rb
index b18406f..ccd94d5 100644
--- a/lib/fantail/queue.rb
+++ b/lib/fantail/queue.rb
@@ -90,8 +90,8 @@ def build
 					depth_limit: @depth_limit,
 					wait_limit: @wait_limit,
 					shed_status: @shed_status,
-					shed_headers: @shed_headers,
-				)
+					shed_headers: @shed_headers.dup.freeze,
+				).freeze
 			end
 		end
 		
@@ -113,8 +113,7 @@ def initialize(name, matcher:, eligibility:, admission:, balance_policy:, depth_
 			@depth_limit = depth_limit
 			@wait_limit = wait_limit
 			@shed_status = shed_status
-			@shed_headers = shed_headers.freeze
-			freeze
+			@shed_headers = shed_headers
 		end
 		
 		attr :name

From 7d2f48e09196005a44e5c1e2d87a09ff695e7778 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@shopify.com>
Date: Thu, 3 Sep 2026 11:14:33 +1200
Subject: [PATCH 8/8] Make scheduler the sole permit owner

Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da
---
 lib/fantail/registry.rb  | 31 +------------------------------
 test/fantail/registry.rb | 27 ++++-----------------------
 2 files changed, 5 insertions(+), 53 deletions(-)

diff --git a/lib/fantail/registry.rb b/lib/fantail/registry.rb
index 4dce9ca..7f6537d 100644
--- a/lib/fantail/registry.rb
+++ b/lib/fantail/registry.rb
@@ -4,7 +4,6 @@
 # Copyright, 2026, by Samuel Williams.
 
 require "async/bus/controller"
-require "async/queue"
 
 require_relative "endpoint"
 require_relative "backend"
@@ -23,8 +22,6 @@ def initialize(exchange_limit: 8, permit_limit: 1, backend_factory: nil)
 			
 			@guard = Thread::Mutex.new
 			@backends = {}
-			@notifications = Async::Queue.new
-			@waiting = 0
 			@available = nil
 			@closed = false
 		end
@@ -77,28 +74,6 @@ def update(upserted, removed)
 			self.size
 		end
 		
-		# Acquire the next backend with processing capacity.
-		# @returns [Backend | Nil] An admitted backend, or nil if no endpoints exist.
-		def acquire
-			waiting = false
-			
-			loop do
-				return nil if self.empty?
-				backend = backends.find{|candidate| candidate.reserve}
-				return backend if backend
-				
-				unless waiting
-					@guard.synchronize{@waiting += 1}
-					waiting = true
-					next
-				end
-				
-				return nil unless @notifications.dequeue
-			end
-		ensure
-			@guard.synchronize{@waiting -= 1} if waiting
-		end
-		
 		# @returns [Array(Backend)] A snapshot of active backends.
 		def backends
 			@guard.synchronize{@backends.values.dup}
@@ -140,7 +115,6 @@ def close
 				@backends.values.tap{@backends = {}}
 			end
 			
-			@notifications.close
 			backends.each(&:retire)
 		end
 		
@@ -151,11 +125,8 @@ def offer(_backend)
 		end
 		
 		def notify_available
-			available, waiting = @guard.synchronize{[@available, @waiting]}
-			@notifications.enqueue(true) if waiting.positive?
+			available = @guard.synchronize{@available}
 			available&.call
-		rescue Async::Queue::ClosedError
-			# The registry is already shutting down.
 		end
 		
 		def make_backend(endpoint, exchange_limit, permit_limit, available)
diff --git a/test/fantail/registry.rb b/test/fantail/registry.rb
index 19817f3..f784e18 100644
--- a/test/fantail/registry.rb
+++ b/test/fantail/registry.rb
@@ -46,7 +46,8 @@
 	
 	it "drains a retired backend before closing it" do
 		registry.replace([{name: "worker-1", url: "http://127.0.0.1:9301"}])
-		backend = registry.acquire
+		backend = registry["worker-1"]
+		backend.reserve
 		client = @clients.fetch("worker-1")
 		
 		backend.processed
@@ -62,7 +63,8 @@
 	
 	it "only permits one request to be processed at a time" do
 		registry.replace([{name: "worker-1", url: "http://127.0.0.1:9301"}])
-		backend = registry.acquire
+		backend = registry["worker-1"]
+		backend.reserve
 		
 		expect(backend.name).to be == "worker-1"
 		expect(backend).to be(:processing?)
@@ -72,24 +74,6 @@
 		expect(backend).not.to be(:processing?)
 	end
 	
-	it "wakes an acquisition when a permit is released" do
-		registry.replace([{name: "worker-1", url: "http://127.0.0.1:9301"}])
-		first = registry.acquire
-		second_task = Async{registry.acquire}
-		Fiber.scheduler.yield
-		
-		expect(second_task).not.to be(:finished?)
-		first.failed
-		first = nil
-		
-		second = second_task.wait
-		expect(second.name).to be == "worker-1"
-	ensure
-		first&.failed
-		second&.failed
-		second_task&.stop
-	end
-	
 	it "supports multiple processing permits" do
 		registry.close
 		registry = make_registry(permit_limit: 2)
@@ -107,7 +91,4 @@
 		registry&.close
 	end
 	
-	it "returns nil when no endpoints exist" do
-		expect(registry.acquire).to be_nil
-	end
 end