diff --git a/.claude/rules/worker.md b/.claude/rules/worker.md index ee720d3..9d9e314 100644 --- a/.claude/rules/worker.md +++ b/.claude/rules/worker.md @@ -16,10 +16,10 @@ paths: Test-only tasks exist only with `BRIDGE_TEST_TASKS=1` (compose's `fault-worker`). - `MODEL_NAMES` is what every result reports as `model` and what the MCP server's `BACKENDS` table must equal (`tests/mcp/test_models.py`); change both together. -- Limits live in three places: concurrency, capacity and retention are `Settings` in `jobs.py` (defaults - from the environment in `server.main`), the body size and the token limits are `MAX_BODY`, `MAX_TOKENS`, - `MAX_TOKENS_FILE` and `TOKEN_NAME` in `server.py`, and the field, count and character limits are the - `MAX_*` constants in `tasks.py`; the deadline range is checked in `JobStore.submit`. `docs/protocol.md` +- Limits live in three places: concurrency, capacity, the per-owner cap and retention are `Settings` in + `jobs.py` (defaults from the environment in `server.main`), the body size and the token limits are + `MAX_BODY`, `MAX_TOKENS`, `MAX_TOKENS_FILE` and `TOKEN_NAME` in `server.py`, and the field, count and + character limits are the `MAX_*` constants in `tasks.py`; the deadline range is checked in `JobStore.submit`. `docs/protocol.md` documents them; change the doc with the limit. - Tests run offline with `python -m unittest discover -s tests/python` inside `python-tests` (`network_mode: none`). A change to the model path belongs on `model-smoke.yml`'s `paths:` list, so diff --git a/CHANGELOG.md b/CHANGELOG.md index 39b87e4..bbaf582 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ All notable changes to this project are documented in this file. The format foll - A graceful drain: on SIGTERM or SIGINT the worker answers `POST /v1/jobs` with 503 `draining` and `Retry-After: 1`, fails queued jobs as `service_restarted` at once, keeps answering `GET`, and lets running jobs finish for up to `BRIDGE_DRAIN_SECONDS` (integer 0 to 300, default 30) before it stops; a second signal stops it at once. - `GET /readyz`, behind any configured token like `/healthz`: 200 when the worker is not draining and a child is idle with its backend's models loaded, otherwise 503 with a reason (`draining`, `no_idle_child`, `models_loading`). A child now loads its backend's models as it starts instead of on its first job. Compose's MCP workers report healthy only once ready. - Optional structured logging: `BRIDGE_LOG=json` writes one JSON object per line on stderr, best effort (a full queue or the exit can drop lines), a `request` line after each reply (method, route template, status, duration, the token's name), a `job` line when a job ends (job ID, task, owner, terminal status, queue and run times), and `start`, `drain` and `stop` lines. A fixed schema checked value by value keeps inputs, results, headers, query strings, tokens and exception text out; a stderr that blocks or fails never holds a request. `off`, the default, writes nothing, and any other value stops the worker at start naming `BRIDGE_LOG`. +- `BRIDGE_OWNER_CAPACITY`, a cap on one token's queued, running and retained jobs (1 to the capacity; unset, the whole capacity, so nothing changes without it). A submission over it is the same 429 `capacity_exceeded` as a full worker, with a `Retry-After` from that token's own first expiry; a repeated idempotency key is still answered, and the total capacity still applies. The `start` log line carries `owner_capacity`. - A full connection budget (32) is answered 503 `service_busy` with `Retry-After: 1` and `Connection: close` instead of a closed socket, and a request line over 8 KiB or a header block over 16 KiB is refused with 431 `headers_too_large`. ### Changed diff --git a/docker/compose.yaml b/docker/compose.yaml index 20abf68..97a1966 100644 --- a/docker/compose.yaml +++ b/docker/compose.yaml @@ -280,6 +280,9 @@ services: PYTHONDONTWRITEBYTECODE: "1" PYTHONPATH: /work/worker BRIDGE_TOKENS: app=test-only-app-client-token-never-use-in-production,other=test-only-other-client-token-never-use-in-production + # Small enough for tests/integration.py, which repeats both numbers, to fill one token's cap and then the worker. + BRIDGE_CAPACITY: "12" + BRIDGE_OWNER_CAPACITY: "8" command: [python, -m, ai_bridge.server, --host, 0.0.0.0] hostile: diff --git a/docs/protocol.md b/docs/protocol.md index 6dfb5f8..0004267 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -21,7 +21,7 @@ The worker takes its tokens from exactly one of three variables, read once at st A name is 1–64 characters of `A-Z`, `a-z`, `0-9`, `_`, `.` and `-`; a token is at least 32 non-whitespace ASCII characters, everything after the first `=` of its entry. At most 64 tokens; no name or token may appear twice. An empty variable counts as unset, so compose's `${X:-}` forms work. Setting `BRIDGE_TOKEN` together with either list, or both lists, is refused. The worker refuses to start on any of these, and on a file it cannot read or that group or others can, with a message naming the variable and never a token. `BRIDGE_TOKEN` and `BRIDGE_TOKENS` leave the worker's environment before it starts any child process; the file's path stays, its content does not enter the environment. -A request is authenticated by comparing its bearer with every configured token in constant time; the matching token's name is the request's owner. A job belongs to the owner that submitted it: `GET /v1/jobs/{id}` and `POST /v1/jobs/{id}/cancel` with another token answer 404 `not_found`, exactly as for a job that does not exist, never 403. `GET /healthz`, `GET /readyz` and `GET /metrics` answer any configured token. The owner is not part of a job's snapshot. Capacity, concurrency and retention are shared by every owner. +A request is authenticated by comparing its bearer with every configured token in constant time; the matching token's name is the request's owner. A job belongs to the owner that submitted it: `GET /v1/jobs/{id}` and `POST /v1/jobs/{id}/cancel` with another token answer 404 `not_found`, exactly as for a job that does not exist, never 403. `GET /healthz`, `GET /readyz` and `GET /metrics` answer any configured token. The owner is not part of a job's snapshot. Capacity, concurrency and retention are shared by every owner; `BRIDGE_OWNER_CAPACITY`, with the other limits below, caps how much of the capacity one owner may hold. Submit: @@ -45,9 +45,9 @@ 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`, `deadline_exceeded` and `service_restarted`. 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, an oversized request line or header block 431 and capacity exhaustion 429. Two answers are 503 with `Retry-After: 1` and create no job: `draining` for a submission to a worker that is stopping ([below](#drain-and-readiness)), and `service_busy` when all 32 of the worker's connections are taken; that one is written as the connection is accepted, before anything of the request is read, so it carries no auth check and answers any route. -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, the worker is shutting down or the job table cannot be written, 1. A 429 created no job, so resubmitting after that delay cannot duplicate work. The PHP client exposes the value as `BridgeException::$retryAfterSeconds`, through the cURL and the PSR-18 transport alike, 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. +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, the worker is shutting down or the job table cannot be written, 1. A submission over its owner's cap (`BRIDGE_OWNER_CAPACITY`) is the same 429 `capacity_exceeded` with the same body, and its delay counts only that owner's jobs: the time until the first of its own finished jobs leaves retention, or 1 when it holds only unfinished ones. The owner's cap is checked before the total, and a repeated idempotency key is answered at either. A 429 created no job, so resubmitting after that delay cannot duplicate work. The PHP client exposes the value as `BridgeException::$retryAfterSeconds`, through the cURL and the PSR-18 transport alike, 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 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`. `BRIDGE_OWNER_CAPACITY` caps the queued, running and retained jobs of one owner (one token), an integer from 1 to the capacity; unset, it is the whole capacity, so one token may fill the worker as before. The total still applies on top, so caps that add up to more than the capacity share what is left. 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. With `BRIDGE_STATE_DIR` set, the job table (never a job's input) is written through to SQLite at `/jobs.sqlite` on every status change; unset, it stays in memory and a restart loses every job. When the worker starts on an existing file, a job still `queued` or `running` there becomes `failed` with error `service_restarted` (its input was never stored, so it is not replayed; a shutdown reports such jobs as `cancelled` to whoever polls the stopping process but leaves them unfinished in the file), a finished job whose retention has not run out by the wall clock is served by GET as before until it does, and an expired one is deleted. The file records each job's owner, the token's name and never its value. A file written before owners existed has no `owner` column: the worker adds it when it opens the file, and its rows belong to `default`, the single-token name, so a worker moved to a token list keeps serving them only to a token named `default`. A 404 for a well-formed ID therefore means the job expired, was never submitted to this instance, belongs to another token, or was lost to a restart without a state directory; the worker cannot tell these apart. @@ -90,7 +90,7 @@ The worker writes no log by default: `BRIDGE_LOG` unset, empty or `off` writes n | --- | --- | --- | | `request` | After the reply to a request is sent, a 401, 408, 413, 431 and 503 included; a connection that ends before any reply writes none | `method` (`GET`, `POST`, `other`, or `null` when the request line could not be read or parsed), `route` (`/healthz`, `/readyz`, `/metrics`, `/v1/jobs`, `/v1/jobs/{id}`, `/v1/jobs/{id}/cancel`, or `other` for any other path, a query string included), `status`, `duration_ms` (from the start of reading the request to the reply sent), `owner` (the token's name, `null` when the request was not authenticated) | | `job` | A job reaches `succeeded`, `failed`, `timed_out` or `cancelled` | `job_id`, `task`, `owner`, `job_status`, `queue_ms` (submission to `running`, `null` if no child started it), `run_ms` (`running` to the terminal status, `null` likewise) | -| `start` | The port is bound and the children are starting | `backend`, `concurrency`, `capacity`, `retention_seconds`, `max_jobs`, `drain_seconds`, `tokens` (how many tokens are configured; never a name or a token) | +| `start` | The port is bound and the children are starting | `backend`, `concurrency`, `capacity`, `owner_capacity` (the capacity when `BRIDGE_OWNER_CAPACITY` is unset), `retention_seconds`, `max_jobs`, `drain_seconds`, `tokens` (how many tokens are configured; never a name or a token) | | `drain` | A drain ends, before the worker stops serving | `duration_ms` (from the signal), `running` (jobs still running when it ended, which the stop then cancels), `interrupted` (a second signal ended it) | | `stop` | The worker has stopped its children and closed the job table | `cancelled` (unfinished jobs the stop ended as `cancelled`) | diff --git a/docs/security.md b/docs/security.md index 2d6d327..05e5c42 100644 --- a/docs/security.md +++ b/docs/security.md @@ -2,7 +2,7 @@ This experimental service is designed for trusted application-to-service traffic on a private network. Do not expose the example routes or Python server directly to the internet. Across hosts, put a TLS proxy in front of the worker as [Cross-host use behind TLS](integration.md#cross-host-use-behind-tls) shows. -- Generate a service token of at least 32 non-whitespace ASCII characters. Never commit it. A token reaches only the jobs it submitted: another token's job answers 404 as if it did not exist, and idempotency keys are per token. Give each application, and the MCP server, its own token in `BRIDGE_TOKENS` or `BRIDGE_TOKENS_FILE` (mode 0600 or stricter), so one can be revoked by removing its entry and restarting the worker without touching the others, and a leaked token exposes only its own client's jobs. Capacity is still shared, so any token can fill the worker; `/healthz`, `/readyz` and `/metrics` answer every token, and the metrics carry the token names, never the tokens. The job table stores a job's owner by name only. +- Generate a service token of at least 32 non-whitespace ASCII characters. Never commit it. A token reaches only the jobs it submitted: another token's job answers 404 as if it did not exist, and idempotency keys are per token. Give each application, and the MCP server, its own token in `BRIDGE_TOKENS` or `BRIDGE_TOKENS_FILE` (mode 0600 or stricter), so one can be revoked by removing its entry and restarting the worker without touching the others, and a leaked token exposes only its own client's jobs. Capacity is shared, so by default any token can fill the worker and starve the others; set `BRIDGE_OWNER_CAPACITY` below `BRIDGE_CAPACITY` to cap one token's queued, running and retained jobs, which is off by default. A token at its cap gets the same 429 as a full worker, so it learns nothing of the other tokens' load; the cap is the same for every token, so where it times the number of tokens exceeds the capacity the tokens together can still fill the worker: set it at most the capacity divided by the number of tokens where every token must always find room; `/healthz`, `/readyz` and `/metrics` answer every token, and the metrics carry the token names, never the tokens. The job table stores a job's owner by name only. - Enforce end-user authentication and per-job ownership in the PHP application. The example deliberately omits those application-specific policies and publishes no ports. - Only trusted task implementations are allowed. The service removes `BRIDGE_TOKEN` and `BRIDGE_TOKENS` from its environment before it starts its child processes, and reads `BRIDGE_TOKENS_FILE` without putting its content there, so no token is in either their environment nor their initial environment block (`/proc//environ`), which a removal inside the child would leave as it was; the children never speak HTTP. Everything else in the service environment stays visible to them. A child serves many jobs in turn and keeps its loaded models; it keeps no reference to a finished job's input or result, but the memory they used is freed, not wiped, so a core dump of a long-lived child can still hold earlier jobs' data where a process that died after each job could not. They are not suitable for arbitrary user-provided scripts. - Container configuration uses a non-root user, read-only filesystem, dropped capabilities, no-new-privileges, CPU/memory/PID bounds, and an internal-only network for execution. It mounts only this checkout and the optional dependency/model cache, never a Docker socket or home directory. diff --git a/docs/testing.md b/docs/testing.md index 302a0c7..e65bad9 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -58,9 +58,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 | 395 tests: lifecycle, validation and HTTP, the optional JSON log (every line parses and holds only its event's keys, an input, token, header, query string, idempotency key and a failing task's exception text never reach it, `off` writes nothing and a bad `BRIDGE_LOG` stops the start, the route templates, the 401 without an owner, consistent job timings, and a stderr that fails or blocks costing a request nothing), the drain (submissions refused with 503, queued jobs failed at once, running ones finished within `BRIDGE_DRAIN_SECONDS`, a second signal), `/readyz` and its reasons for any configured token, the 503 for a full connection budget and the 431 for an oversized request line or header block, per-client tokens (each source and every way it is refused at start, a job and its idempotency key scoped to the submitting token, through the job table and a table from before owners, the `owner` metrics series, and no token in an error, the metrics, the table, a child's environment or a start message), 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, scoring and WNUT-17 reader, the load measurement's settings, percentiles, error counting and table, the ranking measurement's metrics, BM25, dataset reader and fit to the worker's limits, the public datasets' pins and licence gate, 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 | 406 tests: lifecycle, validation and HTTP, the per-token capacity cap (one token refused at its cap with its own `Retry-After` while another submits, a repeated key answered there, the total still enforced with the `Retry-After` of whichever limit applies, jobs restored from the table counted until they expire, the default unchanged, and every out-of-range value refused at start), the optional JSON log (every line parses and holds only its event's keys, an input, token, header, query string, idempotency key and a failing task's exception text never reach it, `off` writes nothing and a bad `BRIDGE_LOG` stops the start, the route templates, the 401 without an owner, consistent job timings, and a stderr that fails or blocks costing a request nothing), the drain (submissions refused with 503, queued jobs failed at once, running ones finished within `BRIDGE_DRAIN_SECONDS`, a second signal), `/readyz` and its reasons for any configured token, the 503 for a full connection budget and the 431 for an oversized request line or header block, per-client tokens (each source and every way it is refused at start, a job and its idempotency key scoped to the submitting token, through the job table and a table from before owners, the `owner` metrics series, and no token in an error, the metrics, the table, a child's environment or a start message), 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, scoring and WNUT-17 reader, the load measurement's settings, percentiles, error counting and table, the ranking measurement's metrics, BM25, dataset reader and fit to the worker's limits, the public datasets' pins and licence gate, 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 | 192 checks per PHP version, a full fault-worker's Retry-After read through the client among them and a two-token `tokens-worker` where another token's read and cancel are 404 and the same idempotency key is two jobs; this **includes** the 148 contract checks (Retry-After from a loopback peer, present, absent, malformed and repeated, and a header block over 16 KiB refused), not 192 + 148; the contract checks include a fake transport's view of each request, the wait deadline across a transport failure, and that no file in `src/` outside `Laravel/` and `Symfony/` names a framework | +| PHP contracts + HTTP | 192 checks per PHP version, a full fault-worker's Retry-After read through the client among them and a two-token `tokens-worker` where another token's read and cancel are 404 and the same idempotency key is two jobs (on which `tests/integration.py` then fills one token's `BRIDGE_OWNER_CAPACITY` of 8 while the other still submits up to the worker's capacity of 12); this **includes** the 148 contract checks (Retry-After from a loopback peer, present, absent, malformed and repeated, and a header block over 16 KiB refused), not 192 + 148; the contract checks include a fake transport's view of each request, the wait deadline across a transport failure, and that no file in `src/` outside `Laravel/` and `Symfony/` names a framework | | PSR-18 transport | The 148 contract checks plus 30 through `Psr18Transport` and in-repo fakes of the PSR-7, PSR-17 and PSR-18 interfaces (`tests/php/psr18_fakes.php`): a round trip with its method, URL, headers and body, a body at the 262144-byte cap and one over it read one byte past the cap and no further, five redirect statuses refused unread, a non-JSON or missing content type, malformed JSON, a 429's Retry-After from one header, a repeated one and none, a client exception and a body read failure as `transport_error` without their text; 178 checks by `php-psr18-tests` with `.cache/php-vendor` on the include path, once, in the MCP server job; `php-tests` skips them with one printed line; no real Symfony HttpClient or Guzzle runs | | 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; then, directly against `tokens-worker`, the same ownership checks as the PHP layer and the `owner` metrics | | PHP syntax | All 25 PHP source/example/test files | diff --git a/tests/integration.py b/tests/integration.py index 54f78d1..8c36482 100644 --- a/tests/integration.py +++ b/tests/integration.py @@ -10,6 +10,8 @@ BASE = os.environ.get("FRANKENPHP_URL", "http://frankenphp:8080") WORKER = os.environ.get("WORKER_URL", "http://worker:8090") TOKENS_WORKER = os.environ.get("TOKENS_WORKER_URL", "http://tokens-worker:8090") +# BRIDGE_OWNER_CAPACITY and BRIDGE_CAPACITY of tokens-worker in docker/compose.yaml. +OWNER_CAPACITY, CAPACITY = 8, 12 def request(path, body=None): @@ -83,6 +85,55 @@ def owners(): assert token not in text +def submit(token, body): + """Status, Retry-After and decoded body of one submission to the two-token worker.""" + headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} + try: + with urlopen(Request(TOKENS_WORKER + "/v1/jobs", data=json.dumps(body).encode(), headers=headers), + timeout=5) as response: + return response.status, response.headers["Retry-After"], json.load(response) + except HTTPError as error: + return error.code, error.headers["Retry-After"], json.load(error) + + +def held(samples, owner): + """Jobs an owner has submitted, all still held: none is older than the worker's 300 s retention.""" + prefix = f'bridge_jobs_submitted_total{{owner="{owner}",' + return sum(value for series, value in samples.items() if series.startswith(prefix)) + + +def owner_capacity(): + """One token fills its cap and is refused while the other still submits, until the worker as a whole is full.""" + app, other = os.environ["BRIDGE_APP_TOKEN"], os.environ["BRIDGE_OTHER_TOKEN"] + rejected = 'bridge_submissions_rejected_total{{owner="{}",reason="capacity"}}' + _, counted = metrics(app, TOKENS_WORKER) + rerank = {"task": "rerank", "input": {"query": "needle", "documents": ["needle"]}} + keyed = dict(rerank, idempotency_key="python-cap-" + os.urandom(8).hex()) + first = None + for _ in range(OWNER_CAPACITY + 1): + status, retry_after, body = submit(app, rerank if first else keyed) + if status != 202: + break + first = first or body + assert first is not None and (status, body["error"]["code"]) == (429, "capacity_exceeded"), status + assert 1 <= int(retry_after) <= 300, retry_after + status, _, again = submit(app, keyed) + assert (status, again["id"]) == (202, first["id"]), "a repeated key is answered at the cap" + accepted = 0 + for _ in range(CAPACITY + 1): + status, retry_after, body = submit(other, rerank) + if status != 202: + break + accepted += 1 + assert accepted >= 1 and status == 429, (accepted, status) + _, moved = metrics(other, TOKENS_WORKER) + assert held(moved, "app") == OWNER_CAPACITY, held(moved, "app") + assert held(moved, "app") + held(moved, "other") == CAPACITY and held(moved, "other") < OWNER_CAPACITY + for owner in ("app", "other"): + name = rejected.format(owner) + assert moved[name] - counted[name] == 1, (name, counted[name], moved[name]) + + def wait_ready(): deadline = time.monotonic() + 30 while time.monotonic() < deadline: @@ -215,7 +266,8 @@ def main(): for document in ["unique0 token0", "needle token", "x@y.io", "4111 1111 1111 1111", "red apple", "café"]: assert document not in text, document owners() - print("PASS: 12 concurrent isolated jobs through a reused FrankenPHP worker; top_k narrowing; embedding; redaction; invalid and missing-job cases; worker metrics; jobs scoped to their token") + owner_capacity() + print("PASS: 12 concurrent isolated jobs through a reused FrankenPHP worker; top_k narrowing; embedding; redaction; invalid and missing-job cases; worker metrics; jobs scoped to their token; a per-token capacity cap") if __name__ == "__main__": diff --git a/tests/python/test_jobs.py b/tests/python/test_jobs.py index 6931067..703526b 100644 --- a/tests/python/test_jobs.py +++ b/tests/python/test_jobs.py @@ -1704,6 +1704,130 @@ def test_keys_are_per_owner(self): finally: store.close() + def test_an_owner_at_its_cap_is_refused_while_the_other_still_submits(self): + store = JobStore(self.settings(capacity=6, owner_capacity=2, retention_seconds=300)) + try: + # The other owner's retained job would give about 300 s; the capped owner holds only unfinished jobs. + self.wait(store, store.submit("test.delay", {"seconds": 0}, owner="other")["id"], "other") + for _ in range(2): + store.submit("test.delay", {"seconds": 5}, owner="app") + for _ in range(2): + with self.assertRaises(CapacityError) as caught: + store.submit("test.delay", {"seconds": 0}, owner="app") + self.assertEqual(caught.exception.retry_after, 1) + store.submit("test.delay", {"seconds": 5}, owner="other") + self.assertEqual(store.snapshot()["rejected"], {("app", "capacity"): 2, ("other", "capacity"): 0}) + with self.assertRaises(CapacityError): + store.submit("test.delay", {"seconds": 0}, owner="other") + self.assertEqual(store.snapshot()["rejected"], {("app", "capacity"): 2, ("other", "capacity"): 1}) + finally: + store.close() + + def test_the_cap_s_retry_after_is_the_owner_s_own_earliest_expiry(self): + store = JobStore(self.settings(capacity=8, owner_capacity=2, retention_seconds=5)) + try: + self.wait(store, store.submit("test.delay", {"seconds": 0}, owner="other")["id"], "other") + time.sleep(2) + for _ in range(2): + self.wait(store, store.submit("test.delay", {"seconds": 0}, owner="app")["id"], "app") + with self.assertRaises(CapacityError) as caught: + store.submit("test.delay", {"seconds": 0}, owner="app") + # The other owner's job leaves in about 3 s, which would free nothing for this one. + self.assertIs(type(caught.exception.retry_after), int) + self.assertIn(caught.exception.retry_after, (4, 5)) + finally: + store.close() + + def test_a_repeated_key_is_answered_at_the_owner_s_cap(self): + store = JobStore(self.settings(capacity=8, owner_capacity=2)) + try: + keyed = store.submit("test.delay", {"seconds": 5}, idempotency_key="k", owner="app") + store.submit("test.delay", {"seconds": 5}, owner="app") + self.assertEqual(store.submit("test.delay", {"seconds": 0}, idempotency_key="k", owner="app")["id"], + keyed["id"]) + with self.assertRaises(CapacityError): + store.submit("test.delay", {"seconds": 0}, idempotency_key="new", owner="app") + self.assertEqual(store.snapshot()["rejected"]["app", "capacity"], 1) + finally: + store.close() + + def test_the_total_capacity_holds_when_the_caps_add_up_to_more(self): + store = JobStore(self.settings(capacity=3, owner_capacity=2)) + try: + for owner in ("app", "app", "other"): + store.submit("test.delay", {"seconds": 5}, owner=owner) + with self.assertRaises(CapacityError) as caught: + store.submit("test.delay", {"seconds": 0}, owner="other") + self.assertEqual(caught.exception.retry_after, 1) + finally: + store.close() + + def test_a_full_store_answers_with_the_limit_that_applies_to_the_owner(self): + store = JobStore(self.settings(capacity=3, owner_capacity=2, retention_seconds=300)) + try: + ages = {} + for owner, age in (("app", 20), ("app", 10), ("other", 250)): + ages[self.wait(store, store.submit("test.delay", {"seconds": 0}, owner=owner)["id"], owner)["id"]] = age + # Stamped finish times: the other owner's job leaves in about 50 s, the capped owner's first in about 280 s. + with store._lock: + now = time.monotonic() + for job_id, age in ages.items(): + store._jobs[job_id]["_finished"] = now - age + with self.assertRaises(CapacityError) as below: + store.submit("test.delay", {"seconds": 0}, owner="other") + self.assertIn(below.exception.retry_after, (49, 50)) + with self.assertRaises(CapacityError) as at_cap: + store.submit("test.delay", {"seconds": 0}, owner="app") + self.assertIn(at_cap.exception.retry_after, (279, 280)) + finally: + store.close() + + def test_restored_jobs_count_toward_the_owner_s_cap_until_they_expire(self): + with tempfile.TemporaryDirectory() as state_dir: + first = JobStore(self.settings(owner_capacity=3, retention_seconds=300, state_dir=state_dir)) + try: + kept = [self.wait(first, first.submit("test.delay", {"seconds": 0}, idempotency_key=key, + owner="app")["id"], "app") for key in ("a", "b")] + unfinished = first.submit("test.delay", {"seconds": 5}, owner="app") + finally: + first.close() + reopened = JobStore(self.settings(owner_capacity=3, retention_seconds=300, state_dir=state_dir)) + try: + self.assertEqual(reopened.get(unfinished["id"], "app")["error"]["code"], "service_restarted") + with self.assertRaises(CapacityError): + reopened.submit("test.delay", {"seconds": 0}, owner="app") + reopened.submit("test.delay", {"seconds": 0}, owner="other") + repeat = reopened.submit("test.delay", {"seconds": 0}, idempotency_key="b", owner="app") + self.assertEqual(repeat["id"], kept[1]["id"]) + with reopened._lock: + reopened._jobs[kept[0]["id"]]["_finished"] = time.monotonic() - 300 + again = reopened.submit("test.delay", {"seconds": 0}, idempotency_key="a", owner="app") + self.assertNotEqual(again["id"], kept[0]["id"]) + with self.assertRaises(CapacityError): + reopened.submit("test.delay", {"seconds": 0}, owner="app") + finally: + reopened.close() + + def test_without_a_cap_one_owner_may_take_the_whole_capacity(self): + self.assertIsNone(Settings().owner_capacity) + store = JobStore(self.settings(capacity=2)) + try: + for _ in range(2): + store.submit("test.delay", {"seconds": 5}, owner="app") + for owner in ("app", "other"): + with self.subTest(owner=owner), self.assertRaises(CapacityError): + store.submit("test.delay", {"seconds": 0}, owner=owner) + finally: + store.close() + + def test_an_owner_cap_outside_1_to_the_capacity_is_refused(self): + for value in (0, -1, 5, True, 1.5, "1"): + with self.subTest(value=value), self.assertRaisesRegex(ValueError, "invalid worker limits"): + JobStore(Settings(capacity=4, owner_capacity=value)) + for value in (1, 4): + with self.subTest(value=value): + JobStore(Settings(concurrency=1, capacity=4, owner_capacity=value)).close() + def test_an_unknown_owner_or_owner_list_is_refused(self): store = JobStore(self.settings()) try: diff --git a/tests/python/test_log.py b/tests/python/test_log.py index 9350fc1..beb9d8a 100644 --- a/tests/python/test_log.py +++ b/tests/python/test_log.py @@ -72,22 +72,23 @@ def test_each_event_writes_exactly_its_keys_with_null_for_a_missing_one(self): with capture() as lines: log.event("request", method="GET", route="/healthz", status=200, duration_ms=1.5, owner="default") log.event("job", job_id="a" * 32, task="rerank", owner="default", job_status="succeeded") - log.event("start", backend="lexical", concurrency=2, capacity=128, retention_seconds=300.0, - max_jobs=1000, drain_seconds=30, tokens=1) + log.event("start", backend="lexical", concurrency=2, capacity=128, owner_capacity=64, + retention_seconds=300.0, max_jobs=1000, drain_seconds=30, tokens=1) log.event("drain", duration_ms=0.0, running=0, interrupted=False) log.event("stop", cancelled=0) self.assertEqual([line["event"] for line in lines], ["request", "job", "start", "drain", "stop"]) self.assertEqual(lines[0], {**lines[0], "method": "GET", "route": "/healthz", "status": 200, "duration_ms": 1.5, "owner": "default", "level": "info"}) self.assertEqual((lines[1]["queue_ms"], lines[1]["run_ms"]), (None, None)) - self.assertEqual(lines[2]["tokens"], 1) + self.assertEqual((lines[2]["owner_capacity"], lines[2]["tokens"]), (64, 1)) def test_the_schema_is_the_documented_one(self): # Written out here, not read from log.EVENTS, so a change to the schema has to change this test too. expected = { "request": ["method", "route", "status", "duration_ms", "owner"], "job": ["job_id", "task", "owner", "job_status", "queue_ms", "run_ms"], - "start": ["backend", "concurrency", "capacity", "retention_seconds", "max_jobs", "drain_seconds", "tokens"], + "start": ["backend", "concurrency", "capacity", "owner_capacity", "retention_seconds", "max_jobs", + "drain_seconds", "tokens"], "stop": ["cancelled"], "drain": ["duration_ms", "running", "interrupted"], } @@ -112,8 +113,8 @@ def test_a_value_outside_its_fixed_set_or_type_is_null(self): log.event("request", method="GET", route="/v1/jobs", status=True, duration_ms=-1, owner="x" * 65) log.event("job", job_id="sentinel" * 4, task="sentinel", owner=["default"], job_status="sentinel", queue_ms="1", run_ms=float("inf")) - log.event("start", backend="sentinel", concurrency=1.5, capacity=-1, retention_seconds="300", - max_jobs=True, drain_seconds=None, tokens="2") + log.event("start", backend="sentinel", concurrency=1.5, capacity=-1, owner_capacity="1", + retention_seconds="300", max_jobs=True, drain_seconds=None, tokens="2") log.event("drain", duration_ms="1", running=-1, interrupted=1) self.assertNotIn("sentinel", capture.text) # A route that is not a template and a method outside the fixed set are null rather than echoed. diff --git a/tests/python/test_server.py b/tests/python/test_server.py index 599fcb8..bc38bea 100644 --- a/tests/python/test_server.py +++ b/tests/python/test_server.py @@ -707,11 +707,11 @@ def test_every_variable_is_read(self): settings = self.main({ "BRIDGE_CONCURRENCY": "3", "BRIDGE_CAPACITY": "9", "BRIDGE_RETENTION_SECONDS": "1.5", "BRIDGE_BACKEND": "lexical", "BRIDGE_MODEL_DIR": "/elsewhere", "BRIDGE_TEST_TASKS": "1", - "BRIDGE_WORKER_MAX_JOBS": "7", "BRIDGE_STATE_DIR": state_dir, + "BRIDGE_WORKER_MAX_JOBS": "7", "BRIDGE_STATE_DIR": state_dir, "BRIDGE_OWNER_CAPACITY": "4", }) self.assertEqual(settings, Settings(concurrency=3, capacity=9, retention_seconds=1.5, backend="lexical", model_dir="/elsewhere", test_tasks=True, max_jobs=7, - state_dir=state_dir)) + state_dir=state_dir, owner_capacity=4)) def test_an_empty_state_dir_is_unset(self): self.assertIsNone(self.main({"BRIDGE_STATE_DIR": ""}).state_dir) @@ -750,6 +750,19 @@ def test_a_malformed_number_names_its_variable(self): self.main({name: value}) self.assertEqual(raised.exception.code, expected) + def test_an_owner_cap_outside_1_to_the_capacity_or_not_an_integer_names_its_variable(self): + for capacity, value in (("9", "0"), ("9", "10"), (None, "129"), ("9", "-1"), ("9", "many"), ("9", "1.5"), + ("9", "")): + environment = {"BRIDGE_OWNER_CAPACITY": value, **({"BRIDGE_CAPACITY": capacity} if capacity else {})} + with self.subTest(environment=environment): + with self.assertRaises(SystemExit) as raised: + self.main(environment) + self.assertIn("BRIDGE_OWNER_CAPACITY", str(raised.exception.code)) + for value in ("1", "9"): + with self.subTest(value=value): + self.assertEqual(self.main({"BRIDGE_CAPACITY": "9", "BRIDGE_OWNER_CAPACITY": value}).owner_capacity, + int(value)) + def test_an_out_of_range_job_limit_is_refused_by_the_store(self): # The real server: the store checks its limits before it starts any child or binds the port. for value in ("0", "1000001", "-1"): @@ -1346,6 +1359,39 @@ def test_the_owner_series_exist_at_zero_and_move_for_the_right_owner(self): # The families without an owner label are unchanged. self.assertEqual(sample(families, "bridge_job_duration_seconds_count", task="test.delay"), 1) + def test_a_token_at_its_cap_gets_the_full_store_s_429_and_the_other_still_submits(self): + self.stop() + self.start(Settings(concurrency=1, capacity=3, owner_capacity=2, retention_seconds=300, test_tasks=True)) + + def submit(token): + connection = http.client.HTTPConnection(*self.server.server_address, timeout=2) + try: + connection.request("POST", "/v1/jobs", json.dumps({"task": "test.delay", "input": {"seconds": 0}}), + {"Authorization": "Bearer " + token, "Content-Type": "application/json"}) + response = connection.getresponse() + return response.status, response.getheader("Retry-After"), json.loads(response.read()) + finally: + connection.close() + + for _ in range(2): + status, _, job = submit(APP) + self.assertEqual(status, 202) + self.finished(job["id"], "app") + status, retry_after, capped = submit(APP) + self.assertEqual((status, capped["error"]["code"]), (429, "capacity_exceeded")) + # The capped token's own jobs hold its slots for the 300 s retention. + self.assertRegex(retry_after, r"\A[1-9][0-9]*\Z") + self.assertIn(int(retry_after), (299, 300)) + self.assertEqual(submit(OTHER)[0], 202) + # The store is now full as a whole, and its 429 is the same answer. + status, retry_after, full = submit(OTHER) + self.assertEqual(status, 429) + self.assertIn(int(retry_after), (299, 300)) + self.assertEqual(capped, full) + families = parse_metrics(self.call("GET", "/metrics", APP)[1]) + self.assertEqual(sample(families, "bridge_submissions_rejected_total", owner="app", reason="capacity"), 1) + self.assertEqual(sample(families, "bridge_submissions_rejected_total", owner="other", reason="capacity"), 1) + def test_no_token_value_reaches_the_job_table_and_a_restored_job_keeps_its_owner(self): status, job = self.submit(APP, "table-key") self.assertEqual(status, 202) @@ -1646,16 +1692,20 @@ def main(self, environment, serve=None): def test_json_writes_start_and_stop_with_safe_settings_only(self): text = self.main({"BRIDGE_LOG": "json", "BRIDGE_TOKENS": f"app={APP},{SENTINEL}={OTHER}", - "BRIDGE_CAPACITY": "9", "BRIDGE_DRAIN_SECONDS": "7"}) + "BRIDGE_CAPACITY": "9", "BRIDGE_OWNER_CAPACITY": "3", "BRIDGE_DRAIN_SECONDS": "7"}) lines = parse_log(text) self.assertEqual([line["event"] for line in lines], ["start", "stop"]) self.assertEqual({key: lines[0][key] for key in log.EVENTS["start"]}, - {"backend": "lexical", "concurrency": 2, "capacity": 9, "retention_seconds": 300.0, - "max_jobs": 1000, "drain_seconds": 7, "tokens": 2}) + {"backend": "lexical", "concurrency": 2, "capacity": 9, "owner_capacity": 3, + "retention_seconds": 300.0, "max_jobs": 1000, "drain_seconds": 7, "tokens": 2}) self.assertEqual((lines[1]["cancelled"], lines[1]["level"]), (0, "info")) for value in (APP, OTHER, SENTINEL, "app"): self.assertNotIn(value, text) + def test_without_an_owner_cap_the_start_line_gives_the_capacity(self): + lines = parse_log(self.main({"BRIDGE_LOG": "json", "BRIDGE_CAPACITY": "9", "BRIDGE_TOKEN": TOKEN})) + self.assertEqual((lines[0]["capacity"], lines[0]["owner_capacity"]), (9, 9)) + def serve_with_a_running_job(self, signals): """A serve_forever that first starts a long job, then signals the server and serves until it stops.""" serve_forever = server.BridgeServer.serve_forever diff --git a/worker/ai_bridge/jobs.py b/worker/ai_bridge/jobs.py index cfad449..c6e084a 100644 --- a/worker/ai_bridge/jobs.py +++ b/worker/ai_bridge/jobs.py @@ -70,6 +70,8 @@ class Settings: state_dir: str | None = None # The names of the service's tokens: a job belongs to one, and each has its own metrics series. owners: tuple[str, ...] = (DEFAULT_OWNER,) + # The most queued, running and retained jobs one owner may hold; BRIDGE_OWNER_CAPACITY. None is the capacity. + owner_capacity: int | None = None def _serve(connection, backend, model_dir): @@ -158,6 +160,9 @@ def __init__(self, settings=None): or not 1 <= self.settings.concurrency <= 8 or not 1 <= self.settings.capacity <= 1024 or not 1 <= self.settings.max_jobs <= 1000000): raise ValueError("invalid worker limits") + cap = self.settings.owner_capacity + if cap is not None and (type(cap) is not int or not 1 <= cap <= self.settings.capacity): + raise ValueError("invalid worker limits") if (self.settings.backend not in MODEL_NAMES or not math.isfinite(self.settings.retention_seconds) or self.settings.retention_seconds <= 0): @@ -288,6 +293,11 @@ def _accept(self, task, payload, timeout_ms, idempotency_key, owner): # A repeat adds no job, so it is answered even at capacity. if (owner, idempotency_key) in self._keys: return self._public(self._jobs[self._keys[owner, idempotency_key]]) + # Checked before the total: one of this owner's jobs expiring frees a slot under both limits. + cap = self.settings.owner_capacity + if cap is not None and sum(job["_owner"] == owner for job in self._jobs.values()) >= cap: + raise CapacityError("owner job capacity reached; wait for its retained jobs to expire", + self._retry_after(owner)) if len(self._jobs) >= self.settings.capacity: raise CapacityError("job capacity reached; wait for retained jobs to expire", self._retry_after()) now = time.monotonic() @@ -370,10 +380,11 @@ def snapshot(self): "duration": copy.deepcopy(self._durations), "queue": copy.deepcopy(self._waits), } - 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] + def _retry_after(self, owner=None): + """Seconds until the first retained job (of this owner, when given) expires; 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 and owner in (None, job["_owner"])] if not finished: return 1 return max(1, math.ceil(min(finished) + self.settings.retention_seconds - time.monotonic())) diff --git a/worker/ai_bridge/log.py b/worker/ai_bridge/log.py index 26c5c91..102d380 100644 --- a/worker/ai_bridge/log.py +++ b/worker/ai_bridge/log.py @@ -20,7 +20,8 @@ EVENTS = { "request": ("method", "route", "status", "duration_ms", "owner"), "job": ("job_id", "task", "owner", "job_status", "queue_ms", "run_ms"), - "start": ("backend", "concurrency", "capacity", "retention_seconds", "max_jobs", "drain_seconds", "tokens"), + "start": ("backend", "concurrency", "capacity", "owner_capacity", "retention_seconds", "max_jobs", "drain_seconds", + "tokens"), "stop": ("cancelled",), "drain": ("duration_ms", "running", "interrupted"), } @@ -57,7 +58,8 @@ def _among(choices): "duration_ms": _duration, "queue_ms": _duration, "run_ms": _duration, "owner": _matches(_NAME), "job_id": _matches(_JOB_ID), "task": _among(TASKS), "job_status": _among(STATUSES), "backend": _among(MODEL_NAMES), "retention_seconds": _duration, - "concurrency": _count, "capacity": _count, "max_jobs": _count, "drain_seconds": _count, "tokens": _count, + "concurrency": _count, "capacity": _count, "owner_capacity": _count, "max_jobs": _count, "drain_seconds": _count, + "tokens": _count, "cancelled": _count, "running": _count, "interrupted": lambda value: type(value) is bool, } diff --git a/worker/ai_bridge/server.py b/worker/ai_bridge/server.py index cb667f4..d0c1c10 100644 --- a/worker/ai_bridge/server.py +++ b/worker/ai_bridge/server.py @@ -439,7 +439,10 @@ def main(): test_tasks=os.environ.get("BRIDGE_TEST_TASKS") == "1", max_jobs=_number("BRIDGE_WORKER_MAX_JOBS", "1000", int), state_dir=os.environ.get("BRIDGE_STATE_DIR") or None, + owner_capacity=_number("BRIDGE_OWNER_CAPACITY", None, int) if "BRIDGE_OWNER_CAPACITY" in os.environ else None, ) + if settings.owner_capacity is not None and not 1 <= settings.owner_capacity <= settings.capacity: + raise SystemExit(f"BRIDGE_OWNER_CAPACITY must be an integer from 1 to BRIDGE_CAPACITY ({settings.capacity})") drain_seconds = _number("BRIDGE_DRAIN_SECONDS", "30", int) if not 0 <= drain_seconds <= 300: raise SystemExit("BRIDGE_DRAIN_SECONDS must be an integer from 0 to 300") @@ -460,6 +463,7 @@ def main(): log.configure(log_mode) try: log.event("start", backend=settings.backend, concurrency=settings.concurrency, capacity=settings.capacity, + owner_capacity=settings.capacity if settings.owner_capacity is None else settings.owner_capacity, retention_seconds=settings.retention_seconds, max_jobs=settings.max_jobs, drain_seconds=drain_seconds, tokens=len(tokens)) with bridge as server: