Add configurable request queues and load shedding

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
+~~~

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
+~~~

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"

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

lib/fantail/balance.rb

@@ -0,0 +1,60 @@
+# 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
+			# 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
+		end
+
+		# 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|
+					[-backend.processing_for(affinity), backend.processing, backend.name]
+				end
+			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
+				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