diff --git a/.socraticodecontextartifacts.json b/.socraticodecontextartifacts.json index 580b2ef..dd0a60e 100644 --- a/.socraticodecontextartifacts.json +++ b/.socraticodecontextartifacts.json @@ -13,12 +13,17 @@ { "name": "design-plans", "path": "./docs/plans/", - "description": "Founding MVP design (2026-06-25, adopted). The command/fact contracts as shipped in co-core v0.7.0 — ContentFetchCommand (URL-addressed, command_id as the wire idempotency key) and BlobAvailableEvent (content_fingerprint, blob_uri, size_bytes, media_type, url, optional command_id) — the stream taxonomy and DLQ naming, the temp-storage interface, the two-level idempotency argument, why fingerprint parity dissolves when Replicator is the sole fetcher, explicit MVP scope cuts, and the build sequence. The founding plan's three open questions were settled 2026-07-31 in a companion doc: blob_uri is a file:// URI over a two-level sharded, content-addressed path (///.bin, constant .bin extension, temp-file + os.replace so presence means complete); the MVP command issuer is an in-repo scripts/seed_fetch.py rather than Watcher or Archiver; and the MVP boundary is stop-at-fact, with archiver-writeback reclassified from MVP+ to won't-do — Archiver consumes blob_available and writes the SourceRevision itself (archiver#118, which also carries the info_source_id contract gap). Blob retention is out of MVP scope and tracked in #5." + "description": "Founding MVP design (2026-06-25, adopted). The command/fact contracts as shipped in co-core v0.7.0 \u2014 ContentFetchCommand (URL-addressed, command_id as the wire idempotency key) and BlobAvailableEvent (content_fingerprint, blob_uri, size_bytes, media_type, url, optional command_id) \u2014 the stream taxonomy and DLQ naming, the temp-storage interface, the two-level idempotency argument, why fingerprint parity dissolves when Replicator is the sole fetcher, explicit MVP scope cuts, and the build sequence. The founding plan's three open questions were settled 2026-07-31 in a companion doc: blob_uri is a file:// URI over a two-level sharded, content-addressed path (///.bin, constant .bin extension, temp-file + os.replace so presence means complete); the MVP command issuer is an in-repo scripts/seed_fetch.py rather than Watcher or Archiver; and the MVP boundary is stop-at-fact, with archiver-writeback reclassified from MVP+ to won't-do \u2014 Archiver consumes blob_available and writes the SourceRevision itself (archiver#118, which also carries the info_source_id contract gap). Blob retention is out of MVP scope and tracked in #5." }, { "name": "issuer-contract", "path": "./docs/contracts/content-fetch-issuer-contract.md", - "description": "Normative contract for any service publishing content.fetch commands (seed_fetch.py today, Watcher from Phase 4) — and its permanent home, linked to rather than copied by issuer-side repos. Opens with the frame shape, because getting it wrong is the likeliest and quietest mistake: what lands on the stream is a co-core envelope produced by to_wire (key, payload as JSON, event_type, schema_version, occurred_at, content_type), never the model's fields flattened, and an XADD of top-level command_id/url fails from_wire into content.fetch.dlq silently. key is derived by to_wire (command_id for a command, content_fingerprint for a fact) and is not what Replicator dedupes on — that is the decoded payload.command_id — but it is what makes a DLQ entry correlatable, so an issuer can tail content.fetch.dlq with a plain XREAD and close a pending entry deterministically instead of waiting out a timeout. Then why the wire carries no info_source_id and what that pushes onto the issuer: mint a fresh ULID command_id per fetch occasion (a resource-stable id is silently deduped away for 24h by replicator:cmd:); url is one-to-many onto InfoSources and therefore not a correlation key; persist the command_id -> domain map before publishing, outbox-style; correlation must be idempotent because a crash between publish and the dedupe SET emits a second fact under the same command_id; consumers must not dedupe their inbox on content_fingerprint, which is storage identity rather than correlation identity; there is no fetch_failed fact, so silence is the only failure signal and the issuer needs its own pending-entry reaper that re-issues rather than concluding loss; blob_uri is a host-local file:// URI whose 7-day TTL runs from last fetch-reference, not last consumer read. Carries the full failure taxonomy (which conditions dead-letter to content.fetch.dlq silently, which retry indefinitely, which park in the PEL under the byte ceiling), what Replicator does and does not guarantee, the unauthenticated-capability trust posture resting on Archiver's localhost bus ACL, and the deferred fetch_failed fact as a co-core change. Written for #8." + "description": "Normative contract for any service publishing content.fetch commands (seed_fetch.py today, Watcher from Phase 4) \u2014 and its permanent home, linked to rather than copied by issuer-side repos. Opens with the frame shape, because getting it wrong is the likeliest and quietest mistake: what lands on the stream is a co-core envelope produced by to_wire (key, payload as JSON, event_type, schema_version, occurred_at, content_type), never the model's fields flattened, and an XADD of top-level command_id/url fails from_wire into content.fetch.dlq silently. key is derived by to_wire (command_id for a command, content_fingerprint for a fact) and is not what Replicator dedupes on \u2014 that is the decoded payload.command_id \u2014 but it is what makes a DLQ entry correlatable, so an issuer can tail content.fetch.dlq with a plain XREAD and close a pending entry deterministically instead of waiting out a timeout. Then why the wire carries no info_source_id and what that pushes onto the issuer: mint a fresh ULID command_id per fetch occasion (a resource-stable id is silently deduped away for 24h by replicator:cmd:); url is one-to-many onto InfoSources and therefore not a correlation key; persist the command_id -> domain map before publishing, outbox-style; correlation must be idempotent because a crash between publish and the dedupe SET emits a second fact under the same command_id; consumers must not dedupe their inbox on content_fingerprint, which is storage identity rather than correlation identity; there is no fetch_failed fact, so silence is the only failure signal and the issuer needs its own pending-entry reaper that re-issues rather than concluding loss; blob_uri is a host-local file:// URI whose 7-day TTL runs from last fetch-reference, not last consumer read. Carries the full failure taxonomy (which conditions dead-letter to content.fetch.dlq silently, which retry indefinitely, which park in the PEL under the byte ceiling), what Replicator does and does not guarantee, the unauthenticated-capability trust posture resting on Archiver's localhost bus ACL, and the deferred fetch_failed fact as a co-core change. Written for #8." + }, + { + "name": "boundaries-charter", + "path": "./docs/contracts/replicator-boundaries.md", + "description": "Normative charter for what Replicator is allowed to become \u2014 sibling to the issuer contract, which settles the wire while this settles the service. The rule: Replicator owns the mechanics of acquiring bytes from a network and holding them briefly, never why, when, or what they mean. Three tests for any proposed capability, field, or setting (durable per-resource history -> issuer; cross-command coordination over a resource only the fetcher can see -> Replicator; expressible without domain vocabulary or it is the issuer's), resolved when both fire by mechanism-to-Replicator / policy-to-the-issuer / config-over-the-bus. Worked examples: per-host politeness (the hard case, where the tests do not resolve to one side) and conditional GET (ETag/Last-Modified are durable per-resource history, so the fact returns them and the next command replays them as headers \u2014 conditional GET with zero per-URL state here). Config taxonomy of three channels (env for host facts, command for the occasion, a policy stream for cluster policy needing cross-command state) with a fourth \u2014 an inbound admin HTTP API \u2014 rejected by name; ingress is read-only liveness and self-description, and the invariant that matters is asserted about the worker, which binds no port, not the dev-only FastAPI app. Specifies the agreed-but-unbuilt content.fetch.policy stream: last-write-wins per host, replayed from 0-0 at boot, MAXLEN-trimmed because periodic full republication onto an untrimmed stream is unbounded growth, a Redis hash rejected for having no schema_version and coupling to a key name, a conservative default for unknown hosts, and the enforcement idiom (park in the PEL, return via claim_stale) whose 60s granularity misses Watcher's 1.0s baseline by 60x \u2014 so the stream is a precondition of the Phase 4 cutover, not a follow-on (watcher#245). Records blob_uri's host-local file:// coupling as a tracked violation (#7) rather than omitting it. Enforced by tests/test_boundaries.py in CI: eight invariants including an AST scan of src/ for domain nouns in identifiers and string literals, a recursive route walk, no locally-defined wire models (payload shapes, not producer-owned token vocabularies like FailureReason), and REPLICATOR_-prefixed settings with BUILD_ID exempted by name. Written for #12." }, { "name": "commands", @@ -38,17 +43,17 @@ { "name": "wheelhouse-sync", "path": "./scripts/sync_wheelhouse.py", - "description": "Mirrors the private cannobserv package index (gs://co-gcs-pypi/wheels/) into ./.wheelhouse, from which uv resolves co-core and co-core-aio via find-links. Must run before uv sync and must not import the project, since the project's deps are what it provides — hence the isolated `uv run --no-project --with google-cloud-storage` invocation. ADC auth: the SA key on the VM, keyless WIF in CI." + "description": "Mirrors the private cannobserv package index (gs://co-gcs-pypi/wheels/) into ./.wheelhouse, from which uv resolves co-core and co-core-aio via find-links. Must run before uv sync and must not import the project, since the project's deps are what it provides \u2014 hence the isolated `uv run --no-project --with google-cloud-storage` invocation. ADC auth: the SA key on the VM, keyless WIF in CI." }, { "name": "redis-floor-guard", "path": "./scripts/check_redis_floor.sh", - "description": "ExecStartPre guard asserting the Redis >=7.0 change-bus server floor. Replicator is the cluster's first user of AsyncBusConsumer.claim_stale, which reads XAUTOCLAIM's three-element reply — the deleted-ids element added in Redis server 7.0 — so a lower version breaks crash recovery. Soft on an unreachable broker (it may still be starting), fatal on a genuine downgrade." + "description": "ExecStartPre guard asserting the Redis >=7.0 change-bus server floor. Replicator is the cluster's first user of AsyncBusConsumer.claim_stale, which reads XAUTOCLAIM's three-element reply \u2014 the deleted-ids element added in Redis server 7.0 \u2014 so a lower version breaks crash recovery. Soft on an unreachable broker (it may still be starting), fatal on a genuine downgrade." }, { "name": "ci-pipeline", "path": "./.github/workflows/ci.yml", - "description": "GitHub Actions CI. Keyless Workload Identity Federation auth to GCP for the wheelhouse mirror, a fail-fast guard asserting the GCP_WIF_PROVIDER org variable is visible to this repo, ruff check and format gates, a co-core extras import smoke proving [extract] and [bus] wire up in a clean environment, and pytest with the coverage gate. Deliberately no Postgres service and no alembic steps — Replicator is DB-free." + "description": "GitHub Actions CI. Keyless Workload Identity Federation auth to GCP for the wheelhouse mirror, a fail-fast guard asserting the GCP_WIF_PROVIDER org variable is visible to this repo, ruff check and format gates, a co-core extras import smoke proving [extract] and [bus] wire up in a clean environment, and pytest with the coverage gate. Deliberately no Postgres service and no alembic steps \u2014 Replicator is DB-free." }, { "name": "dependency-contract", diff --git a/AGENTS.md b/AGENTS.md index d2bbd78..45ad6b6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,6 +66,7 @@ src/worker/loop.py — The consume path: poll → dispatch → ack, DLQ, dedup src/worker/handler.py — The byte path behind the Handler seam: fetch → fingerprint → store → publish src/worker/reporter.py — The failure fact behind the FailureReporter seam: fetch_failed on content.blobs src/worker/retention.py — The sweep task: cadence, usage accounting, ceiling reporting +src/worker/pacing.py — Per-host request spacing; the interim politeness default (#12) src/storage/ — Temp storage; BlobStore protocol + the local-FS backend src/storage/base.py — BlobStore protocol (store / exists / open) src/storage/local.py — Content-addressed local backend; file:// URIs, sharded paths @@ -81,7 +82,7 @@ scripts/ — sync_wheelhouse.py, check_redis_floor.sh, seed_fetch.py scripts/seed_fetch.py — the MVP command issuer; publishes content.fetch, --watch tails the facts tests/ — Mirrors src/ structure; integration tests in `@pytest.mark.integration` docs/ — Reference docs (COMMANDS, SKILLS) -docs/contracts/ — Normative bus contracts; the issuer-facing half of the wire, linked to from sibling repos +docs/contracts/ — Normative contracts, linked to from sibling repos: the issuer-facing half of the wire, and the boundaries charter (what Replicator may become) docs/plans/ — Implementation plans deploy/ — Systemd unit + deployment config .wheelhouse/ — Local mirror of the private cannobserv index (git-ignored except .gitkeep) @@ -169,6 +170,7 @@ In `/etc/replicator/.env` (read by the service): - `REPLICATOR_CONSUMER_NAME` — this worker's identity within the group; defaults to `replicator@`. Two workers must never share one — Redis tracks pending entries per consumer name, and a shared name makes independent `claim_stale` recovery impossible - `REPLICATOR_CONSUMER_START_ID` — group start position; default `"$"` (new messages only), `"0"` drains the backlog. Applies **only at group creation** — once `replicator.fetch` exists, changing this also needs a manual `XGROUP SETID` - `REPLICATOR_READ_BLOCK_MS` — blocking-read window; default `5000`. Bounds worst-case shutdown latency, so the unit's `TimeoutStopSec` must exceed it plus the handler budget **plus an in-flight sweep** — `asyncio.to_thread` puts the tree walk beyond cancellation, so SIGTERM waits it out +- `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` — minimum spacing between two requests to the same host; default `1.0`. The **interim** politeness default (#12) standing in for the `content.fetch.policy` stream: enforcement is mechanism and lives here, the numbers are the issuer's and do not yet travel over the bus. 1.0 is Watcher's own `DEFAULT_MIN_INTERVAL`, chosen because it invents nothing — the cutover changes who paces, not how much. A wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through in the handler; a longer one raises `TransientFetchError` and parks the command, so the effective floor on a *parked* wait is `REPLICATOR_CLAIM_MIN_IDLE_MS`. `0` disables pacing outright — an operator escape hatch, and a deployment that sets it is choosing to have no politeness at all. Capped at `3600`: past an hour the command parks and re-parks without ever dead-lettering (transient failures are exempt from the delivery ceiling) while the issuer's reaper concludes loss, so a fat-fingered extra zero should fail at startup rather than read as healthy - `REPLICATOR_CLAIM_MIN_IDLE_MS` — idle time before a pending entry may be reclaimed; default `60000`. Doubles as the retry cadence - `REPLICATOR_MAX_DELIVERY_ATTEMPTS` — delivery ceiling for *unclassified* failures before DLQ; default `5`. Counted from XPENDING's delivery counter, which only advances on a reclaim ⇒ a bound in time, not retries - `REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` / `REPLICATOR_ERROR_BACKOFF_MAX_SECONDS` — backoff for a poll *cycle* that raised (broker outage); defaults `1.0` / `30.0`, escalating `base * 2**(n-1)` @@ -182,10 +184,12 @@ In `/etc/replicator/.env` (read by the service): Replicator is a **consumer** first. Follow the conventions co-core and the archiver producer established: - **The issuer contract is written down and lives here.** `docs/contracts/content-fetch-issuer-contract.md` is the normative statement of what a `content.fetch` producer must do — per-occasion `command_id`, `url` is not a correlation key, persist the `command_id → domain` map before publishing, correlate idempotently, and keep a reaper as a backstop for the outcomes no fact can carry. Issuer-side repos (Watcher, Phase 4) link to it rather than copying it. Anything asserted there is asserted about this repo's code; change one, change both (#8). +- **What Replicator is allowed to become is also written down.** `docs/contracts/replicator-boundaries.md` is the sibling charter: mechanism to Replicator, policy to the issuer, config over the bus, and an inbound admin HTTP API rejected by name. Run its three tests against any proposed capability, field, or setting before writing code — a database, domain vocabulary, or a write route is reached one defensible step at a time, not in one commit. `tests/test_boundaries.py` enforces eight invariants in CI, including the one that catches the regression review misses: an AST scan of `src/` for domain nouns in identifiers *and* string literals. Known violation recorded and pinned rather than omitted — `blob_uri` is host-local `file://` (#7). Change the charter and the tests together (#12). - **`content.blobs` carries both outcomes.** `blob_available` on success, `fetch_failed` on a command closed without bytes (#9, co-core cannobserv#270 — v0.7.2). One stream so an issuer's single consumer group sees either. The reason is named at the *raise site* (`PermanentFetchError.reason`), never recovered from a message string, because three unrelated permanent conditions share one exception type. Three rows stay DLQ-only and permanently silent, and only one of them for want of an id: a frame that did not decode, a command whose `command_id` is blank (refused before the fetch — an empty id would otherwise take the dedupe key `replicator:cmd:` and make every later blank-id command a silent no-op, CR #6), and a frame that decoded to a **non-command payload** — the last is unreportable not because it lacks a `command_id` but because any it carries is *another command's* (`BlobAvailableEvent`'s names one that succeeded), so a terminal fact keyed on it would contradict a fact the issuer already applied (CR #1). `_close` refuses a correlator-less report at the one choke point rather than at each call site. Non-terminal facts are deferred (#9 §3): the stream is broadcast and nothing trims it, so a fact per reclaim during an origin outage is unbounded growth. `src/worker/reporter.py`. - **`blob_available` carries the fetch, not just the bytes.** Six optional fields beyond the blob itself (#10, cannobserv#271 + #279 — v0.7.5): `final_url`, `status_code`, `fetched_at`, `content_type_raw`, `etag`, `last_modified`. Each is what Replicator holds at publish time and a broadcast consumer cannot recover now that fetching lives here rather than in Watcher. **`None` means nobody said it** — never a stand-in: `final_url` is never backfilled from `command.url` (an issuer could no longer tell "landed where I asked" from "nobody knows"), and `content_type_raw` is never backfilled with `DEFAULT_MEDIA_TYPE` (which is the value a consumer reads as "unknown, guess from the URL"). `media_type` keeps its normalized semantics beside the raw channel; the two are not interchangeable. `fetched_at` is stamped where the fetch *returns*, not at publish — `occurred_at` under a reclaim is minutes late. `status_code` is always 2xx on this fact, so it distinguishes 200 from 203/206 and is not a success branch. The passthroughs are **dropped over `MAX_HEADER_VALUE_LENGTH`, never truncated**: these are origin-controlled strings on a stream nothing trims, and a truncated ETag replayed in an `If-None-Match` is a validator that can never match. `src/worker/handler.py::_passthrough`. - **The failure fact is a seam, not a call in the loop.** `FailureReporter` is injected exactly as `Handler` is, so `loop.py` stays ignorant of `content.blobs` and `blobs_topic` stays a defaulted argument a live-broker test can move. The loop owns the *decision* (`terminal` is "did this hit the delivery ceiling", which only the loop knows); the reporter owns the *publish*. `_close()` publishes then dead-letters, so **fact-before-ack** cannot be got wrong one call site at a time — `dead_letter` acks inside itself, and a fact published after it is lost outright on a crash. A failed fact-publish is **swallowed**, deliberately asymmetric with the byte path's `_publish`: there raising prevents an orphan blob, here the DLQ entry is already the durable record and raising would burn the delivery ceiling to reach the same DLQ minutes later. - **A command shapes its own fetch, and everything unsendable is refused rather than fixed.** `headers` and `timeout_seconds` (#11, cannobserv#272 — v0.7.3) reach the driver as `FetchContent.headers` / `.timeout`; omitted means the pre-#11 wire byte-for-byte. Header names are **lower-cased before the merge** — `AsyncFetchDriver` merges `{"user-agent": DEFAULT, **effect.headers}` case-*sensitively*, and httpx does not resolve the collision: an unfolded `User-Agent` puts **two** field lines on the wire, default first, for the origin to disambiguate (measured, not inferred). That is exactly the fingerprint-continuity case Watcher needs at cutover. Refusals are **`PermanentFetchError(INVALID_REQUEST_OPTIONS)` raised before the fetch**: hop-by-hop and httpx-derived names (`host`, `content-length`, `proxy-*`, …), non-token names (padding included — OWS is a *value* rule, not a name one), values outside **printable US-ASCII**, case-collisions, over `MAX_REQUEST_HEADERS`/`MAX_REQUEST_HEADER_BYTES`, and a timeout that is non-finite, ≤ 0, or over `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS`. Refused, never stripped or clamped — same argument as dropping an over-long passthrough rather than truncating it: a change to the fetch the issuer cannot see is one it cannot account for in its own fingerprints. **The value charset is pinned to what httpx can actually send, not to RFC 9110** — see `_HEADER_VALUE`'s comment for why each edge sits where it does (CR #1). The rule the guard exists to enforce: a value httpx refuses does not fail as a classified fetch error, so an unguarded one closes the command under the wrong reason or retries forever. `tests/worker/test_handler_request_options.py` walks the whole single-byte range against `httpx.Request` so the two cannot drift again. Validation runs **ahead of the storage ceiling** so a permanently-bad command does not park in the PEL waiting for a sweep. Header **names only** are logged, never values. `src/worker/handler.py::_request_options`. +- **Politeness is enforced here and decided elsewhere, and the interim is a default rather than a decision.** `src/worker/pacing.py` holds a host → last-request map in memory (derived, bounded, rebuildable — one of the three state shapes the boundaries charter permits) and reports a wait; `handler.py::_pace` spends it. **Two ways to spend it, split by duration, because neither works alone on a serial consume path**: a wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through, a longer one raises `TransientFetchError` and parks the command for `claim_stale`. Park-only was the obvious design and is wrong by 60× — a parked wait cannot be shorter than `REPLICATOR_CLAIM_MIN_IDLE_MS` (60 s) while the normal interval is 1 s, so every host would have been paced at 1/60th of today's rate, silently and in the safe direction. Sleep-only holds every *other* host's commands, and a SIGTERM, behind one origin's politeness. Transient in both directions so being polite can never burn the delivery ceiling. The pacer is built from settings when not injected — the seam fails **open**, and a byte path that quietly stopped pacing is indistinguishable from one that is working. Only a request that actually goes out calls `record()`: stamping a parked attempt would space the origin from requests it never received. **Keyed on the host asked for, not the host reached** — httpx follows redirects inside the driver, so URLs funnelling into one portal hit it at N× the intended rate; recorded as a known limitation in the charter rather than fixed here, because the fix breaks "one request, one record". Signal: `paced_seconds` on the byte path's success line (per-fetch, correlate with `duration_ms`), `tracked_hosts` on `_pace`'s own INFO line, which fires only when a wait was actually spent. This is the **interim** for the Phase 4 cutover, not the design — the numbers belong to the issuer and reach Replicator over `content.fetch.policy` (#12, watcher#245, cannobserv#285). - **The timeout ceiling and the unit's `TimeoutStopSec` are one decision.** A command's own timeout replaced the driver's fixed 30 s as the handler's worst-case budget, and SIGTERM waits out the message in flight, so `TimeoutStopSec` must exceed `REPLICATOR_READ_BLOCK_MS` + `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` + an in-flight sweep. `tests/test_deploy.py` enforces the first two terms; the third is the margin. - **At-least-once ⇒ idempotent.** Two idempotency keys, two levels: the **command** dedupes on `command_id`, the **fact** on `content_fingerprint` — and `fetch_failed` on neither, keyed `command_id:occurred_at` so a consumer's dedup-on-key cannot collapse a multi-emission sequence and drop the terminal event. The two serve different purposes and are not interchangeable — the fingerprint is *storage* identity, `command_id` is *correlation* identity, and a consumer deduping its inbox on the fingerprint silently loses the second of two commands that fetched identical bytes. Content-addressed storage makes re-storing identical bytes a no-op regardless. - **Consumers must be idempotent; producers own the outbox.** The cluster split (parent strategy, "Delivery + correctness") assigns the transactional outbox to producers with a DB system of record. Replicator has none — its durable record of intent is the consumer group's PEL, recovered via `claim_stale`. Do not add a Postgres outbox to the consume path. diff --git a/README.md b/README.md index 22f20a8..7933b55 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,7 @@ rather than deleting bytes a consumer was promised. | `REPLICATOR_CONSUMER_NAME` | `replicator@` | This worker's identity in the group — never share one | | `REPLICATOR_CONSUMER_START_ID` | `$` | Group start position. Applies only at group *creation*; changing it later also needs `XGROUP SETID` | | `REPLICATOR_READ_BLOCK_MS` | `5000` | Blocking-read window. Bounds shutdown latency, so the unit's `TimeoutStopSec` must exceed it plus `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` (the handler's worst-case budget since #11) and an in-flight sweep — `tests/test_deploy.py` pins the first two terms | +| `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` | `1.0` | Minimum spacing between two requests to the same host (#12). The interim politeness default until the numbers travel over the bus; matches Watcher's own `DEFAULT_MIN_INTERVAL` so the Phase 4 cutover changes who paces, not how much. A wait under `REPLICATOR_READ_BLOCK_MS` is slept through, a longer one parks the command for the next reclaim. `0` disables pacing entirely; capped at `3600` | | `REPLICATOR_CLAIM_MIN_IDLE_MS` | `60000` | Idle time before a pending entry may be reclaimed — also the retry cadence | | `REPLICATOR_MAX_DELIVERY_ATTEMPTS` | `5` | Deliveries of an *unclassified* failure before the DLQ | | `REPLICATOR_DEDUPE_TTL_SECONDS` | `86400` | Lifetime of the `replicator:cmd:` dedupe key | diff --git a/docs/contracts/content-fetch-issuer-contract.md b/docs/contracts/content-fetch-issuer-contract.md index 337f5ec..b0886f1 100644 --- a/docs/contracts/content-fetch-issuer-contract.md +++ b/docs/contracts/content-fetch-issuer-contract.md @@ -8,6 +8,10 @@ from the code the day it was written. **Audience:** any service that publishes a `ContentFetchCommand`. Today that is [`scripts/seed_fetch.py`](../../scripts/seed_fetch.py). From Phase 4 it is Watcher. +**Sibling document.** This settles the *wire*. [`replicator-boundaries.md`](replicator-boundaries.md) +settles the *service* — what Replicator is allowed to become, and therefore which proposed fields +this contract will never grow. Read that one before proposing a payload addition (#12). + **Changing this document.** "Link, don't copy" only holds if a change reaches the issuers. A change to any MUST, or to the failure taxonomy, is announced on the open issuer-side trackers — currently [CannObserv/watcher#241](https://github.com/CannObserv/watcher/issues/241) — in the same change that @@ -464,6 +468,18 @@ A consumer on another host cannot open it, and nothing on the wire says so. - **No failure fact for a command whose `command_id` is blank** — nothing to correlate one on at all. It is dead-lettered before the fetch rather than run. MUST-1, MUST-6. - **No latency bound**, and no SLA on turnaround. +- **No promise that a burst runs at the rate it was issued (#12).** Requests to one host are + spaced by at least `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` — 1 s by default, the interim + stand-in for the politeness numbers until they travel over the bus. Publishing 100 commands + for one host means at least 100 s of fetching. Size a reaper's timeout (MUST-6) against the + depth of your own burst, not against one fetch. Commands for different hosts are unaffected + by each other. + **At the shipped defaults every wait is slept through inside the handler**, so the cost is + seconds of added turnaround and nothing else. Only when an operator configures the interval + *above* `REPLICATOR_READ_BLOCK_MS` (5 s) does a paced command instead stay pending for the + next reclaim, which moves the cadence from seconds to a minute. That is a deployment + decision, not a default — but it is the one that changes what a reaper should expect, so it + is stated here rather than left to be discovered. - **No ordering.** Two commands issued in sequence may produce facts in either order. - **No cross-command dedupe.** Two `command_id`s for one URL are two fetches and two facts, by design — that is what makes MUST-1 work. diff --git a/docs/contracts/replicator-boundaries.md b/docs/contracts/replicator-boundaries.md new file mode 100644 index 0000000..b45d7d0 --- /dev/null +++ b/docs/contracts/replicator-boundaries.md @@ -0,0 +1,308 @@ +# Replicator boundaries charter + +**Status:** normative. **Home:** this file, in the Replicator repo — the invariants are +about this repo's code and are enforced by +[`tests/test_boundaries.py`](../../tests/test_boundaries.py). Sibling repos link here rather +than copying. + +**Audience:** anyone proposing a capability, payload field, or setting for Replicator — +including a future maintainer of this repo, who is the likelier author of the drift this +document exists to prevent. + +**Relationship to the issuer contract.** +[`content-fetch-issuer-contract.md`](content-fetch-issuer-contract.md) settles the *wire*: +what a producer must do so a fact is never lost. This settles the *service*: what Replicator +is allowed to become. They are different documents and until #12 only one of them existed. + +**Why now.** Phase 4 ([CannObserv/watcher#241](https://github.com/CannObserv/watcher/issues/241)) +is the first time capability gets reallocated across the boundary, and the reallocation is not +one-directional: politeness enforcement wants to move *in*, alternate fetch drivers want to +move *in*, and per-URL validator state must be kept *out*. Each arrives as a reasonable-looking +field or a small settings table, and the cumulative result — a Replicator with a database, +domain vocabulary, and an admin API — is reached one defensible step at a time. + +The invariants are all currently **true**. This is drift prevention from a good position, not +remediation. + +--- + +## The rule + +> **Replicator owns the mechanics of acquiring bytes from a network and holding them briefly. +> It never owns why, when, or what they mean.** + +## The three tests + +Any proposed capability, field, or setting runs these in order: + +1. **Does it need durable per-resource history?** → **issuer.** Replicator's state must be + exactly one of: content-addressed on disk (rebuildable by re-fetch), in-memory derived + (rebuildable by replay), or in the broker (PEL, dedupe keys). State outside those three + *is* a database, whatever it is called. +2. **Does it need cross-command coordination over a resource only the fetcher can see?** + (a host's tolerance, the disk, a connection pool, a browser pool) → **Replicator.** Nobody + else can see it, and N issuers being polite independently is a fiction. +3. **Can it be expressed without domain vocabulary?** If it needs the words InfoSource, + InfoItem, WatchedItem, aspect, tenant → **issuer**, always. + +Tests 1 and 2 can both fire. When they do: + +> **Mechanism to Replicator. Policy to the issuer. Config travels over the bus.** + +### Worked example — per-host politeness + +The case worth studying, because the tests do **not** resolve it to one side and the +mechanism/policy split is what finishes the job. + +| Test | Politeness | +|---|---| +| 1 — durable per-resource history? | **No.** Per-*host*, and rebuildable from replay. Not the issuer on this test. | +| 2 — cross-command coordination over a resource only the fetcher can see? | **Yes.** A host's tolerance. "N issuers being polite independently is a fiction" is exactly this row. | +| 3 — expressible without domain vocabulary? | **Yes.** A hostname is a network fact, not an InfoSource. | + +Test 2 fires alone, but a naive reading of it — "politeness is Replicator's, done" — would +move the *numbers* here too, and the numbers are operator policy that lives in Watcher's +`Domain` table today. The resolution is the split: Replicator **enforces**, the issuer +**decides**, and the decision travels over the bus (see **The policy stream**). + +A reader who only skims the easy case below learns the pattern and misses the rule. + +### Worked example — conditional GET + +ETag/Last-Modified *look* like fetcher state. Test 1 sends them to the issuer: they are +durable per-resource history. So the fact returns them +([cannobserv#271](https://github.com/CannObserv/cannobserv/issues/271)) and the next command +replays them as request headers +([cannobserv#272](https://github.com/CannObserv/cannobserv/issues/272)). Replicator gains +conditional GET while holding zero bytes of per-URL history. Cite this one when a proposal +argues that some small table would be simpler. + +## Where things live + +**Replicator (mechanism):** fetch execution, redirect following, retry cadence, +fingerprinting, temp storage and its lifetime, disk ceiling, per-host pacing *enforcement*, +and — future — alternate fetch drivers (browser, archival capture) and escalation between +them. + +**Issuer (policy/domain):** scheduling and cadence, extraction specs, change semantics, +notification, correlation, pending maps and reapers, per-URL validator state, and the +politeness *numbers*. + +**Never Replicator's:** domain identity, extraction specs, change decisions, notification, +InfoSource correlation, scheduling. Named explicitly so a proposal has to argue against a +written line rather than into a vacuum. + +## Config taxonomy + +| Channel | Carries | Examples | +|---|---|---| +| **env** | facts about *this host* | blob TTL, disk ceiling, claim cadence, dedupe TTL | +| **command** | this occasion | `headers`, `timeout_seconds`, later a fetch `strategy` | +| **config stream** | cluster policy needing cross-command state | per-host politeness | + +Env settings carry the `REPLICATOR_` prefix so they never collide with a sibling service on +the shared VM. **`BUILD_ID` is the one exemption** and is deliberate: the systemd unit's +`ExecStartPre` stamps it generically across the cluster's services, so prefixing it here would +mean a per-service variable name for one git SHA. It is exempted by name in the test, not +waved through — an exemption list that grows is this convention ending quietly. + +**The fourth channel is rejected by name: an inbound admin HTTP API.** It is the easy path for +every future config need and it ends the property that makes this service testable and +relocatable. + +**What that forbids is a write surface, not self-description.** The enforced invariant is: + +> No route accepts state-changing input. Ingress is read-only liveness and self-description. + +FastAPI's `/docs`, `/redoc`, `/docs/oauth2-redirect` and `/openapi.json` are read-only and +describe the one real route, so they are allowlisted rather than switched off. Disabling them +outside dev would add a config knob this taxonomy then has to place, and would make the app +under test differ from the app that ships — for a surface no deployment serves. + +**The invariant that matters is about the worker, not the app.** `src/api/` is dev-only; +`replicator.service` runs the bus consumer, which binds no port. An ingress assertion scoped +to the FastAPI app could pass forever while an admin listener grew inside `src/worker/`. Both +are asserted. + +### The policy stream (agreed, not yet built) + +Upstream model tracked at +[CannObserv/cannobserv#285](https://github.com/CannObserv/cannobserv/issues/285) — +`CONTENT_FETCH_POLICY`, `FetchPolicyEvent`, and the two gaps it surfaced: `BusPublish` cannot +carry a `MAXLEN`, so the trimming this section requires is currently unexpressible through +co-core, and `AsyncBusConsumer` is group-only, so the replay-then-tail read has no driver seam. + +`content.fetch.policy` — last-write-wins per host key. Replicator replays it from `0-0` at +boot into memory and tails it thereafter. No DB, rebuildable, no inbound calls, and the data +stays owned by its producer. + +Four properties that must hold or the design fails quietly: + +- the **owner republishes the full set** periodically and on change, so boot replay never + depends on broker retention; +- the producer therefore XADDs with **`MAXLEN ~ N`**, N sized above the host count. Periodic + republication onto a stream nothing trims is unbounded growth — the argument that deferred + non-terminal facts in #9 §3 — and here it costs boot time too, since replay length would + grow with policy history rather than with host count; +- an **unknown host resolves to a conservative default**, never to unlimited; +- **enforcement has a rate floor of `REPLICATOR_CLAIM_MIN_IDLE_MS`** (default 60 s). See below. + +**Why a stream and not a Redis hash.** Broker state is explicitly permitted by test 1, so +`HGETALL` on a per-host hash is a reasonable reach and will be proposed. It is rejected +because it has no `schema_version`, no co-core model, and it couples Replicator to a key name +another service writes instead of to a payload contract. The stream keeps the same validation +posture as every other wire input. + +**Producer:** Watcher for Phase 4 (the numbers live in its `Domain` table today). Because it +is bus-delivered, the producer can later move to Archiver — the natural home if a second +issuer ever exists — with no Replicator change at all. That portability is the reason for the +indirection. + +**Enforcement mechanism:** when a host's bucket is dry, leave the message in the PEL and let +the reclaim bring it back. That is already the idiom for the disk ceiling — a policy check in +the handler, not new machinery — and it inherits the ceiling's safety property: the raise is a +`TransientFetchError`, which is exempt from the delivery ceiling, so a paced command cannot +DLQ for being paced. + +**It also inherits the ceiling's granularity, and parking alone is therefore not a sufficient +mechanism.** A parked message returns via `claim_stale`, so the finest per-host spacing it can +express is `REPLICATOR_CLAIM_MIN_IDLE_MS` — **60 s by default**. Watcher's baseline today is +`DEFAULT_MIN_INTERVAL = 1.0` s (`src/core/rate_limiter.py`), backing off to +`BACKOFF_MAX_INTERVAL = 60.0` s. So parking matches the *backoff* case almost exactly and +misses the *normal* case by 60×: implemented naively, every host would be paced at 1/60th of +the rate the cluster runs at now. The failure is silent and in the safe direction, which is +what makes it easy to ship. + +The constraint, stated so a design has to answer it: **a serial consume path cannot both sleep +for a short wait and stay available to other hosts.** Sleeping in the handler blocks every +other command in the group — which is why parking exists — and parking cannot express a +sub-reclaim interval. + +**Resolved by splitting the wait by duration**, and shipped with the interim default below: +a wait no longer than one poll window (`REPLICATOR_READ_BLOCK_MS`) is slept through in the +handler, and anything longer parks. The bound is derived from an existing setting rather than +given its own, because it is the same quantity — a wait shorter than a poll the loop already +performs adds nothing to the shutdown latency `TimeoutStopSec` is sized for. The stop event +cuts the sleep short, and an interrupted wait is not an elapsed one: the command parks rather +than fetching unpaced on the way out. `src/worker/pacing.py`, `handler.py::_pace`. + +### The interim default (shipped) + +`REPLICATOR_MIN_HOST_INTERVAL_SECONDS`, default **1.0 s** — Watcher's own +`DEFAULT_MIN_INTERVAL`, chosen precisely because it invents nothing. The numbers are the +issuer's under this charter, so until they travel over the bus the least-wrong value is the one +the cluster already commits to; the cutover then changes *who* paces rather than *how much*. +`0` disables pacing outright, an operator escape hatch and a choice to have none. + +Consistent with the charter on both halves: enforcement is mechanism (test 2 — nobody but the +fetcher can see a host's tolerance across commands), and a single default is not policy in the +sense test 3 cares about — it names no domain concept and carries no per-host table. The state +is a host → last-request map in memory: derived, bounded by pruning, and rebuildable by replay, +which is the second of the three permitted state shapes. A cold worker is polite from scratch, +which errs in the safe direction. + +What it is **not** is the design. One number for every origin is exactly the "conservative +default" the policy stream exists to replace with real per-host values. + +**Known limitation, for the stream to resolve: the host asked for is not always the host +reached.** httpx follows redirects inside the driver, so a URL that 301s elsewhere is paced +under the name the command carried and not at all under the name that served it. A corpus where +several watched URLs funnel into one portal or CDN therefore hits that host at N times the +intended rate — the failure politeness exists to prevent. `FetchResult.final_url` is available +where the fix would go, but recording the landing host too breaks "one request, one record", +so it belongs in the policy stream's design rather than in a quiet amendment to the interim. +Recorded here for the same reason `blob_uri` is: an unwritten gap and a decorative charter are +the same thing to a reader. + +**The stream is a precondition of the Phase 4 cutover, not a follow-on to it.** Watcher's +limiter (`src/core/rate_limiter.py::acquire_for_domain`, fed by 429s its own fetch path +observes) is load-bearing today and stops functioning the moment that fetch path becomes a +publish path — it does not fail, it silently becomes decorative, pacing command publication +rather than origin requests. **The interim default above closes that window**, so the cutover +is no longer blocked on the stream; what remains is that one number for every origin is not +what the cluster wants for long. Tracked issuer-side at +[CannObserv/watcher#245](https://github.com/CannObserv/watcher/issues/245). + +## Reviewing a proposed payload field + +One question: **does this name a domain concept?** `politeness_key: str` passes — opaque to +Replicator. `info_source_id` fails. The wire's domain-agnosticism is the property the whole +issuer contract is built on; it erodes one plausible field at a time. + +**The rule governs payload *shapes*, not producer-owned token vocabularies.** +[`src/core/errors.py::FailureReason`](../../src/core/errors.py) is a locally-defined `StrEnum` +of `fetch_failed` `reason` tokens and stays local by design: co-core types that field as a +plain `str` rather than a `Literal` precisely so a producer adding a token cannot crash an +older `extra="ignore"` consumer, which puts the vocabulary on the producer. Defining a wire +*model* here would be the violation; owning the tokens Replicator itself emits is the +contract working as intended. + +## Known violation, tracked + +`blob_uri` is a host-local `file://` path and nothing on the wire says so. Any consumer must +live on Replicator's VM — a shared-filesystem data-plane coupling in a service otherwise +reached only through the broker, and a constraint on the *issuer's* deployment topology that +the issuer never agreed to. Tracked in #7 (object-store blob backend), and **pinned by a +characterization test** so #7 flips a written line rather than quietly satisfying an unstated +one. + +Recorded here rather than omitted: a charter that asserts an isolation the code does not have +teaches its readers that the document is decorative. + +## Cluster-side corollary + +The single-fetcher invariant is only true if issuers hold up their end. Watcher's create-time +`probe_url` currently fetches origins outside Replicator's politeness envelope; Phase 4 +resolves it by making create asynchronous (item enters a probing state, a normal +`content.fetch` is issued, `final_url` on the fact fills in the resolved URL). After that, +**no service but Replicator fetches watched content.** Stated here because the invariant is +the cluster's, not this repo's alone — and worth an issuer-side test, in the issuer's repo. + +--- + +## Enforcement + +[`tests/test_boundaries.py`](../../tests/test_boundaries.py), run by +[`.github/workflows/ci.yml`](../../.github/workflows/ci.yml) on every PR to `main`. The point +is failing a PR, not documenting an intention. + +| Invariant | Test | +|---|---| +| No database | no persistence distribution in `uv.lock` (sqlalchemy, asyncpg, psycopg, alembic, …); no `sqlite3` / `shelve` / `dbm` / `pickle` import in `src/` | +| No domain vocabulary | AST scan of `src/`: `info_source`, `info_item`, `watched_item`, `tenant`, `aspect` appear in no identifier and no string literal | +| Ingress is read-only | recursive route walk: every path in the allowlist, every method in `{GET, HEAD}` | +| The deployed process has no ingress | `src/worker/` imports no server framework; the unit runs `src.worker.main` with no `uvicorn` and no `--port` | +| No locally-defined wire models | no class in `src/` declares an `event_type` field — every wire payload comes from co-core | +| No issuer SDK | no dependency on a sibling repo's client in the lock | +| Config surface | every `Settings` field is `REPLICATOR_*`-prefixed except `build_id`, exempted by name; no `env_file`; no configuration or network call at import time | +| Known violation, pinned | `blob_uri` still starts `file://` (#7) | + +Three notes on the implementation, because each encodes a decision that a "simplification" +would undo: + +**The vocabulary scan is the load-bearing one, and it is AST-based for a reason.** It reads +identifiers and string literals only, skipping comments and docstrings. The grep this replaced +matched `both tasks watch one stop event` in a docstring — and a test whose first tripper is an +English sentence is a test that gets deleted rather than heeded. The bare verb `watch` is +deliberately absent from the token list; `watched_item`, the domain noun, is not. String +literals are in scope alongside identifiers because domain leakage arrives as a dict key or a +log field (`detail={"info_source_id": ...}`) at least as often as it arrives as an attribute. + +**The `event_type` check is an AST check on class bodies, not a grep.** `event_type` appears +twice in `src/worker/loop.py` legitimately — once in a comment, once reading a co-core model's +own field. A grep would cry wolf on both. + +**The detectors are themselves tested.** Each scan has cases running it against synthetic +violating source, and each corpus scan asserts its own file list is non-empty. A structural +test that quietly walks zero files passes forever while enforcing nothing — which is worse +than no test, because this document then cites it. + +## Refs + +- [`content-fetch-issuer-contract.md`](content-fetch-issuer-contract.md) — the wire contract this sits beside +- #7 — object-store blob backend (the tracked violation) +- #9, #10, #11 — the Phase 4 contract additions +- #12 — this charter +- [CannObserv/watcher#241](https://github.com/CannObserv/watcher/issues/241) — Phase 4 issuer +- [CannObserv/watcher#245](https://github.com/CannObserv/watcher/issues/245) — the politeness gap at cutover +- [CannObserv/archiver#72](https://github.com/CannObserv/archiver/issues/72) — cluster integration strategy diff --git a/src/core/config.py b/src/core/config.py index c33c17e..599422b 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -111,6 +111,30 @@ class Settings(BaseSettings): # systemd's TimeoutStopSec must exceed this plus the handler's budget. read_block_ms: int = Field(default=5_000, validation_alias="REPLICATOR_READ_BLOCK_MS") + # Minimum spacing between two requests to the same host — the interim + # politeness default (#12), replacing the limiter Watcher stops exercising + # the moment its fetch path becomes a publish path (watcher#245). + # + # 1.0 s is Watcher's own DEFAULT_MIN_INTERVAL, chosen precisely because it + # invents nothing: the *numbers* are the issuer's under the boundaries + # charter, and until the policy stream carries them the least-wrong value is + # the one the cluster already commits to. Enforcement is mechanism and + # belongs here; this default is a stand-in for a decision, not the decision. + # + # 0 disables pacing: an operator escape hatch, and the value the pacer's own + # unit tests pin. A deployment setting it is choosing no politeness at all. + # + # Capped at an hour (CR #8). Past that the mechanism is the wrong one rather + # than a stricter setting of the right one: the command parks and re-parks + # for an hour of reclaim cycles, never dead-letters (transient failures are + # exempt from the delivery ceiling), and the issuer's own reaper — the + # backstop the contract requires precisely because silence carries no + # cause — will have concluded loss long before. A fat-fingered extra zero + # should fail at startup, not become a black hole that reads as healthy. + min_host_interval_seconds: float = Field( + default=1.0, ge=0, le=3600, validation_alias="REPLICATOR_MIN_HOST_INTERVAL_SECONDS" + ) + # start_id applies only at group *creation* — once replicator.fetch exists # this value is inert, and switching to "0" (drain the backlog) additionally # needs a manual XGROUP SETID. Kept configurable so the eventual change is a diff --git a/src/worker/handler.py b/src/worker/handler.py index 69e525c..d95bb15 100644 --- a/src/worker/handler.py +++ b/src/worker/handler.py @@ -6,6 +6,7 @@ message pending for the next reclaim. """ +import asyncio import math import re from datetime import UTC, datetime @@ -26,7 +27,8 @@ from src.core.logging import get_logger from src.storage.base import BlobStore from src.storage.sweeper import BlobUsage -from src.worker.loop import Handler +from src.worker.loop import Handler, park +from src.worker.pacing import HostPacer logger = get_logger(__name__) @@ -170,6 +172,9 @@ def build_handler( client: Redis, settings: Settings, usage: BlobUsage | None = None, + pacer: HostPacer | None = None, + park_above_seconds: float | None = None, + stop: asyncio.Event | None = None, blobs_topic: str = streams.CONTENT_BLOBS, ) -> Handler: """Wire the byte path into a handler the loop can dispatch to. @@ -180,6 +185,20 @@ def build_handler( private to this handler, which only makes the ceiling later to notice; the worker passes the shared instance. + ``pacer`` is per-host politeness (#12). Built from ``settings`` when not + injected, deliberately: unwired it fails *open*, and a byte path that + silently stopped pacing looks exactly like one that is working. Tests inject + one with a controlled interval and clock. + + ``park_above_seconds`` is where a pacing wait stops being slept through and + starts parking the message. Defaults to the poll window — a wait no longer + than one blocking read adds nothing to the shutdown latency the unit's + ``TimeoutStopSec`` is already sized for, and a longer one would hold the + serial consume path against every other host's commands. + + ``stop`` lets a sleeping handler notice a SIGTERM. Unset, the sleep simply + runs its course; the bound above is what keeps that from mattering. + ``blobs_topic`` is a defaulted argument rather than a setting, for the same reason ``build_consumer``'s ``topic`` is: the only caller that moves it is a live-broker test, which must keep its facts on a scratch stream. A fact @@ -188,6 +207,10 @@ def build_handler( """ publisher = AsyncBusPublisher(client) usage = usage if usage is not None else BlobUsage() + pacer = pacer if pacer is not None else HostPacer(settings.min_host_interval_seconds) + if park_above_seconds is None: + park_above_seconds = settings.read_block_ms / 1000 + stop = stop if stop is not None else asyncio.Event() async def handle(command: ContentFetchCommand) -> None: # Ahead of the ceiling deliberately. Validation is pure and free; the @@ -196,6 +219,14 @@ async def handle(command: ContentFetchCommand) -> None: # sweep interval to reach a conclusion available immediately. options = _request_options(command, settings.max_fetch_timeout_seconds) _raise_for_ceiling(usage, settings.blob_max_total_bytes) + # Last of the three, and after the ceiling on purpose: spending a wait to + # reach a check that was going to park the message anyway is a wait the + # origin never benefits from. The cost of that ordering is that the tree + # can cross the ceiling *during* a wait — bounded by park_above_seconds, + # and the ceiling is an inter-sweep estimate either way (CR #9). + paced_seconds = await _pace( + pacer, command, stop=stop, park_above_seconds=park_above_seconds + ) result = await _fetch(fetcher, command, options) # Stamped here rather than at publish: occurred_at is when the fact went # onto the bus, which under a reclaim is minutes after the bytes were on @@ -272,6 +303,15 @@ async def handle(command: ContentFetchCommand) -> None: # read than can write to the bus. "request_headers": sorted(options.headers or {}), "request_timeout_seconds": options.timeout, + # Politeness, on the line that already exists rather than one of + # its own (CR #3): without it a mechanism that caps per-host + # throughput is absent from the journal, and an operator seeing a + # slow drain cannot tell "waiting politely" from "origin is + # slow". A per-fetch datum, correlated with duration_ms above — + # the *gauge* (how much of the corpus is under pacing) rides + # _pace's own line instead, so a slowly-changing number is not + # repeated once per command (CR #13). + "paced_seconds": paced_seconds, }, ) @@ -465,6 +505,74 @@ async def _fetch( raise TransientFetchError(f"{command.url} failed to fetch: {exc}") from exc +async def _pace( + pacer: HostPacer, + command: ContentFetchCommand, + *, + stop: asyncio.Event, + park_above_seconds: float, +) -> float: + """Give the origin its space before asking it for anything (#12). + + Returns the seconds actually waited, for the success line to report. + + Two ways to spend a wait, split by duration, because neither is correct + alone on a serial consume path: + + * **Sleep** a short one. The consume path is serial, so a sub-second pause + costs the group a sub-second pause — the same order as the fetch it is + about to do, and far cheaper than a reclaim round-trip. + * **Park** a long one, transiently, so the message returns via + ``claim_stale`` like a command over the disk ceiling. Sleeping instead + would hold every *other* host's commands behind this one origin's + politeness, and hold a SIGTERM behind it too. + + Parking cannot express a wait shorter than ``REPLICATOR_CLAIM_MIN_IDLE_MS`` + (60 s by default), which is the whole reason the sleep branch exists: the + normal interval is a second, and a park-only implementation would pace every + host at 1/60th of the rate the cluster runs at today. Silently, and in the + safe direction, which is what would have made it easy to ship. + + Transient in both directions — a paced command has done nothing wrong, and + burning its delivery ceiling on politeness would dead-letter perfectly good + work. + """ + wait = pacer.wait_seconds(command.url) + if wait <= 0: + pacer.record(command.url) + return 0.0 + if wait > park_above_seconds: + raise TransientFetchError( + f"{command.url} is inside its host's {wait:.1f}-second politeness window; " + f"leaving it for the next reclaim" + ) + # INFO, and only on the branch that actually waits (CR #13). At DEBUG this + # was invisible under the root INFO level; on every command it repeated a + # gauge that changes only when the corpus does. Here it appears exactly when + # the mechanism acts, which is when an operator wants it, and carries + # `tracked_hosts` as the periodic-ish gauge that has nowhere better to live — + # the sweep's line is the other candidate, and it is silent on an idle tree. + logger.info( + "waiting out a host's politeness window", + extra={ + "command_id": command.command_id, + "wait_seconds": wait, + "tracked_hosts": pacer.tracked_hosts, + }, + ) + await park(stop, wait) + # park returns early on SIGTERM, and an interrupted wait is not an elapsed + # one — the origin has had no space. The message stays in the PEL, which is + # where a command interrupted mid-flight belongs anyway. + if stop.is_set(): + raise TransientFetchError( + f"{command.url} was still inside its host's politeness window when the worker " + f"began stopping" + ) + pacer.record(command.url) + return wait + + def _raise_for_ceiling(usage: BlobUsage, ceiling_bytes: int) -> None: """Stop fetching once the blob tree has grown past what this deployment holds. diff --git a/src/worker/main.py b/src/worker/main.py index 81436f5..ebaeecb 100644 --- a/src/worker/main.py +++ b/src/worker/main.py @@ -316,6 +316,10 @@ async def run(stop: asyncio.Event | None = None) -> None: client=client, settings=settings, usage=usage, + # The same stop event the loop and the sweeper ride, so a + # handler waiting out a politeness window does not hold a + # SIGTERM for it (#12). + stop=stop, ), # The other outcome of a command, on the same stream: an issuer # closes a pending entry off one consumer group either way (#9, diff --git a/src/worker/pacing.py b/src/worker/pacing.py new file mode 100644 index 0000000..396af3f --- /dev/null +++ b/src/worker/pacing.py @@ -0,0 +1,145 @@ +"""Per-host request spacing — the mechanism half of politeness (#12). + +`docs/contracts/replicator-boundaries.md` splits this capability: Replicator +**enforces** a per-host rate because it is the only process that can see the +origin's tolerance across commands, and the issuer **decides** the numbers, +which are operator policy. Until the `content.fetch.policy` stream exists there +is nothing to decide with, so this carries a single conservative default from +env — a default is mechanism, and the alternative at the Phase 4 cutover is no +politeness at all: Watcher's limiter paces its own fetches, and the moment that +fetch path becomes a publish path it silently starts pacing nothing +(CannObserv/watcher#245). + +The state is a host -> last-request timestamp map, in memory. That is one of the +three shapes the charter permits: derived, bounded, and rebuildable by replay — +a cold worker is simply polite from scratch, which errs in the safe direction. +""" + +import time +from collections.abc import Callable +from urllib.parse import urlsplit + +# When to prune. Each entry is a hostname and a float; the bound is not about +# memory pressure at this size but about the shape — an unbounded map that only +# ever grows is a leak whatever its constant. Pruning is O(n) and runs only on +# the record that crosses the bound, so the amortized cost is negligible. +MAX_TRACKED_HOSTS = 4096 + + +class HostPacer: + """How long before this host may be asked for something again. + + Deliberately *not* a token bucket: a bucket permits a burst up to its depth, + and a burst is precisely what an origin notices. Fixed minimum spacing is the + same rule Watcher enforces today (``DEFAULT_MIN_INTERVAL``), so the cutover + changes who paces rather than how. + + Reports a wait; it does not spend one. The caller decides whether a wait is + short enough to sleep through or long enough to park the message, because + only the caller knows what else is behind it on a serial consume path. + """ + + def __init__( + self, min_interval_seconds: float, *, clock: Callable[[], float] = time.monotonic + ) -> None: + self._interval = min_interval_seconds + self._clock = clock + self._last: dict[str, float] = {} + # Earliest a prune could reclaim anything; see _prune. + self._prune_not_before = 0.0 + + @property + def tracked_hosts(self) -> int: + """How many hosts currently hold an entry. + + Reported on ``handler.py::_pace``'s log line — the branch that actually + waits — so a rising figure is visible as the corpus widening rather than + inferred from throughput, without repeating a slow gauge once per + command. + """ + return len(self._last) + + def wait_seconds(self, url: str) -> float: + """Seconds before a request to ``url``'s host may go out. ``0.0`` for now. + + A URL with no host is never paced: it is also unfetchable, and the driver + refuses it as ``not_fetchable``. Returning a wait would park a + permanently-bad command in the PEL instead of letting it reach the + terminal fact its issuer is waiting for. + """ + host = _host(url) + if host is None or self._interval <= 0: + return 0.0 + last = self._last.get(host) + if last is None: + return 0.0 + return max(0.0, self._interval - (self._clock() - last)) + + def record(self, url: str) -> None: + """Stamp a request as having gone out. + + Called after the wait, never before it — only a request that reaches the + origin resets that origin's clock. Recording an *attempt* would let a run + of parked redeliveries push the next real fetch out indefinitely, spacing + the origin from requests it never received. + """ + host = _host(url) + if host is None: + return + now = self._clock() + self._last[host] = now + if len(self._last) > MAX_TRACKED_HOSTS and now >= self._prune_not_before: + self._prune(now) + + def _prune(self, now: float) -> None: + """Drop hosts whose interval has already elapsed. + + Those entries impose no wait, so removing them changes no decision this + pacer can make. Hosts still inside their interval are kept however far + over the bound that leaves us — enforcing the bound by forgetting a host + that is still owed space would turn a memory limit into a politeness + breach. + + Which means the bound can be exceeded with nothing to reclaim, and a + prune that frees nothing must not run again on the next record: with more + than ``MAX_TRACKED_HOSTS`` hosts all inside a long interval, that would + be a full dict rebuild per message forever (CR #7). The retry is deferred + until the oldest entry could plausibly have aged out. + """ + before = len(self._last) + self._last = { + host: last for host, last in self._last.items() if now - last < self._interval + } + if len(self._last) < before: + self._prune_not_before = 0.0 + return + # Nothing was reclaimable. The earliest anything can be is one interval + # after the oldest entry still held. + self._prune_not_before = min(self._last.values(), default=now) + self._interval + + +def _host(url: str) -> str | None: + """The URL's hostname, or ``None`` when there is nothing to pace. + + ``urlsplit().hostname`` is already lowercased and already excludes the port, + which is the key this wants: politeness is about how often a *server* is + asked, so a second port is the same machine and a capitalized host is the + same name. Keying on anything finer would let an issuer multiply its own + rate limit by spelling the URL differently. + + **Known limitation: the host asked for, not the host reached (CR #4).** httpx + follows redirects inside the driver, so a URL that 301s elsewhere is paced + under the name the command carried and not at all under the name that + actually served it. A corpus where several watched URLs funnel into one + portal or CDN therefore hits that host at N times the intended rate — the + failure politeness exists to prevent. Recording the landing host too + (``FetchResult.final_url`` is available at the call site) would fix it at the + cost of "one request, one record", which wants its own decision rather than + a quiet change here; the policy stream is where it should land. + """ + try: + return urlsplit(url).hostname + except ValueError: + # An unparseable authority (a bad IPv6 literal, say). Unfetchable for the + # same reason a hostless URL is, and handled the same way. + return None diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 3be17d8..2404c88 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -2,7 +2,10 @@ from pathlib import Path -from src.core.config import get_settings +import pytest +from pydantic import ValidationError + +from src.core.config import Settings, get_settings def test_defaults_match_the_shared_vm(monkeypatch): @@ -87,3 +90,39 @@ def test_retention_env_overrides(monkeypatch): assert settings.blob_ttl_seconds == 60 assert settings.blob_sweep_interval_seconds == 5 assert settings.blob_max_total_bytes == 1024 + + +def test_the_pacing_default_matches_what_watcher_already_commits_to(monkeypatch): + """1.0s is Watcher's own DEFAULT_MIN_INTERVAL, and that is the whole argument. + + The politeness *numbers* belong to the issuer under the boundaries charter, + so until the policy stream carries them the interim must not invent one. A + change here is a change to what the cluster promises origins (#12). + """ + monkeypatch.delenv("REPLICATOR_MIN_HOST_INTERVAL_SECONDS", raising=False) + + assert get_settings().min_host_interval_seconds == 1.0 + + +def test_pacing_can_be_disabled_and_configured(monkeypatch): + monkeypatch.setenv("REPLICATOR_MIN_HOST_INTERVAL_SECONDS", "0") + assert get_settings().min_host_interval_seconds == 0 + + get_settings.cache_clear() + monkeypatch.setenv("REPLICATOR_MIN_HOST_INTERVAL_SECONDS", "2.5") + assert get_settings().min_host_interval_seconds == 2.5 + + +@pytest.mark.parametrize("value", ["-1", "7200"], ids=["negative", "over-the-cap"]) +def test_an_out_of_range_pacing_interval_fails_at_startup(monkeypatch, value): + """The cap has to bite where the comment says it does (CR #14). + + Past an hour the command parks and re-parks without ever dead-lettering — + transient failures are exempt from the delivery ceiling — while the issuer's + reaper concludes loss. A fat-fingered extra zero must fail loudly at + construction rather than become a black hole that reads as healthy. + """ + monkeypatch.setenv("REPLICATOR_MIN_HOST_INTERVAL_SECONDS", value) + + with pytest.raises(ValidationError): + Settings() diff --git a/tests/test_boundaries.py b/tests/test_boundaries.py new file mode 100644 index 0000000..cd7e316 --- /dev/null +++ b/tests/test_boundaries.py @@ -0,0 +1,587 @@ +"""Executable half of the boundaries charter. + +Reasoning lives in +[`docs/contracts/replicator-boundaries.md`](../docs/contracts/replicator-boundaries.md) — +read it before deleting an assertion here. The charter's rule is that Replicator +owns the mechanics of acquiring bytes and holding them briefly, and never owns +why, when, or what they mean. Every assertion below is one way that rule stops +being true one defensible commit at a time. + +Two conventions worth knowing before editing: + +**The detectors are themselves tested.** ``test_the_*_detector_*`` cases run each +scan against synthetic violating source. A structural test that quietly walks +zero files, or a substring check that stops matching after a refactor, passes +forever while enforcing nothing — which is worse than no test, because the +charter then cites it. The corpus scans also assert their own file list is +non-empty for the same reason. + +**Scans are AST-based, never grep.** The vocabulary scan in particular reads +identifiers and string literals only, skipping comments and docstrings: prose +uses these words legitimately (``both tasks watch one stop event``), and a test +whose first tripper is an English sentence is a test that gets deleted rather +than heeded. The regression it exists to catch always arrives as a field name, a +parameter, or a dict key. +""" + +import ast +import tomllib +from pathlib import Path + +import pytest +from starlette.routing import Route, WebSocketRoute + +from src.api.main import app +from src.core.config import Settings +from src.storage.local import LocalBlobStore + +REPO = Path(__file__).resolve().parents[1] +SRC = REPO / "src" +WORKER = SRC / "worker" +UNIT = REPO / "deploy" / "replicator.service" +LOCK = REPO / "uv.lock" + + +def _python_files(root: Path) -> list[Path]: + """Every module under ``root``, sorted for a stable failure message.""" + return sorted(root.rglob("*.py")) + + +def _parse(path: Path) -> ast.Module: + return ast.parse(path.read_text(), filename=str(path)) + + +# -------------------------------------------------------------------------- +# 1. No database +# -------------------------------------------------------------------------- + +# Charter test 1: Replicator's state is content-addressed on disk, in memory, or +# in the broker. Anything else is a database whatever it is called. Third-party +# names are checked against the lock rather than the installed environment — +# a dirty local venv must not be able to make this pass. +PERSISTENCE_DISTRIBUTIONS = frozenset( + { + "alembic", + "aiosqlite", + "asyncpg", + "motor", + "peewee", + "psycopg", + "psycopg2", + "psycopg2-binary", + "psycopg-binary", + "pymongo", + "sqlalchemy", + "sqlmodel", + "tortoise-orm", + } +) + +# The stdlib half. sqlite3 is the obvious one; shelve and dbm are the same thing +# with less ceremony, and pickle is how a cache becomes a file nobody calls a +# database. None has a legitimate use in a service whose entire durable state is +# rebuildable — if one acquires it, this list is the place to argue about it. +PERSISTENCE_MODULES = frozenset({"sqlite3", "shelve", "dbm", "pickle"}) + + +def _locked_distributions() -> set[str]: + """Every distribution name in ``uv.lock``, normalized.""" + lock = tomllib.loads(LOCK.read_text()) + return {package["name"].lower().replace("_", "-") for package in lock["package"]} + + +def _imported_modules(tree: ast.Module) -> set[str]: + """Top-level module names imported anywhere in ``tree``.""" + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module and not node.level: + modules.add(node.module.split(".")[0]) + return modules + + +def test_no_persistence_dependency_resolves_in_the_lock(): + assert not PERSISTENCE_DISTRIBUTIONS & _locked_distributions() + + +def test_no_module_imports_stdlib_persistence(): + files = _python_files(SRC) + assert files, f"no modules found under {SRC} — the scan is a no-op" + + offenders = { + path.relative_to(REPO).as_posix(): sorted( + PERSISTENCE_MODULES & _imported_modules(_parse(path)) + ) + for path in files + if PERSISTENCE_MODULES & _imported_modules(_parse(path)) + } + assert not offenders + + +def test_the_import_detector_sees_a_planted_import(): + tree = ast.parse("import sqlite3\nfrom dbm import gnu\n") + + assert PERSISTENCE_MODULES & _imported_modules(tree) == {"sqlite3", "dbm"} + + +# -------------------------------------------------------------------------- +# 2. No domain vocabulary +# -------------------------------------------------------------------------- + +# Charter test 3: anything needing these words belongs to the issuer. The bare +# verb "watch" is deliberately absent — it is ordinary English, and scoping this +# scan to identifiers is what lets the domain noun stay. +DOMAIN_TOKENS = frozenset({"info_source", "info_item", "watched_item", "tenant", "aspect"}) + + +def _docstring_nodes(tree: ast.Module) -> set[int]: + """Ids of the ``Constant`` nodes that are docstrings, so the scan can skip them.""" + skip: set[int] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.Module | ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef): + continue + first = node.body[0] if node.body else None + if ( + isinstance(first, ast.Expr) + and isinstance(first.value, ast.Constant) + and isinstance(first.value.value, str) + ): + skip.add(id(first.value)) + return skip + + +def _vocabulary_surface(tree: ast.Module) -> set[str]: + """Every identifier and non-docstring string literal in ``tree``. + + String literals are in scope alongside identifiers because domain leakage + arrives as a dict key or a log field (``detail={"info_source_id": ...}``) at + least as often as it arrives as an attribute — and that form is the one a + reviewer skims past. + """ + skip = _docstring_nodes(tree) + surface: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Name): + surface.add(node.id) + elif isinstance(node, ast.Attribute): + surface.add(node.attr) + elif isinstance(node, ast.arg): + surface.add(node.arg) + elif isinstance(node, ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef): + surface.add(node.name) + elif isinstance(node, ast.keyword) and node.arg: + surface.add(node.arg) + elif ( + isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in skip + ): + surface.add(node.value) + return surface + + +def _domain_hits(tree: ast.Module) -> set[str]: + """Tokens from ``DOMAIN_TOKENS`` appearing anywhere in the vocabulary surface.""" + lowered = [text.lower() for text in _vocabulary_surface(tree)] + return {token for token in DOMAIN_TOKENS if any(token in text for text in lowered)} + + +def test_no_module_names_a_domain_concept(): + """The load-bearing one. + + A domain field on the wire is the regression no type checker and no reviewer + reliably catches, because it always arrives looking reasonable: one optional + field, one small settings table, and Replicator has a domain model. + """ + files = _python_files(SRC) + assert files, f"no modules found under {SRC} — the scan is a no-op" + + offenders = { + path.relative_to(REPO).as_posix(): sorted(hits) + for path in files + if (hits := _domain_hits(_parse(path))) + } + assert not offenders + + +@pytest.mark.parametrize( + "source", + [ + pytest.param("def f(info_source_id): ...", id="parameter"), + pytest.param("command.watched_item_id", id="attribute"), + pytest.param('detail = {"info_item": 1}', id="dict-key"), + pytest.param("class TenantScope: ...", id="class-name"), + ], +) +def test_the_vocabulary_detector_sees_a_planted_domain_name(source): + assert _domain_hits(ast.parse(source)) + + +@pytest.mark.parametrize( + "source", + [ + pytest.param('"""Both tasks watch one stop event."""', id="module-docstring"), + pytest.param( + "def f():\n '''Watch the tenant aspect of nothing.'''\n return 1", + id="function-docstring", + ), + pytest.param("# info_source_id belongs to the issuer\nx = 1", id="comment"), + ], +) +def test_the_vocabulary_detector_ignores_prose(source): + """Prose is where these words are legitimate, and where a false positive kills the test.""" + assert not _domain_hits(ast.parse(source)) + + +# -------------------------------------------------------------------------- +# 3. Ingress is read-only +# -------------------------------------------------------------------------- + +# The charter rejects an inbound admin HTTP API by name. What that forbids is a +# route accepting state-changing input; FastAPI's three introspection endpoints +# are read-only and describe the one real route, so they are allowlisted rather +# than switched off — a dev/prod docs flag would make the app under test differ +# from the app that ships, for a surface no deployment serves. +ALLOWED_PATHS = frozenset({"/health", "/docs", "/docs/oauth2-redirect", "/redoc", "/openapi.json"}) +READ_ONLY_METHODS = frozenset({"GET", "HEAD"}) + + +def _routes(node, prefix: str = "") -> set[tuple[str, str]]: + """Every (path, method) pair reachable from ``node``, flattened and prefixed. + + Recursive because ``/health`` is registered through ``include_router`` and so + appears in ``app.routes`` as a router rather than a ``Route`` — under FastAPI + 0.141 an ``_IncludedRouter``, which carries its children on + ``original_router`` and its mount prefix on ``include_context``. A flat + comparison against ``app.routes`` fails *and* never sees ``/health``, which + is the shape of test that looks strict and enforces nothing. + + Both container shapes are handled (``routes`` for an app or ``Mount``, + ``original_router`` for an included one) so this survives FastAPI moving + between them. ``test_the_route_walk_sees_a_planted_write_route`` is what + reports it if the walk ever stops descending. + + **Two ways this used to fail open, both fixed in CR #1 and both now planted + as tests.** A ``Mount`` carries its prefix on ``.path`` rather than on an + ``include_context``, so a sub-app mounted at ``/admin`` reported its + children's bare paths — ``GET /admin/health`` passed the allowlist as + ``/health``. And a ``WebSocketRoute`` is not a ``Route`` and has no + ``methods``, so it fell through to the container branch, found no children, + and contributed nothing at all. Mounting a sub-app is the most plausible + shape an admin API would actually arrive in, which is exactly why the walk + has to see it. + """ + if isinstance(node, Route): + return {(prefix + node.path, method) for method in node.methods or {"GET"}} + if isinstance(node, WebSocketRoute): + # No methods of its own, and a socket is a write surface by construction: + # named so it fails the read-only assertion rather than vanishing. + return {(prefix + node.path, "WEBSOCKET")} + + context = getattr(node, "include_context", None) + prefix += getattr(context, "prefix", "") or "" + # A Mount's own prefix. The app root and an _IncludedRouter have no `path`, + # so this is additive for them; for a Mount it is the whole point. + prefix += getattr(node, "path", "") or "" + children = getattr(node, "routes", None) + if children is None: + children = getattr(getattr(node, "original_router", None), "routes", ()) + return {pair for child in children for pair in _routes(child, prefix)} + + +def test_ingress_is_read_only(): + """No route accepts state-changing input. + + The charter rejects an inbound admin HTTP API; what that forbids is a write + surface, not self-description. Both halves are asserted because either alone + is weak: an allowlist of paths would admit a ``POST /health``, and a + method check alone would admit a read-only config dump. + """ + surface = _routes(app) + + assert ("/health", "GET") in surface, "the route walk stopped seeing /health" + assert {path for path, _ in surface} <= ALLOWED_PATHS + assert {method for _, method in surface} <= READ_ONLY_METHODS + + +def test_the_route_walk_sees_a_planted_write_route(): + from fastapi import APIRouter, FastAPI + + probe = FastAPI() + router = APIRouter() + router.post("/policy")(lambda: None) + probe.include_router(router, prefix="/admin") + + assert ("/admin/policy", "POST") in _routes(probe) + + +def test_the_route_walk_sees_through_a_mounted_sub_app(): + """The shape an admin API would actually arrive in (CR #1). + + A sub-app is the natural way to add one — it carries its own router, its own + docs, its own everything — and before this the walk reported its children + without the mount prefix. A ``GET /admin/health`` returning the worker's + configuration passed both assertions as ``/health``. + """ + from fastapi import FastAPI + + sub = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) + sub.get("/health")(lambda: {"config": "everything"}) + probe = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) + probe.mount("/admin", sub) + + surface = _routes(probe) + + assert ("/admin/health", "GET") in surface + assert not {path for path, _ in surface} <= ALLOWED_PATHS + + +def test_the_route_walk_sees_a_websocket(): + """Not a ``Route``, no ``methods`` — it used to contribute nothing at all. + + A socket is a write surface by construction, so it is named with a method of + its own rather than left to fail the path allowlist alone: a websocket at + ``/health`` would otherwise be invisible on both assertions. + """ + from fastapi import FastAPI + + probe = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) + + @probe.websocket("/health") + async def _socket(websocket): ... + + surface = _routes(probe) + + assert ("/health", "WEBSOCKET") in surface + assert not {method for _, method in surface} <= READ_ONLY_METHODS + + +# -------------------------------------------------------------------------- +# 4. No locally-defined wire models +# -------------------------------------------------------------------------- + + +def _classes_declaring(tree: ast.Module, field: str) -> set[str]: + """Classes in ``tree`` with an annotated or assigned attribute named ``field``. + + An AST check on class bodies, deliberately not a grep: ``event_type`` appears + twice in ``src/worker/loop.py`` legitimately — once in a comment and once + reading a co-core model's own field — and a grep that cries wolf on those + gets replaced by nothing. + """ + declared: set[str] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + for statement in node.body: + targets: list[ast.expr] = [] + if isinstance(statement, ast.AnnAssign): + targets = [statement.target] + elif isinstance(statement, ast.Assign): + targets = list(statement.targets) + if any(isinstance(t, ast.Name) and t.id == field for t in targets): + declared.add(node.name) + return declared + + +def test_no_wire_payload_is_defined_here(): + """Every payload shape comes from co-core. + + The rule governs payload **shapes**, not producer-owned token vocabularies: + ``src/core/errors.py::FailureReason`` is a local ``StrEnum`` of ``reason`` + tokens and stays local by design, because co-core types that field as a + plain ``str`` rather than a ``Literal`` so a producer adding a token cannot + crash an older ``extra="ignore"`` consumer. + """ + files = _python_files(SRC) + assert files, f"no modules found under {SRC} — the scan is a no-op" + + offenders = { + path.relative_to(REPO).as_posix(): sorted(declared) + for path in files + if (declared := _classes_declaring(_parse(path), "event_type")) + } + assert not offenders + + +def test_the_wire_model_detector_sees_a_planted_field(): + tree = ast.parse("class Thing(BaseModel):\n event_type: str = 'thing'\n") + + assert _classes_declaring(tree, "event_type") == {"Thing"} + + +def test_the_wire_model_detector_ignores_reading_the_field(): + tree = ast.parse( + "class Handler:\n def go(self, command):\n return command.event_type\n" + ) + + assert not _classes_declaring(tree, "event_type") + + +# -------------------------------------------------------------------------- +# 5. No issuer SDK +# -------------------------------------------------------------------------- + +# A client library for a sibling repo is how domain vocabulary arrives wholesale +# rather than one field at a time. co-core is the shared contract layer and is +# not one of these: it is owned by no issuer. +ISSUER_DISTRIBUTIONS = frozenset( + {"archiver", "archiver-client", "notifier", "notifier-client", "watcher", "watcher-client"} +) + + +def test_no_sibling_repo_client_resolves_in_the_lock(): + assert not ISSUER_DISTRIBUTIONS & _locked_distributions() + + +# -------------------------------------------------------------------------- +# 6. Config surface +# -------------------------------------------------------------------------- + +# Named exemption, not an oversight: the systemd unit's ExecStartPre stamps +# BUILD_ID generically across the cluster's services, so prefixing it here would +# mean every unit needs a per-service variable name for one git SHA. Documented +# in AGENTS.md and in the charter's config taxonomy. +UNPREFIXED_SETTINGS = frozenset({"build_id"}) + + +def test_every_setting_is_replicator_prefixed(): + offenders = { + name: field.validation_alias + for name, field in Settings.model_fields.items() + if name not in UNPREFIXED_SETTINGS + and not str(field.validation_alias).startswith("REPLICATOR_") + } + assert not offenders + + +def test_the_exemption_stays_a_short_list(): + """An exemption list that grows is the prefix convention ending quietly.""" + assert UNPREFIXED_SETTINGS == {"build_id"} + assert Settings.model_fields["build_id"].validation_alias == "BUILD_ID" + + +def test_settings_reads_no_file_of_its_own(): + """Env files are loaded by systemd or the developer, never by this process. + + ``env_file`` would make the service's configuration depend on the working + directory it was launched from — and on the repo ``.env``, which holds + org-wide PATs the worker has no use for. + """ + assert Settings.model_config.get("env_file") is None + + +# Import must not read configuration and must not touch the network. Deliberately +# a denylist rather than an allowlist of permitted calls: building a logger, a +# router, or a Field default at import is ordinary and an allowlist would have to +# grow for each, which is how a structural test becomes a rubber stamp. The one +# import-time file read in the repo — `pkg_version("replicator")` in src/api — +# is metadata, not configuration, and lives in the surface no deployment serves. +IMPORT_TIME_FORBIDDEN = frozenset( + { + "Settings", + "get_settings", + "getenv", + "open", + "read_text", + "read_bytes", + "gethostname", + "connect", + "Redis", + "from_url", + } +) + + +def _import_time_calls(tree: ast.Module) -> set[str]: + """Names called while the module is being imported. + + Module scope and class bodies both execute on import; function bodies do + not, so the walk stops at them. Scanning them anyway is what made an earlier + version of this test report ``Settings`` for ``get_settings``'s own body. + """ + called: set[str] = set() + stack: list[ast.AST] = list(tree.body) + while stack: + node = stack.pop() + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda): + continue + if isinstance(node, ast.Call): + func = node.func + called.add(func.id if isinstance(func, ast.Name) else getattr(func, "attr", "")) + stack.extend(ast.iter_child_nodes(node)) + return called + + +def test_no_module_reads_configuration_or_the_network_at_import_time(): + files = _python_files(SRC) + assert files, f"no modules found under {SRC} — the scan is a no-op" + + offenders = { + path.relative_to(REPO).as_posix(): sorted(hits) + for path in files + if (hits := IMPORT_TIME_FORBIDDEN & _import_time_calls(_parse(path))) + } + assert not offenders + + +def test_the_import_time_detector_scans_class_bodies_but_not_functions(): + planted = ast.parse("class C:\n x = open('f')\n") + deferred = ast.parse("def f():\n return open('f')\n") + + assert "open" in _import_time_calls(planted) + assert "open" not in _import_time_calls(deferred) + + +# -------------------------------------------------------------------------- +# 7. The deployed process has no ingress +# -------------------------------------------------------------------------- + +# The assertion above covers src/api, which no deployment serves. This one covers +# the process that does: replicator.service runs the worker, and the worker binds +# no port. Without it the charter's ingress invariant could hold forever while an +# admin listener grows inside src/worker/ — the exact thing the rejected fourth +# channel is about. +SERVER_MODULES = frozenset( + {"fastapi", "uvicorn", "starlette", "aiohttp", "flask", "http", "socketserver", "wsgiref"} +) + + +def test_the_worker_imports_no_server_framework(): + files = _python_files(WORKER) + assert files, f"no modules found under {WORKER} — the scan is a no-op" + + offenders = { + path.relative_to(REPO).as_posix(): sorted(SERVER_MODULES & _imported_modules(_parse(path))) + for path in files + if SERVER_MODULES & _imported_modules(_parse(path)) + } + assert not offenders + + +def test_the_unit_starts_the_worker_and_binds_no_port(): + unit = UNIT.read_text() + + assert "src.worker.main" in unit + assert "uvicorn" not in unit + assert "--port" not in unit + + +# -------------------------------------------------------------------------- +# 8. The tracked violation +# -------------------------------------------------------------------------- + + +def test_a_blob_uri_is_still_host_local(tmp_path): + """Characterization, not endorsement — see the charter's "Known violation". + + ``file://`` means every ``content.blobs`` consumer must share Replicator's + filesystem: a data-plane coupling in a service otherwise reached only through + the broker, and a constraint on the issuer's deployment topology the issuer + never agreed to. Pinned so #7's object-store backend flips a written line + rather than quietly satisfying an unstated one. + """ + store = LocalBlobStore(tmp_path) + + uri = store.store(b"bytes", "0" * 64, "text/plain") + + assert uri.startswith("file://") diff --git a/tests/test_deploy.py b/tests/test_deploy.py index 8d8da33..fed2546 100644 --- a/tests/test_deploy.py +++ b/tests/test_deploy.py @@ -60,3 +60,24 @@ def test_the_stop_timeout_outlasts_the_slowest_fetch_a_command_may_ask_for(): timeout_stop = float(_directive("TimeoutStopSec")) assert timeout_stop > settings.read_block_ms / 1000 + settings.max_fetch_timeout_seconds + + +def test_the_stop_timeout_absorbs_a_pacing_wait_as_well(): + """The #12 term. + + A handler may now sleep out a per-host politeness window before it fetches, + bounded by the poll window (``build_handler``'s ``park_above_seconds`` + default — anything longer parks instead). The stop event cuts that sleep + short, so this is belt-and-braces rather than the primary guard: the sum is + asserted because the alternative is discovering at the next deploy that + three separately-reasonable numbers no longer fit inside one. + """ + settings = Settings() + timeout_stop = float(_directive("TimeoutStopSec")) + worst_case = ( + settings.read_block_ms / 1000 # a poll already in flight + + settings.read_block_ms / 1000 # the pacing sleep bound, derived from it + + settings.max_fetch_timeout_seconds # the slowest fetch a command may ask for + ) + + assert timeout_stop > worst_case diff --git a/tests/worker/conftest.py b/tests/worker/conftest.py index 65d62e5..e65d073 100644 --- a/tests/worker/conftest.py +++ b/tests/worker/conftest.py @@ -93,6 +93,30 @@ def effect(self) -> FetchContent: return effect +class Clock: + """A monotonic clock the test advances by hand. + + Pacing is the one thing in the byte path whose whole behaviour is a duration, + so every assertion about it would otherwise be a real sleep — slow, and flaky + in exactly the direction that hides an off-by-one. Shared by the pacer's unit + tests and the handler's, so both express a wait the same way. + """ + + def __init__(self) -> None: + self.now = 1_000.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture +def clock() -> Clock: + return Clock() + + def now() -> datetime: """A tz-aware UTC stamp — the only kind co-core's payloads accept. diff --git a/tests/worker/test_handler_pacing.py b/tests/worker/test_handler_pacing.py new file mode 100644 index 0000000..d7b8fd7 --- /dev/null +++ b/tests/worker/test_handler_pacing.py @@ -0,0 +1,238 @@ +"""How the byte path spends a pacing wait (#12). + +Two outcomes, split by duration: a short wait is slept through in the handler, a +long one parks the message in the PEL for ``claim_stale`` to bring back. The +split exists because neither alone is correct on a serial consume path — sleeping +through a long wait blocks every other host's commands and a SIGTERM behind them, +and parking is bounded below by ``REPLICATOR_CLAIM_MIN_IDLE_MS``, so it cannot +express the sub-minute spacing that is the normal case. +""" + +import asyncio + +import pytest +from co_core.pure.adapters.bus import streams + +from src.core.config import get_settings +from src.core.errors import TransientFetchError +from src.storage.local import LocalBlobStore +from src.worker.handler import build_handler +from src.worker.pacing import HostPacer +from tests.worker.conftest import URL, Clock, FakeFetcher, command, published_facts + + +@pytest.fixture +def paced(fake_redis, tmp_path): + """A handler whose pacer and sleep bound the test controls.""" + + def build( + pacer: HostPacer, + park_above_seconds: float = 5.0, + stop: asyncio.Event | None = None, + ): + fetcher = FakeFetcher() + return build_handler( + fetcher=fetcher, + store=LocalBlobStore(tmp_path), + client=fake_redis, + settings=get_settings(), + pacer=pacer, + park_above_seconds=park_above_seconds, + stop=stop, + ), fetcher + + return build + + +async def test_a_short_wait_is_slept_through_and_the_fetch_still_happens(paced, fake_redis): + """The normal case. A sub-second space between two commands is not a failure.""" + handler, fetcher = paced(HostPacer(0.05)) + + await handler(command("cmd-1")) + await handler(command("cmd-2")) + + assert fetcher.urls == [URL, URL] + assert len(await published_facts(fake_redis)) == 2 + + +async def test_two_requests_to_one_host_are_actually_spaced(paced): + """Measured across both fetches, not across the sleep. + + The wait is the *remainder* of the interval — the store and publish between + the two commands have already spent part of it. An assertion on the sleep + alone would fail for the handler doing its job quickly, which is backwards. + """ + handler, _ = paced(HostPacer(0.2)) + loop = asyncio.get_running_loop() + + started = loop.time() + await handler(command("cmd-1")) + await handler(command("cmd-2")) + + assert loop.time() - started >= 0.2 + + +async def test_a_wait_longer_than_the_bound_parks_the_message(paced, fake_redis): + """Transient, so the delivery ceiling is not burned for being polite. + + The command comes back through ``claim_stale`` — the same idiom the disk + ceiling uses, and the reason a paced command can never dead-letter for pacing. + """ + handler, fetcher = paced(HostPacer(60.0), park_above_seconds=5.0) + await handler(command("cmd-1")) + + with pytest.raises(TransientFetchError, match="60.0-second"): + await handler(command("cmd-2")) + + assert fetcher.urls == [URL], "the parked command must not have been fetched" + assert len(await published_facts(fake_redis)) == 1 + + +async def test_a_shutdown_during_the_sleep_parks_rather_than_fetches(paced): + """SIGTERM must not buy an unpaced request on the way out. + + ``park`` returns early when the stop event is set, and the handler cannot + treat that as the wait having elapsed — the origin has not had its space. The + message stays in the PEL, which is where an interrupted command belongs. + """ + stop = asyncio.Event() + handler, fetcher = paced(HostPacer(30.0), park_above_seconds=60.0, stop=stop) + + await handler(command("cmd-1")) + stop.set() + with pytest.raises(TransientFetchError, match="stopping"): + await handler(command("cmd-2")) + + assert fetcher.urls == [URL] + + +async def test_pacing_runs_after_the_option_guards(paced, fake_redis): + """A command that can never succeed must not wait first. + + Same ordering argument as the storage ceiling: validation is pure and free, + and a permanently-bad command deserves its terminal fact now rather than + after a reclaim cycle. + """ + handler, fetcher = paced(HostPacer(60.0), park_above_seconds=0.0) + await handler(command("cmd-1")) + + with pytest.raises(Exception, match="invalid|refused") as caught: + await handler(command("cmd-2", headers={"Host": "elsewhere.test"})) + + assert not isinstance(caught.value, TransientFetchError) + assert fetcher.urls == [URL] + + +async def test_a_parked_command_does_not_consume_its_own_space(paced): + """Only a request that goes out resets the clock. + + Recording the attempt would let a burst of parked redeliveries push the next + real fetch out indefinitely — the origin would be spaced from requests it + never received. + + Driven by a hand-advanced clock (CR #6). The earlier form compared two live + ``wait_seconds`` readings, which did kill the mutant but only via an implicit + argument about which of two elapsed intervals was larger — sound, and + impossible to see. Here the remaining wait is exact: 60 s minus the 10 s + advanced, and unchanged by the park in between. + """ + clock = Clock() + pacer = HostPacer(60.0, clock=clock) + handler, _ = paced(pacer, park_above_seconds=5.0) + await handler(command("cmd-1")) + + clock.advance(10.0) + with pytest.raises(TransientFetchError): + await handler(command("cmd-2")) + + assert pacer.wait_seconds(URL) == pytest.approx(50.0) + + +async def test_an_uninjected_pacer_still_paces(fake_redis, tmp_path, monkeypatch): + """The seam fails *open*, so the default must not be "no pacing". + + A byte path that quietly stopped pacing looks exactly like one that is + working — no error, no fact, nothing in the journal — so the pacer is built + from settings rather than left to the caller to remember. This is the + ``BlobUsage`` argument one seam over: wired wrong, both halves stay + individually correct and the guard is simply never reached. + """ + monkeypatch.setenv("REPLICATOR_MIN_HOST_INTERVAL_SECONDS", "30") + get_settings.cache_clear() + handler = build_handler( + fetcher=(fetcher := FakeFetcher()), + store=LocalBlobStore(tmp_path), + client=fake_redis, + settings=get_settings(), + ) + + await handler(command("cmd-1")) + with pytest.raises(TransientFetchError, match="politeness window"): + await handler(command("cmd-2")) + + assert fetcher.urls == [URL] + + +async def test_the_default_park_bound_is_the_poll_window(fake_redis, tmp_path): + """Sleep up to one blocking read, park beyond it. + + Derived from ``REPLICATOR_READ_BLOCK_MS`` rather than given its own knob: a + wait no longer than a poll the loop already performs adds nothing to the + worst-case shutdown latency ``TimeoutStopSec`` is sized for + (``tests/test_deploy.py``), and a second setting would be one more number to + keep in agreement with the unit. + """ + settings = get_settings() + handler = build_handler( + fetcher=FakeFetcher(), + store=LocalBlobStore(tmp_path), + client=fake_redis, + settings=settings, + pacer=HostPacer(settings.read_block_ms / 1000 + 1), + ) + + await handler(command("cmd-1")) + + with pytest.raises(TransientFetchError, match="politeness window"): + await handler(command("cmd-2")) + + +async def test_a_wait_inside_the_poll_window_sleeps_rather_than_parks(fake_redis, tmp_path): + """The other half of the bound (CR #11). + + Asserted from below as well as above, because the park assertion alone would + also pass with a default of ``0`` — which would park every paced command and + put the whole corpus on a 60-second reclaim cadence. That is the 60x failure + the split exists to avoid, and it must not be one line from passing. + """ + settings = get_settings() + handler = build_handler( + fetcher=(fetcher := FakeFetcher()), + store=LocalBlobStore(tmp_path), + client=fake_redis, + settings=settings, + pacer=HostPacer(0.05), + ) + + await handler(command("cmd-1")) + await handler(command("cmd-2")) + + assert fetcher.urls == [URL, URL] + + +async def test_an_unpaced_handler_is_the_pre_12_byte_path(fake_redis, tmp_path): + """Zero interval is the escape hatch, and it must cost nothing.""" + fetcher = FakeFetcher() + handler = build_handler( + fetcher=fetcher, + store=LocalBlobStore(tmp_path), + client=fake_redis, + settings=get_settings(), + pacer=HostPacer(0.0), + blobs_topic=streams.CONTENT_BLOBS, + ) + + await handler(command("cmd-1")) + await handler(command("cmd-2")) + + assert fetcher.urls == [URL, URL] diff --git a/tests/worker/test_pacing.py b/tests/worker/test_pacing.py new file mode 100644 index 0000000..cbf3ef4 --- /dev/null +++ b/tests/worker/test_pacing.py @@ -0,0 +1,159 @@ +"""Per-host pacing: the interim politeness default (#12). + +The state machine only. How the handler spends the wait it is told about — sleep +below the bound, park above it — is ``tests/worker/test_handler_pacing.py``. +""" + +import pytest + +from src.worker.pacing import MAX_TRACKED_HOSTS, HostPacer + + +def test_the_first_request_to_a_host_waits_for_nothing(): + """There is no previous request to space this one from.""" + pacer = HostPacer(1.0) + + assert pacer.wait_seconds("https://example.test/a") == 0.0 + + +def test_a_second_request_to_the_same_host_waits_out_the_interval(clock): + pacer = HostPacer(1.0, clock=clock) + pacer.record("https://example.test/a") + + clock.advance(0.25) + + assert pacer.wait_seconds("https://example.test/b") == pytest.approx(0.75) + + +def test_the_wait_is_gone_once_the_interval_has_elapsed(clock): + pacer = HostPacer(1.0, clock=clock) + pacer.record("https://example.test/a") + + clock.advance(1.0) + + assert pacer.wait_seconds("https://example.test/b") == 0.0 + + +def test_pacing_is_per_host(clock): + """A slow origin must not throttle an unrelated one.""" + pacer = HostPacer(1.0, clock=clock) + pacer.record("https://example.test/a") + + assert pacer.wait_seconds("https://other.test/a") == 0.0 + + +def test_the_host_is_the_key_regardless_of_scheme_port_or_case(clock): + """One machine, one budget. + + Politeness is about how often a *server* is asked, so a second scheme or port + is the same origin's socket and a capitalized host is the same name. Keying + on anything finer would let an issuer multiply its own rate limit by spelling + the URL differently. + """ + pacer = HostPacer(1.0, clock=clock) + pacer.record("https://example.test/a") + + assert pacer.wait_seconds("http://EXAMPLE.test:8080/b") > 0 + + +def test_a_hostless_url_is_never_paced(): + """It cannot be fetched either — the driver refuses it as not_fetchable. + + Returning a wait would park a permanently bad command in the PEL instead of + letting it reach the terminal fact the issuer is waiting for. + """ + pacer = HostPacer(1.0) + pacer.record("not-a-url") + + assert pacer.wait_seconds("not-a-url") == 0.0 + + +def test_an_unparseable_authority_is_never_paced(): + """``urlsplit`` raises on a malformed IPv6 literal rather than returning None. + + Same outcome as a hostless URL and for the same reason — it is unfetchable — + but it arrives as an exception, and an uncaught one here would turn a bad URL + into an unclassified handler failure retried to the delivery ceiling. + """ + pacer = HostPacer(1.0) + pacer.record("http://[::1") + + assert pacer.wait_seconds("http://[::1") == 0.0 + + +def test_a_zero_interval_disables_pacing(clock): + pacer = HostPacer(0.0, clock=clock) + pacer.record("https://example.test/a") + + assert pacer.wait_seconds("https://example.test/a") == 0.0 + + +def test_recording_prunes_hosts_that_no_longer_constrain_anything(clock): + """In-memory derived state still has to be bounded. + + A worker fetching a long tail of one-off hosts would otherwise grow a + permanent entry per host. An entry older than the interval imposes no wait, + so dropping it changes no decision this pacer can make. + """ + pacer = HostPacer(1.0, clock=clock) + for n in range(MAX_TRACKED_HOSTS + 1): + pacer.record(f"https://host{n}.test/a") + + clock.advance(1.0) + pacer.record("https://trigger.test/a") + + assert pacer.tracked_hosts == 1 + + +def test_pruning_keeps_the_hosts_that_do_constrain_something(clock): + """The bound must not be enforced by forgetting a host that is still waiting.""" + pacer = HostPacer(60.0, clock=clock) + for n in range(MAX_TRACKED_HOSTS + 1): + pacer.record(f"https://host{n}.test/a") + + assert pacer.wait_seconds("https://host0.test/a") > 0 + + +def test_a_prune_that_reclaims_nothing_is_not_retried_every_record(clock, monkeypatch): + """Over the bound with nothing reclaimable is a steady state, not an emergency. + + Every host inside a long interval means the bound cannot be honoured, and + retrying the O(n) rebuild on each subsequent record would make a full dict + copy per message for as long as that lasts (CR #7). The retry waits until the + oldest entry could plausibly have aged out. + + **Why counting calls on a private attribute is safe here (CR #12).** The + worry about a spy like this is that a refactor stops going through the + patched attribute, leaving a test that passes while measuring nothing. The + ``prunes == 1`` assertion is what forecloses that: inline ``_prune``'s body + into ``record`` and this fails ``assert 0 == 1`` rather than passing. + Verified by doing exactly that, not by argument. The ``tracked_hosts`` + assertions below say the same thing in observable terms, so a reader who + distrusts the spy still has the behaviour. + """ + pacer = HostPacer(60.0, clock=clock) + for n in range(MAX_TRACKED_HOSTS + 1): + pacer.record(f"https://host{n}.test/a") + + prunes = 0 + original = pacer._prune + + def counting(now: float) -> None: + nonlocal prunes + prunes += 1 + original(now) + + monkeypatch.setattr(pacer, "_prune", counting) + for n in range(10): + pacer.record(f"https://later{n}.test/a") + + assert prunes == 0 + # Observable half: nothing was reclaimable, so the set only grew — the bound + # is deliberately exceeded rather than honoured by forgetting a waiting host. + assert pacer.tracked_hosts == MAX_TRACKED_HOSTS + 11 + + clock.advance(61.0) + pacer.record("https://trigger.test/a") + + assert prunes == 1 + assert pacer.tracked_hosts == 1