From 50b46de473623531c1d16a248cdb48aee5461652 Mon Sep 17 00:00:00 2001 From: Cristian Girlea Date: Wed, 21 Oct 2026 02:04:30 +0300 Subject: [PATCH 1/2] feat: send Retry-After on 429 and expose it to the PHP client and the MCP server --- CHANGELOG.md | 4 +++ docs/integration.md | 20 +++++++++++- docs/protocol.md | 4 ++- docs/testing.md | 6 ++-- mcp_server/bridge_mcp/protocol.py | 13 ++++++-- mcp_server/bridge_mcp/tools.py | 14 +++++--- src/BridgeException.php | 2 ++ src/Client.php | 36 +++++++++++++++++++-- tests/mcp/test_protocol.py | 54 +++++++++++++++++++++++++++++++ tests/mcp/test_tools.py | 52 ++++++++++++++++++++++++++++- tests/php/integration.php | 16 +++++++++ tests/php/run.php | 54 +++++++++++++++++++++++++++++++ tests/python/test_jobs.py | 31 +++++++++++++++++- tests/python/test_server.py | 43 ++++++++++++++++++++++++ worker/ai_bridge/jobs.py | 14 ++++++-- worker/ai_bridge/server.py | 13 +++++--- 16 files changed, 353 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d68609..fac44a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ All notable changes to this project are documented in this file. The format foll ## [Unreleased] +### Added + +- A 429 from the worker carries `Retry-After` in whole seconds: when the first retained job expires, or 1 when every slot holds an unfinished job. The PHP client exposes it as `BridgeException::$retryAfterSeconds` (and refuses a response header block over 16 KiB as `invalid_response`) without retrying by itself; the MCP server waits for it within a call's budget instead of guessing. + ### Changed - The dist archive now ships only `src/`, `composer.json`, `LICENSE`, `README.md`, `CHANGELOG.md` and `docs/`. diff --git a/docs/integration.md b/docs/integration.md index fb8b908..6cdf7fa 100644 --- a/docs/integration.md +++ b/docs/integration.md @@ -162,10 +162,28 @@ PHP objects reused by a long-lived worker must not retain per-user inputs, authe | `transport_error` during submission | Response unknown; a job may have been created | Do not automatically retry: no idempotency keys exist | | `http_error`, status 401 | Service credentials rejected | Fix server-side configuration, do not forward service secrets | | `http_error`, status 400/413 | Invalid or oversized request | Reject/correct input | -| `http_error`, status 429 | Total queued/running/retained capacity exhausted | Apply backpressure; completed retained jobs also occupy capacity | +| `http_error`, status 429 | Total queued/running/retained capacity exhausted; no job was created | Apply backpressure; resubmit no sooner than `retryAfterSeconds` (below). Completed retained jobs also occupy capacity | | `invalid_response` | Peer violated the client contract | Treat as an integration failure, not a valid task result | | Terminal `failed` or `timed_out` | Task did not succeed | Report a safe failure; retries require an application policy | +When the worker is full, `submit()` throws a `BridgeException` with `httpStatus` 429 and, from the worker's `Retry-After` header, `retryAfterSeconds`: when capacity is expected back. The client never retries by itself. Unlike a `transport_error`, a 429 means no job exists, so scheduling the submission again is safe; how is the application's choice, for instance a delayed message on its own queue: + +```php +try { + $job = $client->submitRerank($query, $documents); +} catch (\PhpAiBridge\BridgeException $error) { + if ($error->httpStatus !== 429) { + throw $error; + } + // Null when the header was missing or malformed (a proxy answered, say): fall back to your own delay. + $delaySeconds = $error->retryAfterSeconds ?? 5; + $this->queue->dispatchLater(new SubmitRerank($applicationJobId), $delaySeconds); + return new Response('', 503, ['Retry-After' => (string) $delaySeconds]); +} +``` + +`dispatchLater()` and `SubmitRerank` stand for your framework's delayed dispatch (Symfony Messenger's `DelayStamp`, Laravel's `->delay()`); cap the attempts, since a worker kept full by other callers can refuse again. + The current default limits are two warm task processes, 128 total job slots and 300 seconds of terminal retention. These are settings, not measured throughput or a production SLA. The model profile keeps one task process, which loads each model on its first job of that kind and keeps it. ## Before shipping a framework integration diff --git a/docs/protocol.md b/docs/protocol.md index f1c9688..890cf53 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -29,8 +29,10 @@ Submit `{"task":"redact","input":{"text":"...","entities":["PER"],"min_score":0. Failures use generic error objects with `code` and `message`. Task errors include `task_failed`, `worker_crashed`, `worker_start_failed` and `deadline_exceeded`. Internal exception text is not sent to clients. HTTP contract failures return 400, missing auth 401, missing routes/jobs 404, a connection idle for five seconds while its body is read 408 (a stall while reading headers closes the connection without a response), oversized bodies 413 and capacity exhaustion 429. +Every 429 carries `Retry-After` in whole delay seconds (an integer of at least 1, never an HTTP date): when every slot is taken and some jobs have finished, the time until the first of them leaves retention, rounded up; when every slot holds an unfinished job, or the worker is shutting down, 1. A 429 created no job, so resubmitting after that delay cannot duplicate work. The PHP client exposes the value as `BridgeException::$retryAfterSeconds` and the MCP server waits for it within a call's budget; both read one header of 1 to 6 digits and ignore anything else, a repeated header included. + The default capacity is 128 total queued, running and retained jobs, concurrency is 2, and terminal retention is 300 seconds. A full result cache rejects submissions until entries expire; it does not silently evict a result early. Configure these with `BRIDGE_CAPACITY`, `BRIDGE_CONCURRENCY`, and `BRIDGE_RETENTION_SECONDS`. The worker keeps as many child processes as its concurrency and replaces a child after `BRIDGE_WORKER_MAX_JOBS` jobs (default 1000, between 1 and 1000000), as well as after a cancellation, a deadline or a crash. A numeric setting that does not parse stops the worker at start with a message naming its variable; one outside its range is refused before any child starts. -The PHP client uses a fresh cURL handle per call, refuses redirects, ignores proxy environment variables, caps response bytes, and verifies that GET/cancel responses match the requested ID. Configure its request timeout separately from job execution and local waiting deadlines. There are no transparent submission retries. The MCP server in `mcp_server/` is a second, independent client with the same checks; unlike the PHP client it cancels a job whose local wait expires, because no later request will collect it. +The PHP client uses a fresh cURL handle per call, refuses redirects, ignores proxy environment variables, caps response bytes, and verifies that GET/cancel responses match the requested ID. Configure its request timeout separately from job execution and local waiting deadlines. There are no transparent submission retries, a 429 included. The MCP server in `mcp_server/` is a second, independent client with the same checks; unlike the PHP client it cancels a job whose local wait expires, because no later request will collect it. Adding a task requires trusted source changes to `worker/ai_bridge/tasks.py`: an explicit name, input validation, bounded output and execution logic. Add a PHP result type where useful, plus successful, malformed-input and lifecycle tests. Task imports, file paths and command strings must not come from request data. diff --git a/docs/testing.md b/docs/testing.md index 7ea9cb2..475f64c 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -48,9 +48,9 @@ PHP branch support was verified locally on 2026-09-17 on PHP 8.2.33, 8.3.33, 8.4 | Layer | Scope and counting | | --- | --- | -| Python worker | 170 tests: lifecycle, validation and HTTP, process crashes, cancellation, timeouts, retention, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample and scoring, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values), and the model-name table every result reports from | -| MCP server | 119 tests, offline with hash-pinned wheels: the protocol client against an in-process worker, every tool through the SDK's in-memory client with test tasks enabled, the entry point as a subprocess, result validators on synthetic data, the index builder (chunking, the format and each corruption it refuses, building against the worker, caching, its command line) `bridge_search` (the model gate, path confinement, reloads, rerank on and off), the reachability of every answer in the LLM evaluation, that a default search sees only part of its corpus and that its two-step question takes two steps, that the demo worker reaches none of the semantic evaluation's answers, that a tool waits out a full worker, before each of a search's jobs and within one budget, and says so when it stays full, the evaluation runner's command, isolation checks, scoring and per-backend launcher and index check without starting `claude`, a check that the server's model names match the worker's, that every path parameter says where it resolves and every position in a result says what it counts, and a check that every timing claim in a tool description is a measured median | -| PHP contracts + HTTP | 134 checks per PHP version; this **includes** the 101 contract checks, not 134 + 101 | +| Python worker | 177 tests: lifecycle, validation and HTTP, process crashes, cancellation, timeouts, retention, the Retry-After a full or closing store gives and its header on every 429, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample and scoring, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values), and the model-name table every result reports from | +| MCP server | 125 tests, offline with hash-pinned wheels: the protocol client against an in-process worker, every tool through the SDK's in-memory client with test tasks enabled, the entry point as a subprocess, result validators on synthetic data, the index builder (chunking, the format and each corruption it refuses, building against the worker, caching, its command line) `bridge_search` (the model gate, path confinement, reloads, rerank on and off), the reachability of every answer in the LLM evaluation, that a default search sees only part of its corpus and that its two-step question takes two steps, that the demo worker reaches none of the semantic evaluation's answers, that a tool waits out a full worker for its Retry-After within the call's budget or with its own backoff without one, before each of a search's jobs and within one budget, and says so when it stays full, the evaluation runner's command, isolation checks, scoring and per-backend launcher and index check without starting `claude`, a check that the server's model names match the worker's, that every path parameter says where it resolves and every position in a result says what it counts, and a check that every timing claim in a tool description is a measured median | +| PHP contracts + HTTP | 144 checks per PHP version, a full fault-worker's Retry-After read through the client among them; this **includes** the 109 contract checks (Retry-After from a loopback peer, present, absent, malformed and repeated, and a header block over 16 KiB refused), not 144 + 109 | | FrankenPHP | 12 jobs submitted by six concurrent test threads through a reused PHP worker, a 512-document `top_k` request, embedding and redaction round trips, and invalid-input, missing-job and multibyte boundary checks | | PHP syntax | All 13 PHP source/example/test files | | Consumer install | `composer validate --strict` on the library, then a throwaway consumer project in tmpfs, with Packagist disabled and the platform pinned to PHP 8.2.0, installs its `git archive` zip from an `artifact` repository (or from a path repository in a git worktree whose git directory is outside the mount; the archive is built from the committed `HEAD`, so an uncommitted change is not what it tests), checks that `vendor/` holds `src/Client.php` and `docs/` and not `worker/`, `tests/` or `docker/`, and autoloads `PhpAiBridge\Client`; runs once, in the MCP server job, with no network | diff --git a/mcp_server/bridge_mcp/protocol.py b/mcp_server/bridge_mcp/protocol.py index 64b3708..30a303c 100644 --- a/mcp_server/bridge_mcp/protocol.py +++ b/mcp_server/bridge_mcp/protocol.py @@ -11,14 +11,17 @@ STATUSES = TERMINAL | {"queued", "running"} JOB_ID = re.compile(r"[a-f0-9]{32}") ERROR_CODE = re.compile(r"[a-z_]{1,40}") +# Delay seconds only, as the PHP client reads them; an HTTP date would need the two clocks to agree. +RETRY_AFTER = re.compile(r"[ \t]*([0-9]{1,6})[ \t]*") class BridgeError(Exception): """A protocol, transport or task failure. Messages never contain the token or worker internals.""" - def __init__(self, message: str, code: str, status: int | None = None, job_id: str | None = None): + def __init__(self, message: str, code: str, status: int | None = None, job_id: str | None = None, + retry_after: int | None = None): super().__init__(message) - self.code, self.status, self.job_id = code, status, job_id + self.code, self.status, self.job_id, self.retry_after = code, status, job_id, retry_after class Bridge: @@ -135,6 +138,9 @@ def _request(self, method: str, path: str, body: dict | None = None) -> dict: connection.request(method, path, data, headers) response = connection.getresponse() status, content_type = response.status, response.getheader("Content-Type") or "" + # A repeated Retry-After is a conflict the caller cannot trust, so it counts as absent. + values = (response.headers.get_all("Retry-After") or []) if status == 429 else [] + retry_after = RETRY_AFTER.fullmatch(values[0]) if len(values) == 1 else None raw = response.read(MAX_BYTES + 1) except (OSError, http.client.HTTPException) as error: raise BridgeError("Bridge transport failed: " + type(error).__name__, "transport_error") from None @@ -154,7 +160,8 @@ def _request(self, method: str, path: str, body: dict | None = None) -> dict: # free text, and a proxy's HTML error page contributes nothing but its status. code = decoded.get("error", {}).get("code") if isinstance(decoded, dict) and isinstance(decoded.get("error"), dict) else None code = code if isinstance(code, str) and ERROR_CODE.fullmatch(code) else "http_error" - raise BridgeError(f"Bridge returned HTTP {status}: {code}", code, status) + raise BridgeError(f"Bridge returned HTTP {status}: {code}", code, status, + retry_after=int(retry_after[1]) if retry_after else None) if not is_json: raise BridgeError("Expected a JSON response", "invalid_response", status) if decoded is None: diff --git a/mcp_server/bridge_mcp/tools.py b/mcp_server/bridge_mcp/tools.py index cf857bc..a2c4a0f 100644 --- a/mcp_server/bridge_mcp/tools.py +++ b/mcp_server/bridge_mcp/tools.py @@ -32,7 +32,8 @@ MAX_FILE_BYTES = 4 * 1024 * 1024 SNIPPET_CHARACTERS = 160 # A full worker refuses submissions until finished jobs age out of its retention; a tool waits within its own call, -# pausing this long at first and doubling, and retries for the last time when the last amount of the call remains. +# for the worker's Retry-After or else this long at first and doubling, and retries for the last time when the last +# amount of the call remains. CAPACITY_PAUSE_S, CAPACITY_MAX_PAUSE_S, CAPACITY_LAST_S = 0.25, 5.0, 1.0 PAIRS = 10 MAX_SEARCH_TOP_K = 50 @@ -385,9 +386,15 @@ def on_progress(completed: int, total: int) -> None: except BridgeError as error: if error.code != "capacity_exceeded": raise ToolError(_explain(error)) from None + retry_after = error.retry_after # The last retry comes when the final CAPACITY_LAST_S of the call is left for the job itself. The jitter - # keeps calls refused together from retrying together. - wait = min(pause * random.uniform(0.5, 1.0), deadline - time.monotonic() - CAPACITY_LAST_S) + # keeps calls refused together from retrying together; a Retry-After of 0 still pauses, never spins. + if retry_after is None: + wanted = pause * random.uniform(0.5, 1.0) + pause = min(pause * 2, CAPACITY_MAX_PAUSE_S) + else: + wanted = max(retry_after, CAPACITY_PAUSE_S) + wait = min(wanted, deadline - time.monotonic() - CAPACITY_LAST_S) if wait <= 0: raise ToolError(f"The worker stayed at capacity (HTTP 429) for the whole call; the tool waited " f"{time.monotonic() - started:.1f} s of its {seconds:g} s. Finished jobs hold capacity " @@ -395,7 +402,6 @@ def on_progress(completed: int, total: int) -> None: # Nothing is sent to the host meanwhile: MCP logging is deprecated as of the 2026-07-28 specification, # and progress must only increase while each job's own starts at zero. await anyio.sleep(wait) - pause = min(pause * 2, CAPACITY_MAX_PAUSE_S) @mcp.tool( name="bridge_health", title="Bridge worker health", diff --git a/src/BridgeException.php b/src/BridgeException.php index fe3de3e..a8f7c00 100644 --- a/src/BridgeException.php +++ b/src/BridgeException.php @@ -10,6 +10,8 @@ public function __construct( string $message, public readonly string $errorCode = 'bridge_error', public readonly ?int $httpStatus = null, + /** On a 429, the worker's Retry-After in whole seconds; null when absent or not 1 to 6 digits. */ + public readonly ?int $retryAfterSeconds = null, ) { parent::__construct($message); } diff --git a/src/Client.php b/src/Client.php index bd8a5f3..d575105 100644 --- a/src/Client.php +++ b/src/Client.php @@ -37,7 +37,12 @@ public function __construct( $this->baseUrl = rtrim($baseUrl, '/'); } - /** @param array $input */ + /** + * A full worker refuses with HTTP 429 and says in retryAfterSeconds when capacity returns. The client never + * retries by itself: whether and when to resubmit is the application's decision. + * + * @param array $input + */ public function submit(string $task, array $input, int $timeoutMs = 30000): Job { if ($task === '' || strlen($task) > 128 || $timeoutMs < 100 || $timeoutMs > 300000) { @@ -210,6 +215,10 @@ private function request(string $method, string $path, ?array $body = null, ?int $handle = curl_init($this->baseUrl . $path); $response = ''; $tooLarge = false; + $headerBytes = 0; + $headersTooLarge = false; + $retryAfter = null; + $retryAfterSeen = false; $timeoutMs ??= $this->requestTimeoutMs; curl_setopt_array($handle, [ CURLOPT_CUSTOMREQUEST => $method, @@ -228,6 +237,26 @@ private function request(string $method, string $path, ?array $body = null, ?int $response .= $chunk; return strlen($chunk); }, + CURLOPT_HEADERFUNCTION => static function ($curl, string $line) use (&$headerBytes, &$headersTooLarge, &$retryAfter, &$retryAfterSeen): int { + // An interim response such as 100 Continue starts a new block; only the final one counts. + if (str_starts_with($line, 'HTTP/')) { + $headerBytes = 0; + $retryAfter = null; + $retryAfterSeen = false; + } + $headerBytes += strlen($line); + if ($headerBytes > 16384) { + $headersTooLarge = true; + return 0; + } + if (preg_match('/\Aretry-after:(.*)\z/is', $line, $match) === 1) { + // Delay seconds only: an HTTP date would need this host's clock to agree with the worker's. + // A repeated header is a conflict the caller cannot trust, so it counts as absent. + $retryAfter = !$retryAfterSeen && preg_match('/\A[ \t]*([0-9]{1,6})[ \t]*\r?\n?\z/', $match[1], $digits) === 1 ? (int) $digits[1] : null; + $retryAfterSeen = true; + } + return strlen($line); + }, ]); if ($json !== null) { curl_setopt($handle, CURLOPT_POSTFIELDS, $json); @@ -236,6 +265,9 @@ private function request(string $method, string $path, ?array $body = null, ?int $ok = curl_exec($handle); $status = (int) curl_getinfo($handle, CURLINFO_RESPONSE_CODE); $contentType = curl_getinfo($handle, CURLINFO_CONTENT_TYPE); + if ($headersTooLarge) { + throw new BridgeException('Response headers exceed 16384 bytes', 'invalid_response'); + } if ($tooLarge) { throw new BridgeException('Response exceeds 262144 bytes', 'invalid_response'); } @@ -249,7 +281,7 @@ private function request(string $method, string $path, ?array $body = null, ?int } if ($status < 200 || $status >= 300) { // Do not reflect server-supplied text that may contain secrets or markup. - throw new BridgeException('Bridge returned HTTP ' . $status, 'http_error', $status); + throw new BridgeException('Bridge returned HTTP ' . $status, 'http_error', $status, $status === 429 ? $retryAfter : null); } if (!is_string($contentType) || strtolower(trim(explode(';', $contentType)[0])) !== 'application/json') { throw new BridgeException('Expected a JSON response', 'invalid_response'); diff --git a/tests/mcp/test_protocol.py b/tests/mcp/test_protocol.py index 5d58bd9..d0bd8d3 100644 --- a/tests/mcp/test_protocol.py +++ b/tests/mcp/test_protocol.py @@ -99,6 +99,60 @@ def log_message(self, *_): proxy.server_close() self.assertEqual((caught.exception.code, caught.exception.status), ("http_error", 503)) + def refusal(self, *headers): + """What the client makes of a 429 that carries these headers.""" + class Full(BaseHTTPRequestHandler): + def do_GET(self): + body = b'{"error":{"code":"capacity_exceeded","message":"Worker capacity reached"}}' + self.send_response(429) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + for name, value in headers: + self.send_header(name, value) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_): + pass + + stub = HTTPServer(("127.0.0.1", 0), Full) + thread = threading.Thread(target=stub.serve_forever, daemon=True) + thread.start() + try: + with self.assertRaises(BridgeError) as caught: + Bridge("http://127.0.0.1:%d" % stub.server_address[1], TOKEN).health() + finally: + stub.shutdown() + thread.join() + stub.server_close() + return caught.exception + + def test_a_capacity_refusal_carries_the_retry_after_seconds(self): + error = self.refusal(("Retry-After", "7")) + self.assertEqual((error.status, error.code, error.retry_after), (429, "capacity_exceeded", 7)) + self.assertIsNone(self.refusal().retry_after) + self.assertEqual(self.refusal(("Retry-After", "0")).retry_after, 0) + for value in ["Wed, 21 Oct 2026 07:28:00 GMT", "1234567", "-1", "1.5", ""]: + with self.subTest(value=value): + self.assertIsNone(self.refusal(("Retry-After", value)).retry_after) + self.assertIsNone(self.refusal(("Retry-After", "7"), ("Retry-After", "3")).retry_after) + + def test_the_real_worker_sends_retry_after_on_a_full_store(self): + server = BridgeServer(("127.0.0.1", 0), TOKEN, Settings(capacity=1, test_tasks=True)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + bridge = Bridge("http://127.0.0.1:%d" % server.server_address[1], TOKEN) + bridge.submit("test.delay", {"seconds": 5}, 10000) + with self.assertRaises(BridgeError) as caught: + bridge.submit("test.delay", {"seconds": 0}, 10000) + finally: + server.shutdown() + thread.join() + server.server_close() + # The one slot holds a job still running, so capacity returns no sooner than the worker's short default. + self.assertEqual((caught.exception.status, caught.exception.retry_after), (429, 1)) + def test_task_failures_become_errors_with_the_worker_code(self): for task, code in [("test.crash", "worker_crashed"), ("test.error", "task_failed")]: with self.subTest(task=task), self.assertRaises(BridgeError) as caught: diff --git a/tests/mcp/test_tools.py b/tests/mcp/test_tools.py index 49d77d8..12776a8 100644 --- a/tests/mcp/test_tools.py +++ b/tests/mcp/test_tools.py @@ -4,6 +4,7 @@ import os import tempfile import threading +import time import unittest from pathlib import Path @@ -12,7 +13,7 @@ from ai_bridge.jobs import Settings from ai_bridge.server import BridgeServer -from bridge_mcp.protocol import Bridge +from bridge_mcp.protocol import Bridge, BridgeError from bridge_mcp.tools import BY_VALUE_DOCUMENTS, TOOL_NAMES, _rankings, _snippet, _spans, _vectors, build_server TOKEN = "test-only-bridge-token-never-use-in-production" @@ -229,6 +230,55 @@ async def test_rerank_pre_checks_include_the_query_in_the_budget(self): self.assertTrue(result.is_error) self.assertIn("200000", result.content[0].text) + def full_once(self, retry_after): + """The shared worker behind a bridge whose first job is refused for capacity with this Retry-After.""" + real = self.bridge + + class FullOnce: + calls = 0 + + def health(self): + return real.health() + + def run(self, *args, **kwargs): + FullOnce.calls += 1 + if FullOnce.calls == 1: + raise BridgeError("full", "capacity_exceeded", 429, retry_after=retry_after) + return real.run(*args, **kwargs) + + return FullOnce() + + async def test_a_full_worker_is_waited_for_as_long_as_its_retry_after_says(self): + server = build_server(self.full_once(1), backend="lexical", root=None, timeout_ms=10000) + started = time.monotonic() + result = await self.call("bridge_rerank", {"query": "needle", "documents": ["needle"]}, server=server) + self.assertFalse(result.is_error, result.content) + # The fallback's first pause is at most a quarter of a second, so a wait this long is the header's. + self.assertGreater(time.monotonic() - started, 0.9) + + async def test_without_retry_after_the_wait_is_the_tools_own_backoff(self): + server = build_server(self.full_once(None), backend="lexical", root=None, timeout_ms=10000) + started = time.monotonic() + result = await self.call("bridge_rerank", {"query": "needle", "documents": ["needle"]}, server=server) + self.assertFalse(result.is_error, result.content) + self.assertLess(time.monotonic() - started, 0.8) + + async def test_a_retry_after_of_zero_still_pauses_and_then_retries(self): + server = build_server(self.full_once(0), backend="lexical", root=None, timeout_ms=10000) + started = time.monotonic() + result = await self.call("bridge_rerank", {"query": "needle", "documents": ["needle"]}, server=server) + self.assertFalse(result.is_error, result.content) + self.assertGreater(time.monotonic() - started, 0.2) + + async def test_a_retry_after_beyond_the_call_is_bounded_by_its_budget(self): + server = build_server(self.full_once(60), backend="lexical", root=None, timeout_ms=3000) + started = time.monotonic() + result = await self.call("bridge_rerank", {"query": "needle", "documents": ["needle"]}, server=server) + # One wait to the last second of the call, then the job runs in what is left. + self.assertFalse(result.is_error, result.content) + self.assertGreater(time.monotonic() - started, 1.8) + self.assertLess(time.monotonic() - started, 3) + async def test_worker_failures_become_tool_errors_without_internals(self): broken = build_server(Bridge("http://127.0.0.1:1", TOKEN, request_timeout_s=0.5), backend="lexical", root=None, timeout_ms=10000) result = await self.call("bridge_health", {}, server=broken) diff --git a/tests/php/integration.php b/tests/php/integration.php index 739317a..daaf0fc 100644 --- a/tests/php/integration.php +++ b/tests/php/integration.php @@ -101,5 +101,21 @@ function ready(Client $client): void $job = $real->submitRedact($text); $redaction = RedactResult::fromJob($real->wait($job->id), $text); check($redaction->text === 'Mail me at [EMAIL] or on [PHONE] today' && count($redaction->spans) === 2, 'redaction round trip through the PHP client'); +// Last, since the fault-worker then stays full for its 300 s retention: its default capacity is 128 jobs, and the +// jobs above have finished, so the earliest of them to expire says when a slot returns. +$refusal = null; +for ($submitted = 0; $submitted <= 128 && $refusal === null; ++$submitted) { + try { + $client->submit('test.delay', ['seconds' => 0]); + } catch (BridgeException $error) { + $refusal = $error; + } +} +if ($refusal === null) { + throw new RuntimeException('Expected the fault-worker to fill up'); +} +check($refusal->httpStatus === 429, 'a full worker refuses with HTTP 429'); +check($refusal->retryAfterSeconds !== null && $refusal->retryAfterSeconds >= 1 && $refusal->retryAfterSeconds <= 300, + 'the real worker\'s Retry-After reaches the PHP client in whole seconds within its retention'); /** @var int $count counted by check() in run.php */ echo "PASS: $count total PHP checks including real HTTP failure cases\n"; diff --git a/tests/php/run.php b/tests/php/run.php index cab601e..dab18e0 100644 --- a/tests/php/run.php +++ b/tests/php/run.php @@ -181,4 +181,58 @@ function sample(): array $redacted['result']['text'] = $original; check(RedactResult::fromJob(Job::fromArray($redacted), $original)->spans === [], 'clean redact result'); +// A loopback peer in a child process answers each connection with the next canned response; no network is needed. +$refusal = static fn (int $status, string $headers): string => "HTTP/1.1 $status Refused\r\nContent-Type: application/json\r\n" + . "Content-Length: 2\r\nConnection: close\r\n" . $headers . "\r\n{}"; +$responses = [ + $refusal(429, "Retry-After: 7\r\n"), + $refusal(429, ''), + $refusal(429, "Retry-After: Wed, 21 Oct 2026 07:28:00 GMT\r\n"), + $refusal(429, "Retry-After: 1234567\r\n"), + $refusal(429, "Retry-After: -1\r\n"), + $refusal(429, "Retry-After: 7\r\nRetry-After: 3\r\n"), + $refusal(503, "Retry-After: 5\r\n"), + $refusal(429, str_repeat('X-Padding: ' . str_repeat('p', 100) . "\r\n", 160) . "Retry-After: 3\r\n"), +]; +$peer = proc_open([PHP_BINARY, '-r', ' + $responses = unserialize(base64_decode($argv[1])); + $server = stream_socket_server("tcp://127.0.0.1:0"); + echo parse_url("tcp://" . stream_socket_get_name($server, false), PHP_URL_PORT), "\n"; + foreach ($responses as $response) { + $connection = stream_socket_accept($server, 10); + $request = ""; + while (!str_contains($request, "\r\n\r\n") && ($chunk = fread($connection, 8192)) !== false && $chunk !== "") { + $request .= $chunk; + } + fwrite($connection, $response); + fclose($connection); + }', base64_encode(serialize($responses))], [1 => ['pipe', 'w']], $pipes); +if ($peer === false) { + throw new RuntimeException('The loopback peer did not start'); +} +try { + $peerClient = new Client('http://127.0.0.1:' . (int) fgets($pipes[1]), $token); + $refused = static function () use ($peerClient): BridgeException { + try { + $peerClient->get(str_repeat('a', 32)); + } catch (BridgeException $error) { + return $error; + } + throw new RuntimeException('Expected a refusal'); + }; + $error = $refused(); + check($error->httpStatus === 429 && $error->errorCode === 'http_error' && $error->retryAfterSeconds === 7, 'a 429 carries its Retry-After seconds'); + foreach (['absent', 'an HTTP date', 'seven digits', 'a sign', 'repeated'] as $form) { + check($refused()->retryAfterSeconds === null, 'a Retry-After that is ' . $form . ' is ignored'); + } + $error = $refused(); + check($error->httpStatus === 503 && $error->retryAfterSeconds === null, 'only a 429 reads Retry-After'); + $error = $refused(); + check($error->errorCode === 'invalid_response' && $error->retryAfterSeconds === null, 'a header block over 16 KiB is refused'); +} finally { + fclose($pipes[1]); + proc_terminate($peer); + proc_close($peer); +} + echo "PASS: $count PHP contract checks\n"; diff --git a/tests/python/test_jobs.py b/tests/python/test_jobs.py index 2105483..f9f5550 100644 --- a/tests/python/test_jobs.py +++ b/tests/python/test_jobs.py @@ -1042,6 +1042,34 @@ def test_retained_results_bound_memory(self): self.assertIsNone(self.store.get(job["id"])) self.store.submit("test.delay", {"seconds": 0}) + def test_a_full_store_says_when_its_first_retained_job_expires(self): + self.store.close() + self.store = JobStore(Settings(concurrency=1, capacity=2, retention_seconds=2.5, test_tasks=True)) + self.wait(self.store.submit("test.delay", {"seconds": 0})["id"]) + self.store.submit("test.delay", {"seconds": 5}) + time.sleep(1) + with self.assertRaises(CapacityError) as caught: + self.store.submit("test.delay", {"seconds": 0}) + # About 1.5 s of the finished job's retention is left, rounded up; the running job does not count. + self.assertEqual(caught.exception.retry_after, 2) + + def test_a_store_full_of_unfinished_jobs_says_one_second(self): + self.store.close() + self.store = JobStore(Settings(concurrency=1, capacity=1, retention_seconds=300, test_tasks=True)) + self.store.submit("test.delay", {"seconds": 5}) + with self.assertRaises(CapacityError) as caught: + self.store.submit("test.delay", {"seconds": 0}) + self.assertEqual(caught.exception.retry_after, 1) + + def test_a_store_full_of_long_retained_jobs_says_their_retention(self): + self.store.close() + self.store = JobStore(Settings(concurrency=1, capacity=1, retention_seconds=300, test_tasks=True)) + self.wait(self.store.submit("test.delay", {"seconds": 0})["id"]) + with self.assertRaises(CapacityError) as caught: + self.store.submit("test.delay", {"seconds": 0}) + self.assertIs(type(caught.exception.retry_after), int) + self.assertIn(caught.exception.retry_after, (299, 300)) + def test_unknown_jobs(self): self.assertIsNone(self.store.get("a" * 32)) self.assertIsNone(self.store.cancel("a" * 32)) @@ -1055,8 +1083,9 @@ def test_shutdown_cancels_jobs_and_refuses_submission(self): job = self.store.submit("test.delay", {"seconds": 5}) self.store.close() self.assertEqual(self.store.get(job["id"])["status"], "cancelled") - with self.assertRaises(CapacityError): + with self.assertRaises(CapacityError) as caught: self.store.submit("test.delay", {}) + self.assertEqual(caught.exception.retry_after, 1) class WarmPoolTests(unittest.TestCase): diff --git a/tests/python/test_server.py b/tests/python/test_server.py index 8770a53..765250a 100644 --- a/tests/python/test_server.py +++ b/tests/python/test_server.py @@ -4,6 +4,7 @@ import sys import tempfile import threading +import time import unittest from unittest.mock import patch @@ -102,6 +103,48 @@ def test_weak_token_is_rejected(self): unittest.main() +class CapacityHttpTests(unittest.TestCase): + """A store of its own, so the full one never refuses the other HTTP tests.""" + + def setUp(self): + self.server = BridgeServer(("127.0.0.1", 0), TOKEN, Settings(capacity=1, retention_seconds=300)) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.thread.join() + self.server.server_close() + + def submit(self): + connection = http.client.HTTPConnection(*self.server.server_address, timeout=2) + try: + connection.request("POST", "/v1/jobs", json.dumps({"task": "rerank", "input": {"query": "x", "documents": ["x"]}}), + {"Authorization": "Bearer " + TOKEN, "Content-Type": "application/json"}) + response = connection.getresponse() + return response.status, response.getheader("Retry-After"), json.loads(response.read()) + finally: + connection.close() + + def test_a_refusal_for_capacity_says_when_to_retry_in_whole_seconds(self): + status, retry_after, job = self.submit() + self.assertEqual((status, retry_after), (202, None)) + deadline = time.monotonic() + 5 + while self.server.jobs.get(job["id"])["status"] != "succeeded": + self.assertLess(time.monotonic(), deadline, "the job never finished") + time.sleep(0.01) + status, retry_after, body = self.submit() + self.assertEqual((status, body["error"]["code"]), (429, "capacity_exceeded")) + # Whole seconds, never an HTTP date: the finished job holds its slot for the 300 s retention. + self.assertRegex(retry_after, r"\A[1-9][0-9]*\Z") + self.assertIn(int(retry_after), (299, 300)) + + def test_a_shutting_down_store_says_one_second(self): + self.server.jobs.close() + status, retry_after, _ = self.submit() + self.assertEqual((status, retry_after), (429, "1")) + + class StartupTests(unittest.TestCase): """An ONNX worker refuses to start without its model files, naming what is missing, rather than failing every job as task_failed: a cache fetched before a model was added holds no directory for it.""" diff --git a/worker/ai_bridge/jobs.py b/worker/ai_bridge/jobs.py index a979a07..e76a424 100644 --- a/worker/ai_bridge/jobs.py +++ b/worker/ai_bridge/jobs.py @@ -30,7 +30,9 @@ class CapacityError(Exception): - pass + def __init__(self, message, retry_after=1): + super().__init__(message) + self.retry_after = retry_after # whole seconds until a submission may be accepted, at least 1 @dataclass @@ -148,7 +150,7 @@ def submit(self, task, payload, timeout_ms=30000): raise CapacityError("worker is shutting down") self._expire() if len(self._jobs) >= self.settings.capacity: - raise CapacityError("job capacity reached; wait for retained jobs to expire") + raise CapacityError("job capacity reached; wait for retained jobs to expire", self._retry_after()) now = time.monotonic() job_id = uuid.uuid4().hex self._jobs[job_id] = { @@ -174,6 +176,14 @@ def cancel(self, job_id): self._abort(job, "cancelled") return self._public(job) + def _retry_after(self): + """Seconds until the first retained job expires; a store holding only unfinished jobs gives a short default, + since none of them has a known end.""" + finished = [job["_finished"] for job in self._jobs.values() if job["_finished"] is not None] + if not finished: + return 1 + return max(1, math.ceil(min(finished) + self.settings.retention_seconds - time.monotonic())) + def _public(self, job): return copy.deepcopy({key: value for key, value in job.items() if not key.startswith("_")}) diff --git a/worker/ai_bridge/server.py b/worker/ai_bridge/server.py index d9b38ad..7f3ca22 100644 --- a/worker/ai_bridge/server.py +++ b/worker/ai_bridge/server.py @@ -68,9 +68,11 @@ def log_message(self, *_): # Do not log URLs, credentials, inputs or model results. pass - def _reply(self, status, value): + def _reply(self, status, value, headers=()): body = json.dumps(value, allow_nan=False, separators=(",", ":")).encode() self.send_response(status) + for name, header in headers: + self.send_header(name, header) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") @@ -79,8 +81,8 @@ def _reply(self, status, value): self.close_connection = True self.wfile.write(body) - def _error(self, status, code, message): - self._reply(status, {"error": {"code": code, "message": message}}) + def _error(self, status, code, message, headers=()): + self._reply(status, {"error": {"code": code, "message": message}}, headers) def _authorized(self): values = self.headers.get_all("Authorization", []) @@ -151,8 +153,9 @@ def do_POST(self): self._error(404, "not_found", "Job or route not found") else: self._reply(200, job) - except CapacityError: - self._error(429, "capacity_exceeded", "Worker capacity reached") + except CapacityError as error: + # Delay seconds, never an HTTP date: a client's clock need not agree with this one. + self._error(429, "capacity_exceeded", "Worker capacity reached", [("Retry-After", str(int(error.retry_after)))]) except OverflowError: self._error(413, "body_too_large", "Request exceeds 262144 bytes") except (ValueError, UnicodeError, RecursionError): From cfd219a80cb6547d436388d9cb0e6f43d0fce168 Mon Sep 17 00:00:00 2001 From: Cristian Girlea Date: Wed, 21 Oct 2026 08:18:47 +0300 Subject: [PATCH 2/2] fix: set the worker's test count in testing.md to what the merged suite reports --- docs/testing.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/testing.md b/docs/testing.md index e34328d..5753f82 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -48,7 +48,7 @@ PHP branch support was verified locally on 2026-09-17 on PHP 8.2.33, 8.3.33, 8.4 | Layer | Scope and counting | | --- | --- | -| Python worker | 191 tests: lifecycle, validation and HTTP, the job table in SQLite (a store reopened on its file fails unfinished jobs as `service_restarted`, serves retained results, deletes expired rows from the file and never holds an input; a submission the table refuses is a 429, a later write it refuses leaves the job served, and a key whose row outlived a failed delete runs again) and idempotency keys, process crashes, cancellation, timeouts, retention, the Retry-After a full or closing store gives and its header on every 429, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample and scoring, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values, a state directory that cannot hold the table), and the model-name table every result reports from | +| Python worker | 189 tests: lifecycle, validation and HTTP, the job table in SQLite (a store reopened on its file fails unfinished jobs as `service_restarted`, serves retained results, deletes expired rows from the file and never holds an input; a submission the table refuses is a 429, a later write it refuses leaves the job served, and a key whose row outlived a failed delete runs again) and idempotency keys, process crashes, cancellation, timeouts, retention, the Retry-After a full or closing store gives and its header on every 429, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample and scoring, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values, a state directory that cannot hold the table), and the model-name table every result reports from | | MCP server | 125 tests, offline with hash-pinned wheels: the protocol client against an in-process worker, every tool through the SDK's in-memory client with test tasks enabled, the entry point as a subprocess, result validators on synthetic data, the index builder (chunking, the format and each corruption it refuses, building against the worker, caching, its command line) `bridge_search` (the model gate, path confinement, reloads, rerank on and off), the reachability of every answer in the LLM evaluation, that a default search sees only part of its corpus and that its two-step question takes two steps, that the demo worker reaches none of the semantic evaluation's answers, that a tool waits out a full worker for its Retry-After within the call's budget or with its own backoff without one, before each of a search's jobs and within one budget, and says so when it stays full, the evaluation runner's command, isolation checks, scoring and per-backend launcher and index check without starting `claude`, a check that the server's model names match the worker's, that every path parameter says where it resolves and every position in a result says what it counts, and a check that every timing claim in a tool description is a measured median | | PHP contracts + HTTP | 178 checks per PHP version, a full fault-worker's Retry-After read through the client among them; this **includes** the 141 contract checks (Retry-After from a loopback peer, present, absent, malformed and repeated, and a header block over 16 KiB refused), not 178 + 141 | | FrankenPHP | 12 jobs submitted by six concurrent test threads through a reused PHP worker, a 512-document `top_k` request, embedding and redaction round trips, and invalid-input, missing-job and multibyte boundary checks |