diff --git a/AGENTS.md b/AGENTS.md index 5c151d2..4dde0ee 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,7 @@ Python ≥3.12, uv, pytest, ruff. `ty` is available as a **non-gating** type che uv run --no-project --with 'google-cloud-storage>=2,<4' python scripts/sync_wheelhouse.py ``` -Auth is ADC: on the VM the SA key at `GOOGLE_APPLICATION_CREDENTIALS` (`/etc/replicator/co-pypi-reader.json`), in CI a keyless WIF token. Pin the current minor — `>=0.7.7,<0.8`. The **patch** floor is load-bearing, not tidiness: the change-bus payloads are `extra="ignore"`, so on an older wheel a model constructed with fields it does not have yet succeeds and silently discards them. Raise the floor with every co-core feature the code starts depending on, or a version skew publishes facts that look right and carry nothing (#10). The 0.7.7 floor is the exception that fails *loudly* — `AsyncBusTailReader` and `FetchPolicyState` do not exist below it, so a skew is an ImportError rather than a silent discard (#19). +Auth is ADC: on the VM the SA key at `GOOGLE_APPLICATION_CREDENTIALS` (`/etc/replicator/co-pypi-reader.json`), in CI a keyless WIF token. Pin the current minor — `>=0.8.0,<0.9`. The **patch** floor is load-bearing, not tidiness: the change-bus payloads are `extra="ignore"`, so on an older wheel a model constructed with fields it does not have yet succeeds and silently discards them. Raise the floor with every co-core feature the code starts depending on, or a version skew publishes facts that look right and carry nothing (#10). The 0.8.0 floor fails *loudly* instead: `info_source_id` is required on all three payloads, so a skew is a ValidationError at construction rather than a silent discard (#19, #28). ## Code Exploration Policy @@ -115,9 +115,12 @@ Every variable the service reads, with the reasoning behind each default: Replicator is a **consumer** first. Follow the conventions co-core and the archiver producer established: -- **At-least-once ⇒ idempotent.** The command dedupes on `command_id`, the fact on - `content_fingerprint`, and `fetch_failed` on neither — storage identity and - correlation identity are not interchangeable. +- **At-least-once ⇒ idempotent.** The command dedupes on `command_id`; both facts + are keyed per *occurrence* (`content_fingerprint:command_id`, + `command_id:occurred_at`), so nothing an issuer waits on can collapse — storage + identity and correlation identity are not interchangeable. `info_source_id` + rides both and is **echoed, never read**: the `tests/test_boundaries.py` + carve-out is one field wide, and widening it edits the charter (#28). - **Store, then publish — never the reverse.** A fact pointing at bytes that are not there is unrepairable by the consumer; stored bytes with no fact repair themselves on the reclaim. diff --git a/README.md b/README.md index f96782b..f610aa6 100644 --- a/README.md +++ b/README.md @@ -30,9 +30,9 @@ The founding design lives in first — it is the normative issuer contract and its permanent home, with the refusal list, failure taxonomy and trust posture in its companion [`content-fetch-issuer-reference.md`](docs/contracts/content-fetch-issuer-reference.md). -Publish through co-core's `to_wire`, never hand-rolled fields; and because the wire carries no -domain identity, correlation is entirely the issuer's job. Most ways of getting either wrong fail -silently. +Publish through co-core's `to_wire`, never hand-rolled fields. The wire carries one domain key — +`info_source_id`, echoed onto both facts and read by nothing here — but correlation is still +entirely the issuer's job, on `command_id`. Most ways of getting either wrong fail silently. ## Shape @@ -107,14 +107,15 @@ see [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md) for what each one is and why. Nothing in the cluster issues `content.fetch` commands until the Watcher cutover, so `scripts/seed_fetch.py` is the issuer. The target is never defaulted — `--redis-url` and `--topic` are both required, and db 0 + `content.fetch` (the one pair the running worker -consumes, and therefore actually fetches over the network) additionally needs `--production`: +consumes, and therefore actually fetches over the network) additionally needs `--production` +**and** a real `--info-source-id`, since the facts it publishes echo that value cluster-wide: ```bash # Fetches the local /health app — a target we control, so the smoke test costs # nobody else a request. Start it first (see Dev server below). uv run python -m scripts.seed_fetch \ --redis-url redis://localhost:6379/0 --topic content.fetch \ - --production --watch http://localhost:8041/health + --production --info-source-id isrc-01J9ZK7Q --watch http://localhost:8041/health ``` `--watch` tails the fact stream until each command has an outcome — a `blob_available`, or a diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index df78698..f529376 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -44,14 +44,15 @@ Each rule `AGENTS.md` states in one line, with the reasoning that makes it non-n - **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. Its companion `docs/contracts/content-fetch-issuer-reference.md` (#24) carries the lookup half — the request-options refusal list, the failure taxonomy, the four silent conditions, and the trust posture — and is equally normative. Issuer-side repos (Watcher, Phase 4) link to the contract rather than copying it. Anything asserted in either 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`. +- **`blob_available` carries the fetch, not just the bytes.** Seven optional fields beyond the blob itself: six describing the *fetch* (#10, cannobserv#271 + #279 — v0.7.5) — `final_url`, `status_code`, `fetched_at`, `content_type_raw`, `etag`, `last_modified` — and `blob_expires_at`, which describes the *store* (cannobserv#301, populated by #28: `stored_at + REPLICATOR_BLOB_TTL_SECONDS`, read before the store so the announced horizon can only fall earlier than the real reap; see [STORAGE.md](STORAGE.md)). 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 since #19 the numbers actually arrive.** `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" — **#19 did not resolve it**, it only made the fix more defensible. 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. The pacer resolves each host's interval through the `policy` seam — a bare callable, not the map, so `handler.py` stays ignorant of `content.fetch-policy` the way `loop.py` stays ignorant of `content.blobs` (#12, #19, watcher#245, cannobserv#285). - **`content.fetch-policy` is the third stream kind, and it is read groupless.** `src/worker/policy.py` replays it from `0-0` at boot and tails it thereafter; `FetchPolicyMap` is the state, `run_policy_reader` the poll loop, a peer of the consume loop and the retention sweep in `_run_until_first_exit`. Hyphen, not a third dot segment — `content.fetch.policy` collides with the `.dlq` derivation of the command stream, so use `streams.CONTENT_FETCH_POLICY`. **No consumer group**: every worker needs every message, and a group here grows a PEL nothing drains — hence no `ack` and no DLQ either, and a frame that will never decode is skipped by forcing the cursor (`seek`). Reads are **`count=1` throughout**, deliberately: `AsyncBusTailReader` advances its cursor only on a fully-decoded batch, so recovering from a poison frame at `count>1` means draining the well-formed prefix at `count=1` *before* seeking past it (`seek` only moves forward) — one message per host per republish makes the extra round trips free and deletes that ordering entirely. **`AsyncBusTailReader.replay()` is unusable and `PolicyReader` deliberately omits it** (#19 CR #1): it accumulates across many `read` calls and returns the list only on a clean finish, so *any* raise part-way through discards everything it read while the cursor has already moved — a poison frame at position *k* silently loses the *k−1* policies ahead of it, permanently. Drive `read` yourself and apply each batch as it arrives. It is the obvious API and its name says exactly what a caller wants, which is why this warning is here rather than only in the docstring. Recovery is bounded (`MAX_POISON_SKIPS`, shared with `loop.py`) because an anomaly clears the outage counter, so a run of them would otherwise spin at broker round-trip speed with backoff permanently disarmed; and the boot replay takes the **stop event** so a SIGTERM during an untrimmed stream's replay is not ignored until it finishes. **Replay runs synchronously before the consume loop starts**, or the worker's opening commands are paced against an empty map; a failed replay is absorbed rather than fatal, because the cursor advanced only over what decoded and the tail drains the rest. Four ways to apply a message wrongly and silently — branch on `revoked` before the interval (`min_interval_seconds` is `None` on a tombstone), treat `0.0` as a value rather than an absence, `isinstance`-check the payload (`from_wire`'s table is global, so a `blob_available` here decodes into the wrong model rather than raising), and guard on `occurred_at` per host so a stale full-set republish cannot revert a newer change. The reader **absorbs its own failures** like `run_sweeper` does: politeness is not load-bearing for correctness, and a broker that is genuinely gone surfaces through the consume loop, which has the delivery obligations. Its blocking read uses `REPLICATOR_READ_BLOCK_MS`, the same window the consume loop uses and concurrently with it, so `TimeoutStopSec` gains no term (#19, cannobserv#285 — v0.7.7). - **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. +- **At-least-once ⇒ idempotent, and every key names an *occurrence*.** The **command** dedupes on `command_id`. Both facts are keyed per emission — `blob_available` on `content_fingerprint:command_id` (cannobserv#300; the bare fingerprint through 0.7.7) and `fetch_failed` on `command_id:occurred_at` — so a consumer's dedup-on-key can neither collapse a multi-emission failure sequence and drop the terminal event, nor collapse two InfoSources' fetches of one URL into a single fact naming whichever issuer won the race. The bare-fingerprint key also left the *second* command permanently unclosed, which read as a slow origin rather than a bug. The fingerprint is *storage* identity, `command_id` is *correlation* identity, and they are not interchangeable — a consumer deduping its inbox on the fingerprint still loses a correlation, which is why MUST-5 survives the re-key. Content-addressed storage makes re-storing identical bytes a no-op regardless. +- **`info_source_id` rides both facts and is read by nothing here (#28).** Required on the command and on both facts since co-core 0.8.0, copied verbatim at emit time in `handler.py` and `reporter.py` and carried between them on `loop.py::FailureReport`. It is not parsed, not validated, not in any dedupe or envelope key, and not an input to pacing or policy. The boundaries charter enforces that mechanically rather than by prose: `tests/test_boundaries.py` exempts the token in exactly those three modules and separately asserts the value never appears in a comparison, a branch test, a subscript, or an f-string anywhere in `src/` — naming it is mechanics, keying on it is a domain model. - **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. - **Validation posture:** use the canonical `extra="ignore"` models; **branch on `schema_version` before destructuring**; tolerate additive producer fields. Never use the strict `*Emit` classes on the consume path. - **Batch-poison caveat:** `AsyncBusConsumer.read(count>1)` raises `BusMessageAnomaly` on a malformed frame *before* returning the well-formed ones in the batch. Read `count=1`, or catch the anomaly and route via `dead_letter`. `from_wire` is deliberately fail-loud. diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 885f060..5717824 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -51,13 +51,15 @@ uv run python -m scripts.seed_fetch \ https://example.test/a https://example.test/b # The live loop. --production is required for db 0 + content.fetch, because the -# running service will fetch these URLs for real. --watch tails the fact stream -# until each command has an outcome — blob_available, or a fetch_failed naming -# the reason. Exit 1 if a command failed or no fact ever arrived. +# running service will fetch these URLs for real — and so is a real +# --info-source-id, because the facts it publishes echo that value onto the +# cluster's own content.blobs. --watch tails the fact stream until each command +# has an outcome — blob_available, or a fetch_failed naming the reason. Exit 1 if +# a command failed or no fact ever arrived; exit 2 if either opt-in is missing. # The target below is the local /health app — start it first (see API, below). uv run python -m scripts.seed_fetch \ --redis-url redis://localhost:6379/0 --topic content.fetch \ - --production --watch http://localhost:8041/health + --production --info-source-id isrc-01J9ZK7Q --watch http://localhost:8041/health ``` `--watch` reads `content.blobs` for `content.fetch` and `.blobs` otherwise, so the @@ -65,6 +67,12 @@ scratch invocation above watches its own facts rather than production's. `--blob overrides that. One stream, both outcomes: an issuer needs a single consumer group to see whether its command produced bytes or a reason. +`--info-source-id` sets the domain key the command carries and both facts echo, required on the +wire since co-core 0.8.0 (#28). It defaults to `seed-harness-not-a-real-info-source`, which no +issuer's InfoSource table contains — so a fact a seed run puts on a scratch stream is recognizably +synthetic. **The live target refuses that default, and a blank value**, exiting 2: a real fetch +broadcasts whatever is passed here to the cluster, so it has to name a real InfoSource. + `--header` and `--timeout` set the command's per-fetch request options (#11). They apply to every URL in the run, and omitting them is the pre-#11 wire exactly. diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index ca4ed90..c34350e 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -41,6 +41,24 @@ The **redis-py client** resolves `>=5,<8` transitively via `co-core-aio[bus]`. D The copy is deliberate, for the same reason `/etc/replicator/.env` is not read from the repo: the live unit must survive a repo reset, a worktree switch, or a branch checkout that happens to be mid-edit. +### The co-core 0.8.0 cutover is a two-repo deploy, streams flushed between (#28) + +`schema_version` stays 1, and that is a decision rather than an oversight: bumping to 2 would imply +a v1 consumers must branch on, when the correct operation is to discard the v1 messages. The wire +is pre-production, so it is discardable. + +**Flushing is a prerequisite step, not cleanup.** `content.fetch`, `content.blobs`, and **both +`.dlq` streams** — a v1 dead-letter cannot be replayed under 0.8.0, so leaving it is leaving a trap +for whoever triages next. Add `replicator:cmd:*` if any `command_id` will be reused across the +flush. **Not** `content.fetch-policy`: it is a groupless state stream, and flushing it leaves every +worker with an empty policy map until the next republish. + +**Ship with [CannObserv/watcher#252](https://github.com/CannObserv/watcher/issues/252).** Required +fields mean a half-deployed cluster does not degrade, it dead-letters: a Replicator on 0.8.0 fails +`from_wire` on any command from a Watcher still on 0.7.x, and that failure destroys the +`command_id` correlator before any fact can name it. The two may be worked in parallel; they must +land together. + **Dev server workflow** (the `/health` app, port 8041 so a future live service stays up): ```bash @@ -62,7 +80,7 @@ In `/etc/replicator/.env` (read by the service): - `GOOGLE_APPLICATION_CREDENTIALS` — SA key for the wheelhouse mirror (`/etc/replicator/co-pypi-reader.json`) - `REPLICATOR_REDIS_URL` — change-bus client URL; default `redis://localhost:6379/0` - `REPLICATOR_BLOB_DIR` — temp-storage root for fetched bytes; default `blobs`. Resolved to an absolute path at store construction — `file://` URIs require it -- `REPLICATOR_BLOB_TTL_SECONDS` — how long a blob survives after it was **last referenced**; default `604800` (7 days). Measured from mtime, which the store refreshes on its short-circuit. The number is a published commitment to archiver (archiver#118), not a local tuning knob — raise it if a `content.blobs` consumer says it needs longer +- `REPLICATOR_BLOB_TTL_SECONDS` — how long a blob survives after it was **last referenced**; default `604800` (7 days), accepted range `0 < n ≤ 315360000` (10 years). Measured from mtime, which the store refreshes on its short-circuit, and published on each fact as `blob_expires_at`. The number is a published commitment to archiver (archiver#118), not a local tuning knob — raise it if a `content.blobs` consumer says it needs longer. **Out of range fails at startup**, not at the first fetch: the horizon is arithmetic, and an absurd value would raise mid-handler *after* the bytes were stored, orphaning them (#28) - `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS` — how often the tree is walked; default `900`. Also the staleness bound on the measured byte total the ceiling reads - `REPLICATOR_BLOB_TEMP_GRACE_SECONDS` — how long a `.tmp` may live before the sweep treats it as debris; default `3600`. Deliberately unrelated to the TTL and far shorter — see **Retention** in [STORAGE.md](STORAGE.md) diff --git a/docs/STORAGE.md b/docs/STORAGE.md index 60a1b88..c9b5424 100644 --- a/docs/STORAGE.md +++ b/docs/STORAGE.md @@ -15,6 +15,7 @@ obligation is Replicator's. `docs/plans/2026-07-31-replicator-mvp-open-questions-design.md` §4 scope-cut retention; #5 settles it. Replicator is the producer in archiver's temp-cache protocol, where **the producer cleans up**. - **The TTL runs from last reference, not first store.** `store` short-circuits on an existing content-addressed path but its caller publishes a fresh `blob_available` either way, so a re-fetch of unchanged bytes would otherwise announce a blob already partway through its TTL. `LocalBlobStore` therefore `os.utime`s on the short-circuit branch, swallowing `ENOENT` — the sweep can unlink between the existence check and the touch, and the fallout of that race is a `blob_uri` that fails to open, not a dead-lettered command. +- **The horizon that clock implies is published, and it errs early.** `blob_available.blob_expires_at` carries `stored_at + REPLICATOR_BLOB_TTL_SECONDS` (cannobserv#301, #28), so a consumer records a cache expiry instead of re-deriving one from a retention policy it does not own and a clock start it cannot see. `stored_at` is read *before* `store`, so it lands at or before the mtime the sweep measures against; the sweep then runs only every `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS`. Both terms push the real reap later than the announced horizon, which is the only safe direction — a consumer acting on it re-fetches early rather than opening a dead `blob_uri`. - **The blob tree holds three populations, and they are not interchangeable.** Finished blobs (`//.bin`) reap on the TTL; in-flight temporaries (`...tmp`) are **not garbage** — reaping one makes the writer's `os.replace` fail with `ENOENT` and dead-letters a good command — so the sweep matches `*.bin`, never `iterdir()`, and ages temps out on their own much longer grace. Empty shard directories go **last**, by `rmdir` only, whose refusal to touch a non-empty directory is the safety property. - **The ceiling is backpressure, not a faster clock.** Over `REPLICATOR_BLOB_MAX_TOTAL_BYTES` the byte path raises `TransientFetchError` *before* fetching, so the command stays in the PEL and returns via `claim_stale` once a sweep frees space. What it measures is everything the tree holds — surviving blobs **and** the temporaries the sweep is waiting out, since a crash loop fills the disk with debris the ceiling would otherwise not see. The per-population counts stay split in the sweep log so a rising temp count cannot hide inside a healthy blob one. Reaping a blob still inside its TTL to make room would convert a local disk problem into a `blob_uri` another repo cannot open — the one failure mode with no local symptom. - **One `BlobUsage`, two writers.** The sweep's `observe` is the measured total; the byte path's `add` is the estimate between sweeps, because a burst can cross the ceiling long before the tree is walked again. Wiring the two halves to separate instances leaves both individually correct and the guard permanently unreachable — `tests/worker/test_main.py` pins the identity. diff --git a/docs/contracts/content-fetch-issuer-contract.md b/docs/contracts/content-fetch-issuer-contract.md index 0fb0988..f4120c9 100644 --- a/docs/contracts/content-fetch-issuer-contract.md +++ b/docs/contracts/content-fetch-issuer-contract.md @@ -1,41 +1,44 @@ # The `content.fetch` issuer contract -**Status:** normative. **Home:** this file, in the Replicator repo — Replicator is the sole -consumer of `content.fetch` and the sole producer of `content.blobs`, so the contract lives with -the behaviour it describes. Issuer-side repos link here rather than copying; a copy would drift -from the code the day it was written. +**Status:** normative. **Home:** this file — Replicator is the sole consumer of `content.fetch` and +the sole producer of `content.blobs`, so the contract lives with the behaviour it describes. Issuer +repos link here rather than copying; a copy drifts from the code the day it is 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. +**Audience:** any service publishing a `ContentFetchCommand` — today +[`scripts/seed_fetch.py`](../../scripts/seed_fetch.py), from Phase 4 Watcher. **Companion.** [`content-fetch-issuer-reference.md`](content-fetch-issuer-reference.md) carries the -parts an issuer *looks up* rather than reads through — the request-options refusal list, the -reasoning behind the enriched `blob_available` fields, the failure taxonomy, and the trust posture. -Equally normative; split out in #24 so this file stays readable start to finish. Index at the end. +parts an issuer *looks up* rather than reads through — equally normative, split out in #24 so this +file stays readable start to finish. Index at the end. **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 -edits this file. Whatever is asserted here is asserted about this repo's code: edit one, edit both. - -**Why this document exists.** The bus wire shape is deliberately domain-agnostic: a `content.fetch` -*payload* carries `{command_id, url}`, and `content.blobs` carries **both outcomes** of that -command — `{content_fingerprint, blob_uri, size_bytes, media_type, url, command_id?}` on success, -`{command_id, url, reason, terminal, status_code?, attempts?, detail?}` on failure. All three sit -inside a co-core envelope, which is a shape in its own right and the first thing a producer must -get right (see **The frame**). There is **no `info_source_id`, and no other domain identity, -anywhere in any of them** — Replicator fetches bytes and knows nothing about what they mean. That -keeps Replicator clean, and it pushes the whole of correlation onto the issuer. Most of what -follows fails *silently* when it is got wrong: no error, no dead letter, no log on Replicator's -side — just a fact nobody can match, or a command that was never run. +this contract will never grow. Read it before proposing a payload addition (#12). + +**Changing this document.** "Link, don't copy" only holds if a change reaches the issuers, so 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 edits this file. What is asserted here is asserted about this repo's code: edit both. + +**Why this document exists.** One command stream and one fact stream carrying **both outcomes**, +each payload inside a co-core envelope — a shape in its own right, and the first thing a producer +must get right (see **The frame**). Fields are tabulated under **The payload**. + +`info_source_id` is **carried, not understood** (cannobserv#300, #28): echoed verbatim onto both +facts, never parsed, deduped on, keyed on, or read by any routing or policy decision. It travels so +a consumer of a *broadcast* stream can tell what a fact is about without holding the issuer's +private map. [`replicator-boundaries.md`](replicator-boundaries.md) keeps it at that — its +executable half forbids the value in any branch, lookup key, or constructed string in `src/`. + +Correlation still rests entirely on the issuer, and most of what follows fails *silently* when it is +got wrong: no error, no dead letter, no log on Replicator's side — just a fact nobody can match, or +a command that was never run. Version history and the co-core floor: [the reference](content-fetch-issuer-reference.md#version-history). Replicator requires -**co-core ≥ 0.7.5**. Founding rationale: +**co-core ≥ 0.8.0**, the release that made `info_source_id` and `BlobAvailableEvent.command_id` +required — a 0.7.x command does not degrade, it fails `from_wire` and dead-letters. Founding +rationale: [`docs/plans/2026-06-25-replicator-mvp-design.md`](../plans/2026-06-25-replicator-mvp-design.md). --- @@ -49,35 +52,39 @@ produces exactly six keys, all of them derived from the model. None are caller-s | Key | Value | |---|---| | `key` | the envelope's idempotency key, derived by `to_wire` — see the table below | -| `payload` | the model, JSON-serialized. This is where `command_id`, `url`, and everything in the tables below actually live | +| `payload` | the model, JSON-serialized — where everything in the tables below actually lives | | `event_type` | `content_fetch` / `blob_available` / `fetch_failed` — how `from_wire` picks a model | | `schema_version` | stringified | -| `occurred_at` | ISO 8601 UTC, **tz-aware** — see the command table | +| `occurred_at` | ISO 8601 UTC, **tz-aware** — see below | | `content_type` | `application/json` | -`key` is derived per payload type, and the three rules are not the same shape: +`key` is derived per payload type, and the three rules differ: | Payload | Derived `key` | |---|---| | `content_fetch` | `command_id` | -| `blob_available` | `content_fingerprint` | +| `blob_available` | **`content_fingerprint:command_id`** — per *occurrence*, not per bytes | | `fetch_failed` | **`command_id:occurred_at`** — deliberately *not* the bare `command_id` | -The `fetch_failed` rule is load-bearing (cannobserv#270). One command can emit more than one -failure over its life, so a bare-`command_id` key would make a consumer that dedupes on the -envelope key collapse the sequence and drop the **terminal** event — the one that closes the -pending entry. Correlation rides on the `command_id` *field*, never on the key. +Neither fact is keyed on a bare identifier: a key naming less than the occurrence collapses +occurrences. `fetch_failed` (cannobserv#270) would drop a multi-failure command's **terminal** +event; `blob_available` (cannobserv#300, the bare `content_fingerprint` through 0.7.7) collapsed two +InfoSources fetching one URL into one fact naming whichever issuer won the race, and left the second +command never closed — which reads as a slow origin, not a bug. **The re-key is a delivery-behaviour +change**: emissions that used to collapse now all deliver. MUST-4 already required idempotence, so +it moves in the safe direction, but the volume differs. Correlation rides on the `command_id` +*field*, never on the key. An issuer that `XADD`s `command_id` and `url` as top-level fields produces a frame `from_wire` -cannot decode. It raises `BusMessageAnomaly` from inside the consumer's `read`, and Replicator -routes it to `content.fetch.dlq` — **silently, by the terms of MUST-6**. This is the single -likeliest way to get the contract wrong, so it is stated before the field tables rather than after. +cannot decode: it raises `BusMessageAnomaly` inside the consumer's `read`, and Replicator routes it +to `content.fetch.dlq` — **silently, by the terms of MUST-6**. The single likeliest way to get this +contract wrong, so it is stated before the field tables rather than after. `key` is **not** load-bearing on the consume path: Replicator dedupes on `payload.command_id`, -never on the envelope key. Its value is operational — see +never on the envelope key. What its value *is* for: [the reference](content-fetch-issuer-reference.md#what-the-envelope-key-is-for). -## The payload (inside `payload`) +## The payload ### The command @@ -87,38 +94,34 @@ never on the envelope key. Its value is operational — see |---|---|---| | `schema_version` | `int` = 1 | Replicator supports **1 only**; anything else dead-letters (see the [failure taxonomy](content-fetch-issuer-reference.md#failure-taxonomy-what-happens-and-what-the-issuer-sees)) | | `event_type` | `"content_fetch"` | | -| `occurred_at` | `datetime` | **tz-aware UTC, enforced.** Not used for ordering or expiry by Replicator | +| `occurred_at` | `datetime` | **tz-aware UTC, enforced.** Not used for ordering or expiry here | | `command_id` | `str` | **The idempotency key and the sole correlator.** See MUST-1 | | `url` | `str` | What to fetch. **Not** a key. See MUST-3 | -| `headers` | `dict[str, str] \| None` | co-core ≥ 0.7.3, **honoured since #11**. Merged over the fetcher's defaults, issuer wins. Guards below | -| `timeout_seconds` | `float \| None` | co-core ≥ 0.7.3, **honoured since #11**. Seconds; bounded above. `None` = the driver default | +| `info_source_id` | `str` | **Required.** The domain object this fetch is for. Echoed onto both facts, read by nothing here (#28) | +| `headers` | `dict[str, str] \| None` | **Honoured since #11.** Merged over the fetcher's defaults, issuer wins. Guards below | +| `timeout_seconds` | `float \| None` | **Honoured since #11.** Seconds; bounded above. `None` = the driver default | -> **`occurred_at` must carry a timezone.** Since co-core v0.7.2 (cannobserv#273) it is an -> `AwareDatetime` on every payload: a **naive** value is rejected fail-loud rather than assumed -> to be UTC, because "assume UTC" corrupts the instant when a producer stamps a naive local time. -> An aware non-UTC value is normalized. A naive one fails `from_wire` inside Replicator's `read`, -> which means it dead-letters as a malformed frame and is one of the rows that stay **silent** — -> the same shape as an issuer that flattened the envelope. `datetime.now(UTC)`, not -> `datetime.now()`. +> **`occurred_at` must carry a timezone.** An `AwareDatetime` on every payload since co-core +> v0.7.2 (cannobserv#273): a **naive** value is rejected rather than assumed to be UTC, which would +> corrupt the instant. An aware non-UTC value is normalized. A naive one fails `from_wire` inside +> Replicator's `read`, so it dead-letters as a malformed frame and stays **silent** — the same shape +> as a flattened envelope. `datetime.now(UTC)`, not `datetime.now()`. ### Request options: what Replicator will send, and what it refuses (#11) -`headers` and `timeout_seconds` shape the individual fetch. Both are optional; `None` on both means +`headers` and `timeout_seconds` shape the individual fetch. Both are optional, and `None` on both is **exactly** the pre-#11 behaviour — the fetcher's own `user-agent` and its 30 s timeout, byte for -byte. An issuer that sends neither is unaffected by any of this. - -Two rules bind an issuer that does send them: +byte — so an issuer that sends neither is unaffected by any of this. Two rules bind one that does: - **Everything on the refusal list is refused, not adjusted.** A refusal is a terminal `fetch_failed` · `invalid_request_options` plus the DLQ, arriving **before any request goes out** — a refused command never reaches the origin at all. -- **Neither field touches identity.** `command_id` remains the sole dedupe key and the sole - correlator. Two commands differing only in options are two fetch occasions (MUST-1 unchanged); - a *redelivery* carrying different options is still the same command and is still deduped. +- **Neither field touches identity.** `command_id` remains the sole dedupe key and correlator: two + commands differing only in options are two fetch occasions, and a *redelivery* carrying different + options is still the same command. -The refusal list — hop-by-hop and derived headers, the token and byte-range rules on names and -values, the count and size ceilings, the timeout bounds — plus the header-name folding rule and the -reasoning behind each, is in +The refusal list itself — hop-by-hop and derived headers, the rules on names and values, the +ceilings, the timeout bounds, the header-name folding rule, and the reasoning behind each — is in [the reference](content-fetch-issuer-reference.md#request-options-what-replicator-will-send-and-what-it-refuses-11). ### The success fact @@ -130,36 +133,34 @@ reasoning behind each, is in | `schema_version` | `int` = 1 | | | `event_type` | `"blob_available"` | | | `occurred_at` | `datetime` | UTC, stamped at publish | -| `content_fingerprint` | `str` | sha256 of the bytes. Content identity — **not** a correlator | -| `blob_uri` | `str` | `file://///.bin`. Temporary; see MUST-7 | +| `content_fingerprint` | `str` | sha256 of the bytes. Content identity, **not** a correlator | +| `blob_uri` | `str` | `file://///.bin`. Temporary — MUST-7 | | `size_bytes` | `int` | | | `media_type` | `str` | Normalized, `charset` dropped; `application/octet-stream` when absent | -| `url` | `str` | Echoed from the command. Confirmation and debugging only | -| `command_id` | `str \| None` | Echoed from the command. `None` only for non-command emits | -| `final_url` | `str \| None` | co-core ≥ 0.7.3 (model), **≥ 0.7.5 to be populated**. Where the fetch **landed** after redirects. `None` = *unknown*, see below | -| `status_code` | `int \| None` | co-core ≥ 0.7.3. **Always 2xx on this fact** — see below | -| `fetched_at` | `datetime \| None` | co-core ≥ 0.7.3. tz-aware UTC. When the bytes were on the **wire**, not when the fact was published | -| `content_type_raw` | `str \| None` | co-core ≥ 0.7.3. The **verbatim** `Content-Type`, `charset` and all. `None` = *the origin sent none*, see below | -| `etag` | `str \| None` | co-core ≥ 0.7.3. Verbatim, `W/` prefix and quotes included. Replay unparsed in `If-None-Match` | -| `last_modified` | `str \| None` | co-core ≥ 0.7.3. Verbatim, unparsed. Replay in `If-Modified-Since` | +| `url` | `str` | Echoed. Confirmation and debugging only | +| `command_id` | `str` | **Required.** Echoed from the command; half the envelope key | +| `info_source_id` | `str` | **Required.** Echoed verbatim. Replicator neither parses nor interprets it | +| `final_url` | `str \| None` | Where the fetch **landed** after redirects. `None` = *unknown*, see below | +| `status_code` | `int \| None` | **Always 2xx on this fact** — see below | +| `fetched_at` | `datetime \| None` | tz-aware UTC. When the bytes were on the **wire**, not when the fact was published | +| `content_type_raw` | `str \| None` | The **verbatim** `Content-Type`, `charset` and all. `None` = *the origin sent none*, see below | +| `etag` | `str \| None` | Verbatim, `W/` prefix and quotes included. Replay unparsed in `If-None-Match` | +| `last_modified` | `str \| None` | Verbatim, unparsed. Replay in `If-Modified-Since` | +| `blob_expires_at` | `datetime \| None` | **Populated since #28.** When the blob stops being retrievable at `blob_uri`. Prefer it to re-deriving MUST-7's TTL — see below | The six enriched fields (cannobserv#271, `final_url` sourced by cannobserv#279, produced by #10) carry what Replicator holds at publish time and a broadcast consumer cannot recover once fetching -lives here rather than in Watcher. Three rules govern reading them, and one warning governs acting -on them — all four in -[`content-fetch-issuer-reference.md`](content-fetch-issuer-reference.md#the-enriched-blob_available-fields): - -- **`None` means nobody said, never "the default"**, on `final_url` and `content_type_raw` alike. -- **`status_code` is always 2xx here** — it distinguishes 200 from 203, it is not a success branch. -- **"Verbatim" excludes surrounding whitespace**, and an over-long value is dropped, not truncated. -- **Do not attempt conditional GET yet.** Replicator honours the `headers` you send, so a validator - *will* reach the origin — but a matching one earns a body-less 304, which Replicator still closes - as a terminal `fetch_failed` · `http_status`. Tracked as **#17**. Until it lands, keep `etag` and - `last_modified` in your own records and send the request unconditionally. +lives here rather than in Watcher. Three rules govern reading them — **`None` means nobody said, +never "the default"**; **`status_code` is always 2xx here**, so it is not a success branch; +**"verbatim" excludes surrounding whitespace**, and an over-long value is dropped rather than +truncated — and one warning governs acting on them: **do not attempt conditional GET yet**, because +a validator that matches earns a body-less 304 that Replicator still closes as a terminal +`fetch_failed` · `http_status` (**#17**). All four, with the reasoning, in +[the reference](content-fetch-issuer-reference.md#the-enriched-blob_available-fields). ### The failure fact -**Fact — `content.blobs`, `FetchFailedEvent`** (co-core ≥ 0.7.2, cannobserv#270): +**Fact — `content.blobs`, `FetchFailedEvent`** (cannobserv#270): | Field | Type | Notes | |---|---|---| @@ -168,33 +169,24 @@ on them — all four in | `occurred_at` | `datetime` | tz-aware UTC, stamped at publish. Also half the envelope key | | `command_id` | `str` | **Required — the correlator, and the whole point of the event** | | `url` | `str` | Echoed. Confirmation and debugging only, exactly as on the success fact | +| `info_source_id` | `str` | **Required.** Echoed verbatim, exactly as on the success fact | | `reason` | `str` | Stable token; see below. **Treat an unknown value as opaque** | | `terminal` | `bool` | `True` = the command is closed, no blob will ever arrive. **Branch on this first** | | `status_code` | `int \| None` | Set for `http_status`; absent otherwise | | `attempts` | `int \| None` | Set only where the attempt count is *why* the command closed | -| `detail` | `str \| None` | Free text for the journal. **Never branch on it**, and never surface it to an end user — on `handler_error` it is the text of an exception Replicator did not anticipate, so its content is unbounded | - -`reason` tokens, one per row of the -[failure taxonomy](content-fetch-issuer-reference.md#failure-taxonomy-what-happens-and-what-the-issuer-sees): - -| Token | Condition | -|---|---| -| `http_status` | 4xx, or a body-less 304 (`status_code` set) | -| `not_fetchable` | bad scheme / invalid URL | -| `too_large` | body over `REPLICATOR_MAX_BLOB_BYTES` | -| `unsupported_schema_version` | the command decoded at a `schema_version` Replicator does not support | -| `invalid_request_options` | `headers` or `timeout_seconds` are not sendable — see the [refusal list](content-fetch-issuer-reference.md#request-options-what-replicator-will-send-and-what-it-refuses-11) | -| `handler_error` | unclassified, and it exhausted the delivery ceiling (`attempts` set) | - -`reason` is a plain `str`, not a `Literal`, so the token list is additive and a consumer that -branches on `terminal` first is already correct for tokens that do not exist yet. Replicator never -emits co-core's `wrong_payload_type`, and everything it emits today is `terminal=True` — both -deliberate, both explained in +| `detail` | `str \| None` | Free text for the journal. **Never branch on it**, and never show it to an end user — on `handler_error` it is an unanticipated exception's text, so unbounded | + +The tokens emitted today are `http_status`, `not_fetchable`, `too_large`, +`unsupported_schema_version`, `invalid_request_options` and `handler_error` — one per row of the +[failure taxonomy](content-fetch-issuer-reference.md#failure-taxonomy-what-happens-and-what-the-issuer-sees), +which states each one's condition. `reason` is a plain `str`, not a `Literal`, so the list is +additive and a consumer branching on `terminal` first is already correct for tokens that do not +exist yet. Replicator never emits co-core's `wrong_payload_type`, and everything it emits today is +`terminal=True` — both deliberate, both in [the reference](content-fetch-issuer-reference.md#reading-the-failure-fact). -All three models are `extra="ignore"`: additive producer fields are tolerated. Branch on -`schema_version` **before** destructuring, and never use the strict `*Emit` classes on a consume -path. +All three models are `extra="ignore"`, so additive producer fields are tolerated. Branch on +`schema_version` **before** destructuring, and never use the strict `*Emit` classes on a consume path. --- @@ -205,21 +197,18 @@ path. Replicator dedupes on `command_id` — `replicator:cmd:`, TTL `REPLICATOR_DEDUPE_TTL_SECONDS` (default 24 h, operator-tunable) — in [`src/worker/loop.py::process_message`](../../src/worker/loop.py). A duplicate is **acked and -dropped**: no fetch, no fact, one `INFO` line on Replicator's side and nothing at all on the -issuer's. +dropped**: no fetch, no fact, one `INFO` line here and nothing at all on the issuer's side. So a `command_id` derived from anything resource-stable — a WatchedItem id, a hash of the URL, the -URL itself — means **the second legitimate re-fetch of that URL never happens**. Watcher, whose -entire job is re-fetching a URL over time to detect change, is precisely the service this breaks. +URL itself — means **the second legitimate re-fetch of that URL never happens**. Watcher, whose job +is re-fetching a URL over time to detect change, is precisely the service this breaks. The trap is worse than a flat failure because it is **TTL-bounded**: re-fetches inside the dedupe -window vanish, re-fetches after it work. A daily cadence sits on the boundary of the default 24 h -and fails intermittently. A test run with two fetches a minute apart reproduces it; a test run with -two fetches a day apart does not. +window vanish, re-fetches after it work. A daily cadence sits on the boundary of the default 24 h and +fails intermittently — two fetches a minute apart reproduce it, two a day apart do not. Mint a ULID per fetch intent — one per call, not one per URL and not one per run. [`scripts/seed_fetch.py::build_command`](../../scripts/seed_fetch.py) does exactly this and says why. - Uniqueness is required *for correctness* only within the dedupe TTL, but *for correlation* it must be global and permanent — the issuer's own map is keyed on it. @@ -230,114 +219,111 @@ can be matched against. ### 2. Persist `command_id → domain` durably, **before** publishing -The bus carries nothing that can reconstruct the mapping. If the issuer crashes between minting the -id and recording it, the returning `blob_available` is uncorrelatable — there is no query, on any -stream, that recovers which InfoSource asked. +Persist first, then publish; outbox-style, symmetric to archiver's producer. (Replicator has no +outbox and must not grow one — its durable record of intent is the consumer group's PEL, and the +outbox belongs to producers with a database.) -Persist first, then publish; outbox-style, symmetric to archiver's producer. (Replicator itself has -no outbox and must not grow one — its durable record of intent is the consumer group's PEL. The -outbox belongs to producers with a database, which is the issuer.) +**Narrowed by cannobserv#300, and not retroactively.** From the step-2 deploy onward every fact +carries `info_source_id`, so a crash between minting an id and recording it costs the link to the +*occasion*, not the domain object. Before that deploy, and for any issuer still on 0.7.x, the +original statement holds: nothing on any stream recovers which InfoSource asked. The record stays +required on narrower grounds — request options, audit, MUST-6's reaper — but it is no longer the +only thing between a fetch and an orphaned blob. ### 3. Correlate on `command_id` only — `url` is not a key Archiver's model permits **multiple InfoSources per URL** (non-unique `url`, different extraction -strategies), so `url → info_source` is one-to-many. An issuer that falls back to matching a fact by -its `url` will, sooner or later, attach bytes to the wrong InfoSource — silently, and in a way that -looks like correct data downstream. +strategies), so `url → info_source` is one-to-many. An issuer matching a fact by its `url` will +sooner or later attach bytes to the wrong InfoSource — silently, and in a way that looks like +correct data downstream. `url` on the fact is confirmation and debugging, nothing more. -`url` on the fact is confirmation and debugging. Nothing more. +Since cannobserv#300 there is no *reason* to reach for it either: `info_source_id` is on both +outcomes, so an issuer that lost its `command_id` map can still tell which InfoSource a fact is +about, without the one-to-many guess that makes `url` wrong. ### 4. Make correlation idempotent — one command can yield more than one fact `blob_available` is at-least-once **per `command_id`**, not exactly-once. The dedupe key is written *after* the handler returns, deliberately -([`src/worker/loop.py::process_message`](../../src/worker/loop.py)): -marking first would turn a crash between the mark and a completed handle into permanent loss. The -cost of that ordering is the opposite duplicate — a crash between a successful publish and the -`SET` re-runs the handler on reclaim and emits a **second fact carrying the same `command_id`**. +([`src/worker/loop.py::process_message`](../../src/worker/loop.py)): marking first would turn a +crash between the mark and a completed handle into permanent loss. The cost is the opposite +duplicate — a crash between a successful publish and the `SET` re-runs the handler on reclaim and +emits a **second fact carrying the same `command_id`**. Applying a fact must therefore be safe to do twice. Same `command_id`, same `content_fingerprint`, same `blob_uri` — an upsert, not an append. **This covers `fetch_failed` too, and there the duplicates are not even identical** — a second -failure fact under the same `command_id` with a fresh `occurred_at`, so its envelope key differs -and consumer-side dedup-on-key will not collapse it. Why, and why Replicator cannot engineer it -away, is in [the reference](content-fetch-issuer-reference.md#why-a-duplicate-failure-fact-is-not-identical). - -Closing a pending entry must therefore be idempotent in both directions: applying the same -terminal failure twice, and applying a failure to an entry already closed. +failure fact under the same `command_id` with a fresh `occurred_at`, so its envelope key differs and +consumer-side dedup-on-key will not collapse it ([why, and why Replicator cannot engineer it +away](content-fetch-issuer-reference.md#why-a-duplicate-failure-fact-is-not-identical)). Closing a +pending entry must therefore be idempotent in both directions: the same terminal failure applied +twice, and a failure applied to an entry already closed. ### 5. Do **not** dedupe facts on `content_fingerprint` `content_fingerprint` is the fact's idempotency key *for storage*: identical bytes are the same -blob, at the same path, and re-storing them is a no-op. It is **not** an idempotency key for -correlation. +blob at the same path, and re-storing them is a no-op. It is **not** one for correlation. -Two commands — two occasions of the same URL, or two InfoSources sharing one URL — that return -identical bytes produce two facts with **the same fingerprint and the same `blob_uri`, and -different `command_id`s**. A consumer deduping its inbox on fingerprint drops the second fact and -loses that correlation entirely: the same silent-failure shape as MUST-1, arrived at from the -opposite direction. +Two commands — two occasions of one URL, or two InfoSources sharing it — that return identical +bytes produce two facts with **the same fingerprint and `blob_uri`, and different `command_id`s**. A +consumer deduping its inbox on fingerprint drops the second and loses that correlation entirely: the +same silent-failure shape as MUST-1, reached from the opposite direction. -Dedupe on `command_id`. Treat the fingerprint as content identity. +Dedupe on `command_id`. Treat the fingerprint as content identity. cannobserv#300 closed the +producer half — the envelope key names the occurrence now — but the consumer rule is unchanged. -And do not dedupe **`fetch_failed`** on `command_id` either — for the opposite reason. More than -one failure fact per command is expected (MUST-4), and once non-terminal facts exist a command -may legitimately emit several. Use `terminal` to decide whether the entry closes; use -`command_id` to decide *which* entry. +And do not dedupe **`fetch_failed`** on `command_id` either, for the opposite reason: more than one +failure fact per command is expected (MUST-4), and once non-terminal facts exist a command may +legitimately emit several. Use `terminal` to decide whether the entry closes, `command_id` to decide +*which* entry. ### 6. Handle `fetch_failed`, and **keep a reaper anyway** -A command that fails permanently now publishes a **`fetch_failed` fact on `content.blobs`** -(#9, co-core cannobserv#270) *and* is copied to `content.fetch.dlq` and acked. The fact is the -issuer's surface; the DLQ is the operator's. This added a signal — it did not replace one, and an -entry appears in both places. - -So an issuer's primary mechanism is now the fact: consume `content.blobs`, branch on the payload -type, and on a `fetch_failed` with `terminal=True` close the pending entry **with a reason**. Off -one consumer group, since both outcomes share the stream. - -**Silence has not gone away — it has narrowed.** Four conditions still produce nothing: a frame -that fails `from_wire` entirely, a frame that decodes to a payload that is not a `content.fetch` -command, a command whose `command_id` is blank, and a command that is still retrying. The first -three are permanently silent — no payload, or no `command_id` that is *safe* to key a fact on. The -fourth is silent *for now* (#9 §3), and has no latency bound: transient failures retry indefinitely -at the `REPLICATOR_CLAIM_MIN_IDLE_MS` cadence, and a blob tree over -`REPLICATOR_BLOB_MAX_TOTAL_BYTES` parks in the PEL until a sweep frees space. Each condition, and -why reporting the second would be worse than silence, is in +A command that fails permanently publishes a **`fetch_failed` fact on `content.blobs`** (#9, +cannobserv#270) *and* is copied to `content.fetch.dlq` and acked. The fact is the issuer's surface, +the DLQ the operator's; an entry appears in both. So the issuer's primary mechanism is the fact: +consume `content.blobs`, branch on the payload type, and on a `fetch_failed` with `terminal=True` +close the pending entry **with a reason** — off one consumer group, since both outcomes share the +stream. + +**Silence has not gone away — it has narrowed.** Four conditions still produce nothing. Three are +permanently silent, having no payload or no `command_id` *safe* to key a fact on; the fourth, a +command still retrying, is silent *for now* (#9 §3) with **no latency bound** — transient failures +retry at the `REPLICATOR_CLAIM_MIN_IDLE_MS` cadence indefinitely, and a tree over +`REPLICATOR_BLOB_MAX_TOTAL_BYTES` parks in the PEL until a sweep frees space. All four, and why +reporting one would be worse than silence, are in [the reference](content-fetch-issuer-reference.md#the-four-silent-conditions). -Which is why the **reaper stays**, demoted from primary mechanism to backstop: +Which is why the **reaper stays**, demoted from primary mechanism to backstop. A timeout is grounds +to **re-issue** (fresh `command_id`), not to conclude failure; derive it generously from the reclaim +cadence rather than hardcoding a number, since the cadence is an operator setting on Replicator's +host and an issuer that pins 60 s starts re-issuing under a live retry the day it is tuned. Expect +duplicate work rather than assuming loss. -- A timeout is still grounds to **re-issue** (fresh `command_id`), not to conclude failure. -- Derive the timeout generously from the reclaim cadence rather than hardcoding a number — the - cadence is an operator setting on Replicator's host, and an issuer that pins 60 s starts - re-issuing under a live retry the day it is tuned. Expect duplicate work rather than assuming - loss. - -See the [failure taxonomy](content-fetch-issuer-reference.md#failure-taxonomy-what-happens-and-what-the-issuer-sees) for exactly -which outcomes are visible and which are not. - -**The DLQ is still readable, and still worth reading** — it is the only place the silent rows show -up at all, and it preserves the offending frame, which no fact does. How to read it, and the one -anomaly class that carries no `key` to match on, are in -[the reference](content-fetch-issuer-reference.md#reading-the-dlq). +**The DLQ is still worth reading** — the only place the silent rows appear at all, and it preserves +the offending frame, which no fact does. How to read it, and which outcomes are visible, are in the +reference's [DLQ](content-fetch-issuer-reference.md#reading-the-dlq) and +[failure taxonomy](content-fetch-issuer-reference.md#failure-taxonomy-what-happens-and-what-the-issuer-sees). ### 7. Copy the bytes before the blob expires `blob_uri` is temporary. A blob is reaped once its mtime is older than -`REPLICATOR_BLOB_TTL_SECONDS` — currently 7 days, a published commitment to archiver -(archiver#118) rather than a knob Replicator turns freely, but still a *setting* on Replicator's -host. Treat it as a floor to ask about, not a constant to schedule against: consume the bytes -promptly and re-issue if the URI fails to open, rather than building a pipeline whose timing -assumes seven days. +`REPLICATOR_BLOB_TTL_SECONDS` — currently 7 days, a published commitment to archiver (archiver#118) +rather than a knob Replicator turns freely, but still a *setting* on its host. Treat it as a floor +to ask about, not a constant to schedule against: consume the bytes promptly and re-issue if the URI +fails to open, rather than building a pipeline whose timing assumes seven days. The clock runs from **last reference by a fetch**, not last read by a consumer — holding a -`blob_uri` for a week without a re-fetch loses the bytes. The mechanism is in -[the reference](content-fetch-issuer-reference.md#how-the-blob-ttl-clock-runs). +`blob_uri` for a week without a re-fetch loses the bytes. So **record `blob_expires_at` rather than +re-deriving a horizon** (cannobserv#301, carried since #28): deriving one hard-codes a policy +Replicator owns and starts the clock where no consumer can see it. The published value can only fall +**earlier** than the real reap, so acting on it is early, never too late; `None` means unknown, and +is recorded as absence rather than guessed. +[Mechanism](content-fetch-issuer-reference.md#how-the-blob-ttl-clock-runs). -Also: `blob_uri` is a **`file://` URI on Replicator's host**. The contract is VM-local today. -A consumer on another host cannot open it, and nothing on the wire says so. +Also: `blob_uri` is a **`file://` URI on Replicator's host** — the contract is VM-local today, a +consumer elsewhere cannot open it, and nothing on the wire says so. --- @@ -349,10 +335,10 @@ A consumer on another host cannot open it, and nothing on the wire says so. - **Announce, then ack.** Every path that closes a command without a blob publishes its `fetch_failed` *before* the dead-letter that acks it, so a crash in between costs a duplicate fact (MUST-4) rather than losing the fact outright. -- **`command_id` is echoed** on every fact produced from a command, success or failure. -- **The fingerprint is definitional.** Replicator is the cluster's sole fetcher and sole - fingerprinter, so `content_fingerprint` *is* the content identity — there is nothing to - cross-check it against. +- **`command_id` and `info_source_id` are echoed** on every fact, success or failure. Both are + required on both outcomes, and `info_source_id` is copied verbatim. +- **The fingerprint is definitional.** Replicator is the cluster's sole fetcher and fingerprinter, + so `content_fingerprint` *is* the content identity — there is nothing to cross-check it against. - **Identical bytes are one blob.** Content-addressed storage, so a re-fetch of unchanged content costs an origin request and nothing else. - **At-least-once delivery**, both directions. @@ -362,37 +348,31 @@ A consumer on another host cannot open it, and nothing on the wire says so. - **No *non-terminal* failure fact** — a command that is retrying announces nothing until it either succeeds or is closed. See MUST-6 and [the failure-fact notes](content-fetch-issuer-reference.md#reading-the-failure-fact). -- **No failure fact for a frame that is not a `content.fetch` command** — any `command_id` it - carries is another command's, so there is nothing *safe* to correlate one on. MUST-6. -- **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 failure fact where there is nothing safe to key one on** — a frame that is not a + `content.fetch` command (any `command_id` in it is another command's) or one whose `command_id` is + blank. Both are dead-lettered 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. - Whether a paced command is slept through or parked for the next reclaim depends on the deployed - interval — see [the reference](content-fetch-issuer-reference.md#pacing-at-the-deployed-defaults), - since it changes what a reaper should expect. + spaced by at least `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` (1 s by default), so 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; different hosts are unaffected by each other. Whether a + paced command is slept through or parked for the next reclaim depends on the deployed interval, + and it changes what a reaper should expect — + [the reference](content-fetch-issuer-reference.md#pacing-at-the-deployed-defaults). - **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. - **No retention guarantee on `content.blobs`.** Replicator never trims it — `BusPublish` takes no - `MAXLEN` and nothing in this repo issues `XTRIM` — so whatever policy applies is the broker - operator's, not part of this contract. Do not treat the stream as an archive to reconcile - against later. + `MAXLEN` and nothing here issues `XTRIM` — so whatever policy applies is the broker operator's, + not part of this contract. It is not an archive to reconcile against later. --- ## Where the rest of the contract lives -Split out in #24 so this document stays readable start to finish. Equally normative: - -- [`content-fetch-issuer-reference.md`](content-fetch-issuer-reference.md) — everything this file - points at: the refusal list, the enriched fields, the **failure taxonomy**, the silent - conditions, the DLQ, and **provenance and trust**. +- [`content-fetch-issuer-reference.md`](content-fetch-issuer-reference.md) — **equally normative**, + and everything this file points at: the refusal list, the enriched fields, the failure taxonomy, + the silent conditions, the DLQ, provenance and trust, and the version history. - [`replicator-boundaries.md`](replicator-boundaries.md) — which payload fields this contract will never grow (#12). - [`2026-07-31-fetch-failed-fact-settled.md`](../plans/2026-07-31-fetch-failed-fact-settled.md) — diff --git a/docs/contracts/content-fetch-issuer-reference.md b/docs/contracts/content-fetch-issuer-reference.md index 3404484..a09339d 100644 --- a/docs/contracts/content-fetch-issuer-reference.md +++ b/docs/contracts/content-fetch-issuer-reference.md @@ -72,7 +72,10 @@ shaped the way they are, and the one thing they do not yet let an issuer do. The six enriched fields (cannobserv#271, `final_url` sourced by cannobserv#279, produced by #10) carry what Replicator holds at publish time and a broadcast consumer cannot recover once fetching -lives here rather than in Watcher. Three details are the whole value of them: +lives here rather than in Watcher. `blob_expires_at` (cannobserv#301, populated by #28) is a +seventh, and describes the **store** rather than the fetch — see +[MUST-7](content-fetch-issuer-contract.md#7-copy-the-bytes-before-the-blob-expires). Three details +are the whole value of the six: - **`None` means nobody said, never "the default".** `final_url` is `None` when the *driver* did not report a landing URL — **not** "no redirect occurred", and Replicator never substitutes the @@ -97,13 +100,14 @@ lives here rather than in Watcher. Three details are the whole value of them: and a *truncated* ETag replayed in an `If-None-Match` is a validator that can never match, which is worse than none. -> **These are per-*occasion* values on a fingerprint-keyed fact.** They describe the fetch that -> produced this fact, not the bytes — which is why MUST-5 matters more now than it did. Two -> commands returning identical bytes emit two facts with the same `content_fingerprint` and -> possibly *different* `final_url`, `etag`, `last_modified`, and `fetched_at`. A consumer deduping -> its inbox on the fingerprint — already forbidden — now also pins its stored validators to the -> first emission for those bytes, and will replay a stale `If-None-Match` for as long as that -> content is unchanged. +> **These are per-*occasion* values, and since cannobserv#300 the fact is keyed per occasion too.** +> They describe the fetch that produced this fact, not the bytes. Through 0.7.7 the envelope key was +> the bare `content_fingerprint`, so two commands returning identical bytes collapsed to one fact and +> its `final_url` / `etag` / `last_modified` / `fetched_at` were the *first* emission's — a consumer +> replayed a stale `If-None-Match` for as long as that content stayed unchanged. **That caveat is +> dissolved**: every fetch now emits its own fact carrying its own validators. Recorded because a +> consumer written against 0.7.x may still carry a workaround for it. MUST-5 is unaffected — deduping +> an inbox on the fingerprint is still wrong, and still loses a correlation. > **Do not attempt conditional GET yet — this is now the *only* thing standing in the way.** > `etag` and `last_modified` are the *read* half of the seam, and since #11 the write half exists: @@ -130,7 +134,7 @@ lives here rather than in Watcher. Three details are the whole value of them: | `schema_version` ≠ 1 | fact, then `content.fetch.dlq` | `fetch_failed` · `unsupported_schema_version` | | Frame decodes to a non-`content_fetch` payload | `content.fetch.dlq` | **nothing** — any `command_id` in it is another command's | | Command with a blank `command_id` | `content.fetch.dlq`, before the fetch | **nothing** — no correlator to key a fact on | -| Malformed frame (fails `from_wire`; includes a naive `occurred_at`) | `content.fetch.dlq`, synthesized record | **nothing** — no payload at all | +| Malformed frame (fails `from_wire`; includes a naive `occurred_at`, and any 0.7.x command with no `info_source_id`) | `content.fetch.dlq`, synthesized record | **nothing** — no payload at all | | HTTP 4xx, or a body-less 304 | fact, then `content.fetch.dlq` | `fetch_failed` · `http_status` (+ `status_code`) | | URL not fetchable (bad scheme / invalid URL) | fact, then `content.fetch.dlq` | `fetch_failed` · `not_fetchable` | | Body over `REPLICATOR_MAX_BLOB_BYTES` (default 64 MiB) | fact, then `content.fetch.dlq` | `fetch_failed` · `too_large` | @@ -266,7 +270,10 @@ fetched for real. Contracts settled in cannobserv#266 (co-core v0.7.0); the failure fact added in cannobserv#270 and the tz-aware `occurred_at` in cannobserv#273, both shipped in **co-core v0.7.2**; the enriched `blob_available` metadata in cannobserv#271/#279 and the command's request options in -cannobserv#272, shipped in **v0.7.3** and **v0.7.5**. Replicator requires **co-core ≥ 0.7.5**. +cannobserv#272, shipped in **v0.7.3** and **v0.7.5**. **v0.8.0** required `info_source_id` on all +three payloads and `command_id` on `blob_available`, re-keyed `blob_available` to +`content_fingerprint:command_id` (cannobserv#300), and added `blob_expires_at` (cannobserv#301). +Replicator requires **co-core ≥ 0.8.0**. Founding rationale: [`docs/plans/2026-06-25-replicator-mvp-design.md`](../plans/2026-06-25-replicator-mvp-design.md). --- @@ -319,6 +326,13 @@ the file when a re-fetch short-circuits on existing bytes ([`src/storage/local.py::_touch`](../../src/storage/local.py)), but a consumer opening the path does not touch mtime. Holding a `blob_uri` for a week without a re-fetch loses the bytes. +That is the event `blob_expires_at` exposes, since no consumer can see it. The value is +`stored_at + REPLICATOR_BLOB_TTL_SECONDS`, read **before** the store, so it lands at or before the +mtime the sweep measures against — and the sweep runs only every +`REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS`, putting the real reap later still. The error is one-way: +acting on the horizon is early, never too late. A later fetch pushes the expiry out and emits a +fresh fact carrying the new value. + Also, on the reaper [MUST-6](content-fetch-issuer-contract.md#6-handle-fetch_failed-and-keep-a-reaper-anyway) keeps: diff --git a/docs/contracts/replicator-boundaries.md b/docs/contracts/replicator-boundaries.md index 4d1343f..54e93a1 100644 --- a/docs/contracts/replicator-boundaries.md +++ b/docs/contracts/replicator-boundaries.md @@ -35,17 +35,18 @@ remediation. 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. +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. + (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 Replicator has to *read* the words + InfoSource, InfoItem, WatchedItem, aspect, tenant → **issuer**, always. Carrying one unread is + the one exception — see **Reviewing a proposed payload field**. -Tests 1 and 2 can both fire. When they do: +Tests 1 and 2 can both fire: > **Mechanism to Replicator. Policy to the issuer. Config travels over the bus.** @@ -233,23 +234,20 @@ than fetching unpaced on the way out. `src/worker/pacing.py`, `handler.py::_pace ### The fallback default (was the interim, #12 → #19) -`REPLICATOR_MIN_HOST_INTERVAL_SECONDS`, default **1.0 s** — Watcher's own -`DEFAULT_MIN_INTERVAL`, chosen precisely because it invents nothing. #12 shipped it as one -number for every origin, standing in for a stream that did not exist. #19 gave it its permanent -job: it is what a host with **no explicit policy** resolves to — unknown, revoked, or not yet -replayed — and never "unlimited", because a boot replay cannot tell a consumer whether the set -it received is whole. +`REPLICATOR_MIN_HOST_INTERVAL_SECONDS`, default **1.0 s** — Watcher's own `DEFAULT_MIN_INTERVAL`, +chosen precisely because it invents nothing. #12 shipped it as one number for every origin, standing +in for a stream that did not exist; #19 gave it its permanent job as what a host with **no explicit +policy** resolves to — unknown, revoked, or not yet replayed — and never "unlimited", because a boot +replay cannot tell a consumer whether the set it received is whole. -**`0` no longer disables pacing outright.** It is the fallback for unpublished hosts only; a -host with a policy is still paced by it. The alternative would let a local env var veto a value -the issuer published, which inverts the ownership split this whole document settles. An -operator who wants no politeness at all now has to say so per host, through the producer that -owns the numbers. +**`0` no longer disables pacing outright.** It is the fallback for unpublished hosts only; a host +with a policy is still paced by it. The alternative would let a local env var veto a value the +issuer published, inverting the ownership split this document settles. An operator wanting no +politeness at all says so per host, through the producer that owns the numbers. 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 fallback number is not policy in the -sense test 3 cares about — it names no domain concept, and the per-host table it defers to is -the producer's. **Two in-memory maps** now, with different bounding rules and deliberately so: +fetcher sees a host's tolerance across commands), and a fallback number is not policy in the sense +test 3 cares about — it names no domain concept, and the table it defers to is the producer's. **Two in-memory maps** now, with different bounding rules and deliberately so: the pacer's host → last-request map is consumer-derived and pruned to `MAX_TRACKED_HOSTS`, while the policy map is bounded by what the producer publishes and is **never pruned** — dropping an entry to honour a local limit would silently loosen a host's spacing, the exact @@ -257,32 +255,24 @@ failure the stream exists to remove. Both are derived, rebuildable by replay, an domain vocabulary: the second of the three permitted state shapes, twice. **Boot ordering is part of the design, not an implementation detail.** `replay()` runs -synchronously before the consume loop starts. Started as a peer task, the worker would fetch -its opening commands against an empty map and pace every host at the fallback — safe only -because the fallback is meant to be the stricter number, and that is the one assumption not -worth spending on startup ordering. A failed replay is absorbed rather than fatal: the cursor -advances only over messages that decoded, so the tail resumes from the same place and drains -the rest, while failing the boot would turn a policy-stream hiccup into a total fetch outage. -The rebuilt host count is logged, because an empty map and a working one are otherwise -indistinguishable from outside. +synchronously before the consume loop starts, or the worker's opening commands are paced against +an empty map — safe only because the fallback is the stricter number, and that is the one +assumption not worth spending on startup ordering. Mechanism, including why a failed replay is +absorbed rather than fatal: [ARCHITECTURE.md](../ARCHITECTURE.md). **Known limitation, still open after #19: 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", -which wants its own decision. #19 did **not** resolve it: per-host numbers make the fix more -defensible — the landing host would be paced under its own published policy rather than under a -guess — without making it automatic. Tracked as its own gap rather than left as a promissory -note against a stream that has now shipped, and 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. #12's default closed that window on the consumer side and #19 +reached.** httpx follows redirects inside the driver, so a URL that 301s elsewhere is paced under +the name the command carried, never under the name that served it — and a corpus funnelling into +one portal or CDN hits it at N times the intended rate, the failure politeness exists to prevent. +The fix would read `FetchResult.final_url`, but recording the landing host breaks "one request, one +record" and wants its own decision. #19 made it more defensible without making it automatic. +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 publication rather than origin requests. #12's default closed that window on the consumer side and #19 supplies the numbers; **what remains is issuer-side** — Watcher publishing its `Domain` rows onto this stream, tracked at [CannObserv/watcher#245](https://github.com/CannObserv/watcher/issues/245). Until it does, @@ -292,16 +282,22 @@ half could land first. ## 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. +Replicator. A field Replicator would have to *read* to do its job fails. The wire's +domain-agnosticism is the property the whole issuer contract is built on; it erodes one plausible +field at a time. + +**`info_source_id` is the settled exception, and its shape is the precedent (cannobserv#300, +#28).** What made it acceptable is not that the field is small — it is that Replicator **never reads +it**: delete every line mentioning it and the byte path behaves identically. So the real question is +*does Replicator have to understand this to act on it?* If yes it fails whatever it is called; if no +it is freight. **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. +[`src/core/errors.py::FailureReason`](../../src/core/errors.py) is a local `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 consumer, which +puts the vocabulary on the producer. Defining a wire *model* here would be the violation; owning the +tokens Replicator emits is the contract working. ## Known violation, tracked @@ -335,7 +331,8 @@ 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 | +| No domain vocabulary | AST scan of `src/`: `info_source`, `info_item`, `watched_item`, `tenant`, `aspect` appear in no identifier and no string literal — exact `info_source_id` exempted in the three emit-path modules only | +| The echoed key is never interpreted | AST scan of `src/`: every `info_source_id` occurrence is a field declaration, a parameter, the `info_source_id=` keyword, or that keyword's value; all else fails | | 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 | @@ -348,20 +345,27 @@ 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. +matched `both tasks watch one stop event` in a docstring, and a test whose first tripper is an +English sentence 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 because domain +leakage arrives as a dict key or log field as often as an attribute. + +**The `info_source` exemption is a carve-out, and a second scan is what makes it one.** The wire +requires naming the field to copy it, so it is allowed in exactly `handler.py`, `reporter.py` and +`loop.py`, and only as the exact identifier — no `info_source_policy` map rides in behind it. +The second scan holds the real invariant as an **allow-list**: the four shapes a verbatim echo can +take, everything else refused. It began as a deny-list of reading positions; three review rounds each +found ones it had not enumerated, so it now asks the opposite question and is exhaustive. Naming the field is mechanics; keying on it is a domain model one commit at a time. +Two assertions guard the allowlist itself: it can never name config, storage or the API, and every +entry must be a file. + +**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 runs 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 — worse than no test, because this document cites it. ## Refs diff --git a/pyproject.toml b/pyproject.toml index 952ba3d..1f4f4a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,13 +7,14 @@ license = "MIT" license-files = ["LICENSE"] requires-python = ">=3.12" dependencies = [ - # Floor is 0.7.5, not 0.7: the enriched BlobAvailableEvent fields landed in - # 0.7.3 and FetchResult.final_url in 0.7.5, and the event model is - # extra="ignore" — on an older wheel the enriched kwargs construct cleanly - # and are silently discarded. Without the floor a version skew publishes - # facts that look right and carry nothing (#10, cannobserv#271/#279). - "co-core[extract]>=0.7.7,<0.8", - "co-core-aio[bus]>=0.7.7,<0.8", + # Floor is 0.8.0, and this one fails *loudly* rather than silently. The + # earlier floors (0.7.5's enriched fields, 0.7.7's AsyncBusTailReader) guarded + # against extra="ignore" quietly discarding kwargs an older wheel had no field + # for; 0.8.0 makes info_source_id **required** on ContentFetchCommand, + # BlobAvailableEvent and FetchFailedEvent, so a skew is a ValidationError at + # construction, not a fact that looks right and carries nothing (#28). + "co-core[extract]>=0.8.0,<0.9", + "co-core-aio[bus]>=0.8.0,<0.9", "fastapi>=0.126.0,<1", "pydantic>=2.9,<3", "pydantic-settings>=2.5.0,<3", diff --git a/scripts/seed_fetch.py b/scripts/seed_fetch.py index 60485f9..fdb47ff 100644 --- a/scripts/seed_fetch.py +++ b/scripts/seed_fetch.py @@ -21,6 +21,13 @@ requires ``--production``. A flag rather than a prompt: the script has to stay usable non-interactively. +Every command carries an ``info_source_id``, required on the wire since co-core +0.8.0 and echoed onto both facts (#28). It defaults to a placeholder no issuer's +InfoSource table contains, so the facts a scratch run produces are recognizably +synthetic — and for that same reason the live target refuses both that default +and a blank, since a real fetch broadcasts whatever is passed to the cluster. +Name one with ``--info-source-id``. + ``--watch`` tails the fact stream so a human can see the loop close without hand-writing ``XRANGE``. It accepts **either** outcome — ``blob_available`` or, since #9, ``fetch_failed`` — because a watch that recognized only success would @@ -34,10 +41,11 @@ Exit codes: ``0`` published (and, under ``--watch``, every command produced a blob) · ``1`` the run did not complete — publishing failed, watching failed, a command was closed by a ``fetch_failed``, or no fact ever arrived · ``2`` the -target was refused. Commands are reported on stdout as they land, so a non-zero -exit never hides a command it saw land — a connection lost between the ``XADD`` -and its reply is the one gap, and the "N of M" count on stderr is what marks that -boundary as fuzzy. +target was refused — a missing ``--production`` opt-in, or a placeholder or blank +``info_source_id`` alongside one. Commands are reported on stdout as they land, +so a non-zero exit never hides a command it saw land; a connection lost between +the ``XADD`` and its reply is the one gap, and the "N of M" count on stderr is +what marks that boundary as fuzzy. """ import argparse @@ -70,13 +78,25 @@ DEFAULT_WATCH_TIMEOUT_SECONDS = 30.0 +# The domain key every seeded command carries unless the operator names one +# (#28). Deliberately not id-shaped: no issuer's InfoSource table can contain +# this, so a fact that reaches a real consumer is recognizably from the harness +# rather than plausibly from a real fetch. +SEED_INFO_SOURCE_ID = "seed-harness-not-a-real-info-source" + # Either outcome of a command. ``content.blobs`` has carried both since #9, so # "the fact for this command_id" is no longer synonymous with "the blob". Fact = BlobAvailableEvent | FetchFailedEvent class ProductionTargetError(RuntimeError): - """The requested target is the live command stream and no opt-in was given.""" + """The requested target is the live command stream and something was left implicit. + + Either the ``--production`` opt-in is missing, or it was given while + ``info_source_id`` was left at the placeholder or blank — one exit code, + because all of them mean the same thing: a real fetch was about to happen on + an assumption. + """ @dataclass(frozen=True) @@ -92,6 +112,7 @@ def build_command( url: str, headers: dict[str, str] | None = None, timeout_seconds: float | None = None, + info_source_id: str = SEED_INFO_SOURCE_ID, ) -> ContentFetchCommand: """Mint a command for one URL. @@ -106,6 +127,13 @@ def build_command( ``headers`` / ``timeout_seconds`` apply to every URL in the run and default to ``None`` — the omitted-field shape, which is the worker's pre-#11 behaviour exactly. + + ``info_source_id`` is required by co-core 0.8.0 and has no omitted shape, so + it defaults to a value that is deliberately **not** a real InfoSource id + (#28). The harness is not an issuer: it holds no domain state and has nothing + to name here. A recognizable placeholder makes the facts a seed run produces + identifiable as synthetic; pass ``--info-source-id`` when the point of the + run is to watch a real issuer's correlation work end to end. """ return ContentFetchCommand( occurred_at=datetime.now(UTC), @@ -113,6 +141,7 @@ def build_command( url=url, headers=headers, timeout_seconds=timeout_seconds, + info_source_id=info_source_id, ) @@ -166,20 +195,48 @@ def resolve_db(client: Redis) -> int: return int(client.connection_pool.connection_kwargs.get("db") or 0) -def guard_production_target(topic: str, *, db: int, production: bool) -> None: - """Refuse the live command stream unless the caller opted in. +def guard_production_target(topic: str, *, db: int, production: bool, info_source_id: str) -> None: + """Refuse the live command stream unless the caller opted in, and meant it twice. The gate is the *conjunction*, because that is what determines reach: ``content.fetch`` on a scratch database has no consumer, and a scratch topic on db 0 is not polled by anything. Only both together put bytes through the running service. + + Two refusals behind that one gate. ``--production`` is the first: the worker + will fetch these URLs for real. The placeholder ``info_source_id`` is the + second (CR #8) — a real fetch publishes real facts to the real + ``content.blobs``, and every one of them echoes this value. Left at the + default they would name an InfoSource that exists in no issuer's table, on a + broadcast stream nothing trims. Refused here rather than defaulted away, + because the harness cannot know which InfoSource an operator meant. + + Both checks sit inside the same conjunction on purpose: a scratch run reaches + no consumer, so inventing an id there is exactly what the placeholder is for. """ - if not (db == 0 and topic == streams.CONTENT_FETCH) or production: + if not (db == 0 and topic == streams.CONTENT_FETCH): return - raise ProductionTargetError( - f"{topic} on db {db} is the live command stream — the running worker will fetch " - f"these URLs for real. Pass --production to mean it." - ) + if not production: + raise ProductionTargetError( + f"{topic} on db {db} is the live command stream — the running worker will fetch " + f"these URLs for real. Pass --production to mean it." + ) + if info_source_id == SEED_INFO_SOURCE_ID: + raise ProductionTargetError( + f"{topic} on db {db} publishes real facts, and every one echoes info_source_id. " + f"Pass --info-source-id naming the InfoSource these fetches are for, rather than " + f"announcing {SEED_INFO_SOURCE_ID!r} to the cluster." + ) + # co-core sets no ``min_length``, so a blank id publishes cleanly and names + # nothing — the same shape as the blank ``command_id`` MUST-1 refuses, and + # refused for the same reason. The *worker* deliberately does not check this + # (reading the value would be interpretation); the harness is a producer, so + # that reasoning does not carry over to it. + if not info_source_id.strip(): + raise ProductionTargetError( + f"{topic} on db {db} publishes real facts, and a blank info_source_id names " + f"nothing. Pass --info-source-id with the InfoSource these fetches are for." + ) def resolve_blobs_topic(topic: str, override: str | None) -> str: @@ -204,6 +261,7 @@ async def publish( on_published: Callable[[SeedResult], None] | None = None, headers: dict[str, str] | None = None, timeout_seconds: float | None = None, + info_source_id: str = SEED_INFO_SOURCE_ID, ) -> list[SeedResult]: """XADD one command per URL, in the order given. @@ -219,7 +277,7 @@ async def publish( publisher = AsyncBusPublisher(client) results = [] for url in urls: - command = build_command(url, headers, timeout_seconds) + command = build_command(url, headers, timeout_seconds, info_source_id) result = await publisher.execute(BusPublish(topic, to_wire(command))) published = SeedResult(command.command_id, url, result.bus_message_id) results.append(published) @@ -373,6 +431,17 @@ def build_parser() -> argparse.ArgumentParser: metavar="'Name: value'", help="request header to attach to every command; repeatable (#11)", ) + parser.add_argument( + "--info-source-id", + default=SEED_INFO_SOURCE_ID, + dest="info_source_id", + help=( + "domain key echoed onto both facts, required on the wire since co-core 0.8.0; " + "a real value is required with --production, which refuses both the default " + f"and a blank (default: {SEED_INFO_SOURCE_ID!r}, which no issuer's " + "InfoSource table contains)" + ), + ) parser.add_argument( "--timeout", type=float, @@ -396,6 +465,7 @@ def _print_dry_run( urls: list[str], headers: dict[str, str] | None = None, timeout_seconds: float | None = None, + info_source_id: str = SEED_INFO_SOURCE_ID, ) -> None: """Show the wire frames without publishing them. @@ -405,7 +475,7 @@ def _print_dry_run( travel, after this script's own stripping. """ for url in urls: - command = build_command(url, headers, timeout_seconds) + command = build_command(url, headers, timeout_seconds, info_source_id) print(f"would publish to {topic}: {to_wire(command)}") @@ -442,7 +512,12 @@ async def _seed(client: Redis, args: argparse.Namespace) -> int: published yet (CR #15). """ try: - guard_production_target(args.topic, db=resolve_db(client), production=args.production) + guard_production_target( + args.topic, + db=resolve_db(client), + production=args.production, + info_source_id=args.info_source_id, + ) except ProductionTargetError as exc: print(f"error: {exc}", file=sys.stderr) return 2 @@ -482,6 +557,7 @@ def report(result: SeedResult) -> None: on_published=report, headers=args.headers, timeout_seconds=args.timeout_seconds, + info_source_id=args.info_source_id, ) except (RedisError, OSError) as exc: print( @@ -559,7 +635,9 @@ def main(argv: list[str] | None = None) -> int: """Parse arguments and run; the dry run never opens a connection.""" args = build_parser().parse_args(argv) if args.dry_run: - _print_dry_run(args.topic, args.urls, args.headers, args.timeout_seconds) + _print_dry_run( + args.topic, args.urls, args.headers, args.timeout_seconds, args.info_source_id + ) return 0 return asyncio.run(run(args)) diff --git a/src/core/config.py b/src/core/config.py index 599422b..6ed05a0 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -59,8 +59,19 @@ class Settings(BaseSettings): # (archiver#118). Seven days is deliberately far above any plausible answer # rather than a measured figure, so the open question is a confirmation # rather than a blocker. Raise it if a consumer says it needs longer. + # + # Bounded since #28 made it arithmetic rather than only a comparison: the + # handler publishes ``stored_at + blob_ttl_seconds`` as blob_expires_at, and + # an unbounded float makes that addition raise OverflowError *after* the + # bytes are stored — not a transient error, so the command walks the delivery + # ceiling into the DLQ and leaves its blob behind as an orphan. The ceiling is + # ten years: far past any retention anyone would ask for, and far short of + # what datetime arithmetic refuses (CR #7). blob_ttl_seconds: float = Field( - default=7 * 24 * 60 * 60, validation_alias="REPLICATOR_BLOB_TTL_SECONDS" + default=7 * 24 * 60 * 60, + gt=0, + le=10 * 365 * 24 * 60 * 60, + validation_alias="REPLICATOR_BLOB_TTL_SECONDS", ) # How often the sweep walks the tree. Also the staleness bound on the diff --git a/src/worker/handler.py b/src/worker/handler.py index a30f75a..7ff9d4f 100644 --- a/src/worker/handler.py +++ b/src/worker/handler.py @@ -9,7 +9,7 @@ import asyncio import math import re -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import NamedTuple, Protocol import httpx @@ -256,6 +256,13 @@ async def handle(command: ContentFetchCommand) -> None: # new write goes uncounted. Both drifts are bounded by the sweep # interval, after which observe() replaces the estimate outright. is_new = not store.exists(fingerprint) + # Read *before* the store, so the horizon published below can only be + # earlier than the blob's real one. store() stamps the mtime the sweep + # measures against, and mtime >= stored_at by however long the write + # takes; deriving the horizon from a clock read afterwards would put the + # announced expiry past the real one, which is the direction that leaves + # a consumer holding a dead blob_uri. + stored_at = datetime.now(UTC) blob_uri = store.store(result.content, fingerprint, media_type) if is_new: usage.add(len(result.content)) @@ -270,6 +277,21 @@ async def handle(command: ContentFetchCommand) -> None: media_type=media_type, url=command.url, command_id=command.command_id, + # Copied across, never read (#28). The value is opaque to the + # byte path — nothing here parses it, branches on it, keys on it, + # or stores it — so the only way to get it wrong is to transform + # it. tests/test_boundaries.py holds that line mechanically. + info_source_id=command.info_source_id, + # When these bytes stop being retrievable at blob_uri + # (cannobserv#301). Published because the TTL clock runs from the + # last fetch *reference* — store() touches the mtime on its + # content-addressed short-circuit — and no consumer can observe + # that event. The alternative was every consumer re-deriving the + # horizon from the contract's MUST-7 TTL, hard-coding a retention + # policy this service owns and starting the clock in the wrong + # place. The real reap is later still: the sweep only runs every + # blob_sweep_interval_seconds, so this errs early in both terms. + blob_expires_at=stored_at + timedelta(seconds=settings.blob_ttl_seconds), # The metadata a broadcast consumer cannot recover once fetching # lives here rather than in Watcher (cannobserv#271). Every one # is optional, and None means "nobody said" — never a stand-in diff --git a/src/worker/loop.py b/src/worker/loop.py index e813f47..e40ff06 100644 --- a/src/worker/loop.py +++ b/src/worker/loop.py @@ -88,10 +88,16 @@ class FailureReport: ``reason`` comes from the handler where the handler knows it (``PermanentFetchError.reason``) and from the loop where only the loop does — an unrecognized ``schema_version``, a foreign payload, the delivery ceiling. + + ``info_source_id`` is carried, never consulted (#28). It is required rather + than defaulted because co-core requires it on the fact: a report built + without one could not be published at all, so a default here would only move + the failure from this constructor to the reporter's. """ command_id: str url: str + info_source_id: str reason: FailureReason status_code: int | None = None attempts: int | None = None @@ -185,15 +191,20 @@ async def process_message( reason="unsupported schema_version", detail={"command_id": command.command_id, "schema_version": command.schema_version}, reporter=reporter, - # Reading two fields off a version this worker does not support is + # Reading three fields off a version this worker does not support is # the destructuring the contract warns issuers about — done knowingly - # and only here: command_id and url are the v1 baseline, and a fact - # naming neither could not close anything. If a future version moves - # them, this branch is where it breaks — and ``_close`` refuses a - # report with no correlator rather than publishing an empty one. + # and only here: command_id, url and info_source_id are the v1 + # baseline, and a fact naming none of them could not close anything. + # All three are *required* on the command since co-core 0.8.0, so a + # frame that decoded at all has them; a frame missing one never + # reaches this branch, having failed ``from_wire`` outright. If a + # future version moves them, this branch is where it breaks — and + # ``_close`` refuses a report with no correlator rather than + # publishing an empty one. report=FailureReport( command_id=command.command_id, url=command.url, + info_source_id=command.info_source_id, reason=FailureReason.UNSUPPORTED_SCHEMA_VERSION, detail=f"schema_version={command.schema_version}", ), @@ -249,6 +260,7 @@ async def process_message( report=FailureReport( command_id=command.command_id, url=command.url, + info_source_id=command.info_source_id, reason=exc.reason, status_code=exc.status_code, detail=str(exc), @@ -319,6 +331,7 @@ async def _handle_unclassified( report=FailureReport( command_id=command.command_id, url=command.url, + info_source_id=command.info_source_id, reason=FailureReason.HANDLER_ERROR, attempts=attempts, detail=error, diff --git a/src/worker/reporter.py b/src/worker/reporter.py index 8044a24..7c678af 100644 --- a/src/worker/reporter.py +++ b/src/worker/reporter.py @@ -61,6 +61,10 @@ async def report(failure: FailureReport) -> None: occurred_at=datetime.now(UTC), command_id=failure.command_id, url=failure.url, + # Copied across, never read (#28). Replicator holds no domain state + # and this does not change that: the value is opaque here, and the + # only thing that could go wrong with it is transforming it. + info_source_id=failure.info_source_id, reason=failure.reason, # Every fact Replicator emits today closes its command; see the # module docstring and FailureReport. diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 2404c88..cf9db8f 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -126,3 +126,22 @@ def test_an_out_of_range_pacing_interval_fails_at_startup(monkeypatch, value): with pytest.raises(ValidationError): Settings() + + +@pytest.mark.parametrize("value", ["0", "-1", "1e30"], ids=["zero", "negative", "absurd"]) +def test_an_out_of_range_blob_ttl_fails_at_startup(monkeypatch, value): + """The TTL is arithmetic now, so an absurd value is a crash rather than a knob (CR #7). + + Since #28 the handler publishes ``stored_at + blob_ttl_seconds``. An + unbounded float makes that addition raise ``OverflowError`` — *after* the + bytes are on disk, and not a transient error, so the command burns the + delivery ceiling and dead-letters while its blob stays behind as an orphan. + A config typo should not be able to manufacture those. + + Zero and negative are refused for a plainer reason: they expire every blob + the moment it is written, which the sweep would carry out. + """ + monkeypatch.setenv("REPLICATOR_BLOB_TTL_SECONDS", value) + + with pytest.raises(ValidationError): + Settings() diff --git a/tests/test_boundaries.py b/tests/test_boundaries.py index cd7e316..637fab6 100644 --- a/tests/test_boundaries.py +++ b/tests/test_boundaries.py @@ -134,8 +134,36 @@ def test_the_import_detector_sees_a_planted_import(): # scan to identifiers is what lets the domain noun stay. DOMAIN_TOKENS = frozenset({"info_source", "info_item", "watched_item", "tenant", "aspect"}) +# The one carve-out, and it is one *field* wide, not one word wide (#28). co-core +# 0.8.0 makes ``info_source_id`` required on the command and on both facts, so +# Replicator must name it to copy it across. What the charter still forbids is +# *understanding* it — see ``test_the_echoed_domain_key_is_never_interpreted``, +# which is the assertion that makes this allowlist safe to have written. +# +# ECHOED_TOKEN is the DOMAIN_TOKENS entry the exemption suppresses; +# ECHOED_NAME is the only identifier it may be suppressed *for*. The two are +# separate because the token scan matches substrings: exempting the token alone +# would let ``info_source_policy`` or ``info_sources`` into an allowlisted module +# under cover of the field that earned the carve-out (CR #2). +ECHOED_TOKEN = "info_source" +ECHOED_NAME = "info_source_id" + +# Exactly the three modules on the emit path: the two publishers, and the report +# dataclass that carries the value between them. Deliberately not a directory +# glob — ``src/worker/`` also holds the pacer, the sweep and the policy reader, +# none of which has any business naming a domain object. +DOMAIN_ECHO_MODULES = frozenset( + {"src/worker/handler.py", "src/worker/reporter.py", "src/worker/loop.py"} +) + +# Where the echo may never spread, asserted against the allowlist itself rather +# than against the source. A future change that needs the domain key in the +# settings table or in a blob path would reach for this list first, and adding a +# module here is the moment the charter is actually being edited. +DOMAIN_ECHO_FORBIDDEN = ("src/core/config.py", "src/storage/", "src/api/") -def _docstring_nodes(tree: ast.Module) -> set[int]: + +def _docstring_nodes(tree: ast.AST) -> 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): @@ -151,17 +179,10 @@ def _docstring_nodes(tree: ast.Module) -> set[int]: 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) +def _surface_of(nodes: list[ast.AST], skip: set[int]) -> set[str]: + """Identifiers and non-docstring string literals across ``nodes``.""" surface: set[str] = set() - for node in ast.walk(tree): + for node in nodes: if isinstance(node, ast.Name): surface.add(node.id) elif isinstance(node, ast.Attribute): @@ -179,30 +200,191 @@ def _vocabulary_surface(tree: ast.Module) -> set[str]: 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)] +def _vocabulary_surface(tree: ast.AST) -> 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. + """ + return _surface_of(list(ast.walk(tree)), _docstring_nodes(tree)) + + +def _tokens_in(surface: set[str]) -> set[str]: + """Which ``DOMAIN_TOKENS`` appear anywhere in ``surface``.""" + lowered = [text.lower() for text in surface] return {token for token in DOMAIN_TOKENS if any(token in text for text in lowered)} +def _domain_hits(tree: ast.AST) -> set[str]: + """Tokens from ``DOMAIN_TOKENS`` appearing anywhere in the vocabulary surface.""" + return _tokens_in(_vocabulary_surface(tree)) + + +def _unechoed_domain_names(tree: ast.AST) -> set[str]: + """Names in ``tree`` that carry ``ECHOED_TOKEN`` but are not the echoed field. + + The exemption is for one field. ``info_source_policy``, ``info_sources`` and + ``info_source_cache`` all contain the token, and a substring exemption would + admit every one of them into an allowlisted module — a per-InfoSource map + arriving under cover of the field that earned the carve-out (CR #2). + """ + return { + text + for text in _vocabulary_surface(tree) + if ECHOED_TOKEN in text.lower() and text.lower() != ECHOED_NAME + } + + +def _parents(tree: ast.AST) -> dict[int, ast.AST]: + """Each node's parent, by id. ``ast`` does not record them and the scan needs them.""" + parent: dict[int, ast.AST] = {} + for node in ast.walk(tree): + for child in ast.iter_child_nodes(node): + parent[id(child)] = node + return parent + + +def _is_echoed_value(node: ast.AST, parent: dict[int, ast.AST]) -> bool: + """Whether this occurrence is the value of an ``info_source_id=`` keyword. + + ``Event(info_source_id=command.info_source_id)`` — the one position from + which the value can only travel onward, because a keyword argument named for + the field it fills cannot also be a lookup key or a branch. + """ + context = parent.get(id(node)) + return isinstance(context, ast.keyword) and context.arg == ECHOED_NAME + + +def _echo_violations(tree: ast.AST) -> list[str]: + """Every occurrence of the echoed name that is **not** one of its legal shapes. + + Inverted from the deny-list this replaces (CR #16). That one enumerated the + positions in which reading the value was forbidden — ``Compare``, + ``Subscript``, ``BinOp``, positional call arguments, and so on — and three + consecutive review rounds found positions it had not enumerated: six in the + first (dict keys, ``.get()``, concatenation, path building, …), one in the + second (keyword arguments under a different parameter name), two in the third + (``assert`` and a comprehension filter). Each fix was correct and each left + the next gap unknown until somebody probed again, while the charter cited the + scan as *the* guard making the vocabulary carve-out safe. A deny-list can only + ever be as complete as its last probe. + + So this asks the opposite question. A verbatim echo has exactly four legal + shapes and no more: + + 1. an annotated field declaration — ``info_source_id: str``; + 2. a parameter declaration — a function that carries the value through; + 3. the keyword itself — ``info_source_id=``; + 4. the value of that keyword — ``…=command.info_source_id``. + + Everything else is flagged, including shapes nobody has thought of yet: a + subscript, a branch, an f-string, a set element, a comparison, a string + literal spelling the name as a dict key or log field. Exhaustive by + construction, so it needs no further enumeration. + + Only ``ECHOED_NAME`` is in scope, because it is the only domain name allowed + to appear in ``src/`` at all — every other token in ``DOMAIN_TOKENS`` is + refused outright by ``test_no_module_names_a_domain_concept``, in every + position, which is a strictly stronger rule than this one. + """ + parent = _parents(tree) + skip = _docstring_nodes(tree) + violations: list[str] = [] + for node in ast.walk(tree): + # Shapes 1-3: declarations. Their *uses* are still checked below, so + # allowing a parameter here cannot smuggle a lookup past the scan. + if isinstance(node, ast.arg) and node.arg == ECHOED_NAME: + continue + if isinstance(node, ast.keyword) and node.arg == ECHOED_NAME: + continue + if isinstance(node, ast.Name) and node.id == ECHOED_NAME: + context = parent.get(id(node)) + if isinstance(context, ast.AnnAssign) and context.target is node: + continue + if not _is_echoed_value(node, parent): + violations.append(f"line {node.lineno}: {ECHOED_NAME} used as a bare name") + elif isinstance(node, ast.Attribute) and node.attr == ECHOED_NAME: + if not _is_echoed_value(node, parent): + violations.append(f"line {node.lineno}: .{ECHOED_NAME} read outside the echo") + elif ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and id(node) not in skip + and ECHOED_TOKEN in node.value.lower() + ): + # A string literal spelling it is a dict key or a log field — the + # form a reviewer skims past, per the vocabulary scan's own note. + violations.append(f"line {node.lineno}: {ECHOED_TOKEN!r} in a string literal") + return violations + + 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. + + ``ECHOED_TOKEN`` is exempted **only** in ``DOMAIN_ECHO_MODULES``, only when + every name carrying it is exactly ``ECHOED_NAME``, and only for itself: an + allowlisted module that also named a tenant or a watched item is offending on + that, not on the echo. + """ + files = _python_files(SRC) + assert files, f"no modules found under {SRC} — the scan is a no-op" + + offenders = {} + for path in files: + name = path.relative_to(REPO).as_posix() + tree = _parse(path) + hits = _domain_hits(tree) + if name in DOMAIN_ECHO_MODULES and not _unechoed_domain_names(tree): + hits -= {ECHOED_TOKEN} + if hits: + offenders[name] = sorted(hits) + assert not offenders + + +def test_the_echoed_domain_key_is_never_interpreted(): + """What makes the allowlist a carve-out rather than a hole (#28). + + The charter's rule is not "do not say ``info_source_id``" — the wire contract + now requires saying it. The rule is that Replicator never learns what it + means. Naming it to copy it across is mechanics; branching on it, keying on + it, or building a string out of it is the domain model arriving one + defensible commit at a time, which is the failure this file exists to catch. """ 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) + path.relative_to(REPO).as_posix(): violations for path in files - if (hits := _domain_hits(_parse(path))) + if (violations := _echo_violations(_parse(path))) } assert not offenders +def test_the_echo_allowlist_cannot_reach_config_or_storage(): + """The allowlist is the file a future change edits *first*, so it is guarded. + + Adding a module here is how the domain key would acquire a settings entry, a + blob path segment, or an HTTP surface — each a different way of holding + domain state, and each one this assertion names before the code exists. + """ + for forbidden in DOMAIN_ECHO_FORBIDDEN: + assert not [module for module in DOMAIN_ECHO_MODULES if module.startswith(forbidden)] + + +def test_every_echo_module_exists(): + """An allowlist entry that no longer names a file exempts nothing and hides + that it exempts nothing — the stale-allowlist failure mode.""" + for module in DOMAIN_ECHO_MODULES: + assert (REPO / module).is_file(), module + + @pytest.mark.parametrize( "source", [ @@ -232,6 +414,111 @@ def test_the_vocabulary_detector_ignores_prose(source): assert not _domain_hits(ast.parse(source)) +@pytest.mark.parametrize( + "source", + [ + pytest.param('if command.info_source_id == "x": ...', id="branch"), + pytest.param("if command.info_source_id: ...", id="bare-truthiness"), + pytest.param("policy = table[command.info_source_id]", id="lookup"), + pytest.param('key = f"replicator:{command.info_source_id}"', id="key-building"), + pytest.param("ok = enabled and command.info_source_id", id="bool-op"), + pytest.param( + "match command.info_source_id:\n case _:\n pass", id="match-subject" + ), + # The six CR #1 found missing. Each is a way the echoed value becomes + # state: a table keyed by it, a lookup through one, a Redis key, a path. + pytest.param("routes = {command.info_source_id: handler}", id="dict-literal-key"), + pytest.param("policy = registry.get(command.info_source_id)", id="get-lookup"), + pytest.param("seen.add(command.info_source_id)", id="membership-call"), + pytest.param('key = "replicator:" + command.info_source_id', id="concatenation"), + pytest.param("p = Path(root, command.info_source_id)", id="path-building"), + pytest.param("counts[host] += weights[command.info_source_id]", id="nested-lookup"), + # CR #11: the same hole reached by keyword syntax. redis-py's async API is + # keyword-friendly enough that this is how it would actually get written. + pytest.param("v = client.get(name=command.info_source_id)", id="keyword-lookup"), + pytest.param("await redis.set(name=command.info_source_id, value=1)", id="keyword-setter"), + pytest.param('f(**{"info_source_id": command.info_source_id})', id="splatted-mapping"), + # CR #15: two more branch positions the deny-list had not enumerated. + # Under the inverted scan they need no rule of their own — they are simply + # not one of the four legal shapes. + pytest.param("assert command.info_source_id", id="assert"), + pytest.param("xs = [p for p in ps if p.info_source_id]", id="comprehension-filter"), + # Shapes nobody enumerated, kept as evidence that the inversion holds + # without being told about them. + pytest.param("seen = {command.info_source_id}", id="set-literal"), + pytest.param("del table[command.info_source_id]", id="delete"), + pytest.param("raise KeyError(command.info_source_id)", id="raise-argument"), + pytest.param('log("...", extra={"info_source_id": x})', id="log-field"), + pytest.param("x = command.info_source_id", id="bound-to-a-local"), + ], +) +def test_the_echo_detector_sees_a_planted_read(source): + """Every one of these is *not* one of the four legal shapes, which is the only + question the inverted scan asks.""" + assert _echo_violations(ast.parse(source)) + + +@pytest.mark.parametrize( + "source", + [ + pytest.param("info_source_id: str", id="field-declaration"), + pytest.param("Event(info_source_id=command.info_source_id)", id="keyword-echo"), + # A helper that carries the value through: the parameter declares it, and + # the only thing done with it is fill the keyword named for it. + pytest.param( + "def build(info_source_id):\n return Event(info_source_id=info_source_id)", + id="carried-through-a-parameter", + ), + # The emit path in full: a model built inside a positional argument, whose + # own keyword is the echo. Both halves must stay clean. + pytest.param( + "_publish(pub, topic, Event(info_source_id=command.info_source_id), command=command)", + id="emit-path", + ), + ], +) +def test_the_echo_detector_passes_a_verbatim_echo(source): + """The four legal shapes. A detector that flagged these would force the + allowlist to be deleted rather than obeyed. + + ``x = command.info_source_id`` is deliberately **not** here: binding the value + to a local is the first half of doing something with it, and under the + inverted rule a change that needs one has to argue with this test rather than + slip past it. + """ + assert not _echo_violations(ast.parse(source)) + + +@pytest.mark.parametrize( + "source", + [ + pytest.param("self.info_source_policy = {}", id="policy-map"), + pytest.param("for x in self.info_sources: pass", id="collection"), + pytest.param("def f(info_source_url): ...", id="adjacent-field"), + pytest.param('cache = {"info_source_cache": 1}', id="string-literal"), + ], +) +def test_the_exemption_detector_sees_a_name_that_is_not_the_echoed_field(source): + """CR #2: the carve-out is one field wide. + + Each of these contains ``info_source`` and would have been exempt under a + substring-only exemption — including inside an allowlisted module, where the + interpretation scan does not reach a bare assignment or a ``for`` target. + """ + assert _unechoed_domain_names(ast.parse(source)) + + +@pytest.mark.parametrize( + "source", + [ + pytest.param("info_source_id: str", id="field-declaration"), + pytest.param("Event(info_source_id=command.info_source_id)", id="keyword-echo"), + ], +) +def test_the_exemption_detector_passes_the_echoed_field(source): + assert not _unechoed_domain_names(ast.parse(source)) + + # -------------------------------------------------------------------------- # 3. Ingress is read-only # -------------------------------------------------------------------------- diff --git a/tests/test_seed_fetch.py b/tests/test_seed_fetch.py index fbf18ea..0b11f6c 100644 --- a/tests/test_seed_fetch.py +++ b/tests/test_seed_fetch.py @@ -9,7 +9,7 @@ the script quietly publishing something the worker cannot decode. """ -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest from co_core.pure.adapters.bus import streams @@ -24,6 +24,7 @@ from ulid import ULID from scripts.seed_fetch import ( + SEED_INFO_SOURCE_ID, ProductionTargetError, SeedResult, _report_facts, @@ -41,6 +42,9 @@ TOPIC = "replicator.itest.fetch" BLOBS = "replicator.itest.blobs" URL = "https://example.test/a" +# The domain key the sample facts echo. Unlike the harness's own default, so +# a helper that quietly substituted SEED_INFO_SOURCE_ID would be visible. +INFO_SOURCE_ID = "isrc-sample" async def decoded_commands(client, topic: str = TOPIC) -> list[ContentFetchCommand]: @@ -60,10 +64,10 @@ async def decoded_commands(client, topic: str = TOPIC) -> list[ContentFetchComma def fact_fields(command_id: str) -> dict[str, str]: """The wire frame the worker's handler publishes for a handled command. - Carries the enriched fetch metadata (#10) as well, because the claim in that - first line is what the helper is for: a sample missing six of the fields the - handler actually sends is a quietly wrong model of the wire, and the watch - output is read against it. + Carries the enriched fetch metadata (#10) and the blob lifetime (#28) as + well, because the claim in that first line is what the helper is for: a sample + missing fields the handler actually sends is a quietly wrong model of the + wire, and the watch output is read against it. """ return to_wire( BlobAvailableEvent( @@ -74,12 +78,14 @@ def fact_fields(command_id: str) -> dict[str, str]: media_type="text/html", url=URL, command_id=command_id, + info_source_id=INFO_SOURCE_ID, final_url=URL, status_code=200, fetched_at=datetime.now(UTC), content_type_raw="text/html; charset=utf-8", etag='W/"abc-123"', last_modified="Wed, 21 Oct 2015 07:28:00 GMT", + blob_expires_at=datetime.now(UTC) + timedelta(days=7), ) ) @@ -98,6 +104,7 @@ async def add_failure(client, command_id: str, topic: str = BLOBS) -> bytes: occurred_at=datetime.now(UTC), command_id=command_id, url=URL, + info_source_id=INFO_SOURCE_ID, reason="http_status", terminal=True, status_code=404, @@ -154,21 +161,55 @@ async def test_publishing_preserves_the_order_the_urls_were_given(fake_redis): def test_the_live_command_stream_on_the_live_database_is_refused(): """The one combination the running service picks up: db 0 + content.fetch.""" with pytest.raises(ProductionTargetError): - guard_production_target(streams.CONTENT_FETCH, db=0, production=False) + guard_production_target( + streams.CONTENT_FETCH, db=0, production=False, info_source_id="isrc-real" + ) def test_the_live_target_is_allowed_with_an_explicit_opt_in(): - guard_production_target(streams.CONTENT_FETCH, db=0, production=True) + guard_production_target( + streams.CONTENT_FETCH, db=0, production=True, info_source_id="isrc-real" + ) + + +def test_the_placeholder_domain_key_is_refused_on_the_live_target(): + """CR #8: --production opts into a real fetch, and its facts go to the real + content.blobs. A fact naming an InfoSource that cannot exist is not something + the harness should be able to publish there by omission.""" + with pytest.raises(ProductionTargetError): + guard_production_target( + streams.CONTENT_FETCH, db=0, production=True, info_source_id=SEED_INFO_SOURCE_ID + ) + + +@pytest.mark.parametrize("value", ["", " "], ids=["empty", "whitespace"]) +def test_a_blank_domain_key_is_refused_on_the_live_target(value): + """CR #13: co-core sets no ``min_length``, so a blank id publishes cleanly and + names nothing. The worker must not refuse it — reading the value is + interpretation — but the harness is a *producer*, so the argument does not + carry over: MUST-1's "an empty id is not an id" applies here by analogy.""" + with pytest.raises(ProductionTargetError): + guard_production_target(streams.CONTENT_FETCH, db=0, production=True, info_source_id=value) + + +def test_the_placeholder_is_fine_anywhere_the_guard_does_not_bite(): + """Same conjunction as the target guard: a scratch run reaches no consumer, + so inventing an id there is exactly what the placeholder is for.""" + guard_production_target( + streams.CONTENT_FETCH, db=15, production=False, info_source_id=SEED_INFO_SOURCE_ID + ) def test_the_command_stream_on_a_scratch_database_is_allowed(): """No worker is polling db 15 — that stream reaches nothing.""" - guard_production_target(streams.CONTENT_FETCH, db=15, production=False) + guard_production_target( + streams.CONTENT_FETCH, db=15, production=False, info_source_id="isrc-real" + ) def test_a_scratch_stream_on_the_live_database_is_allowed(): """Nothing consumes ``replicator.itest.*``; the guard is about reach, not db.""" - guard_production_target(TOPIC, db=0, production=False) + guard_production_target(TOPIC, db=0, production=False, info_source_id="isrc-real") @pytest.mark.parametrize( @@ -404,9 +445,27 @@ async def test_a_run_publishes_and_closes_the_client_it_opened(fake_redis, owned assert owned_client == [True] -async def test_a_refused_target_publishes_nothing_and_still_closes(fake_redis, owned_client): - """The guard fires before the first XADD, not after a partial run.""" - code = await run(seed_args("--topic", streams.CONTENT_FETCH, URL)) +@pytest.mark.parametrize( + "argv", + [ + pytest.param(["--topic", streams.CONTENT_FETCH], id="no-production-opt-in"), + pytest.param( + ["--topic", streams.CONTENT_FETCH, "--production"], id="placeholder-domain-key" + ), + pytest.param( + ["--topic", streams.CONTENT_FETCH, "--production", "--info-source-id", " "], + id="blank-domain-key", + ), + ], +) +async def test_a_refused_target_publishes_nothing_and_still_closes(fake_redis, owned_client, argv): + """The guard fires before the first XADD, not after a partial run. + + Parametrized over all three refusal causes (CR #18) because the guard grew a + fourth argument, and the wiring from ``args.info_source_id`` into the call is + exactly the plumbing a unit test of the guard alone cannot see. + """ + code = await run(seed_args(*argv, URL)) assert code == 2 assert await fake_redis.xlen(streams.CONTENT_FETCH) == 0 @@ -560,6 +619,38 @@ async def test_published_commands_carry_the_request_options(fake_redis): assert command.timeout_seconds == 2.5 +async def test_published_commands_carry_the_domain_key(fake_redis): + """CR #5: the flag crosses three functions before it reaches the wire. + + ``build_command`` -> ``publish`` -> the frame, each with a default that would + silently substitute the placeholder if an argument were dropped — and a seed + run would then report success while publishing a synthetic id. + """ + await publish(fake_redis, TOPIC, [URL], info_source_id="isrc-real") + + (command,) = await decoded_commands(fake_redis) + assert command.info_source_id == "isrc-real" + + +async def test_a_command_defaults_to_the_placeholder_domain_key(fake_redis): + """The harness is not an issuer and has nothing real to name here.""" + await publish(fake_redis, TOPIC, [URL]) + + (command,) = await decoded_commands(fake_redis) + assert command.info_source_id == SEED_INFO_SOURCE_ID + + +async def test_the_domain_key_flag_reaches_a_published_command(fake_redis, owned_client): + """End to end through argparse, since the flag is plumbed by hand from + ``args`` into ``publish`` and a missed wiring is invisible below that.""" + args = seed_args("--topic", TOPIC, "--info-source-id", "isrc-from-the-cli", URL) + + assert await run(args) == 0 + + (command,) = await decoded_commands(fake_redis) + assert command.info_source_id == "isrc-from-the-cli" + + async def test_a_command_without_options_carries_neither_field(fake_redis): """Omitted stays omitted: the pre-#11 wire, byte for byte.""" await publish(fake_redis, TOPIC, [URL]) diff --git a/tests/worker/conftest.py b/tests/worker/conftest.py index e65d073..a4c7acd 100644 --- a/tests/worker/conftest.py +++ b/tests/worker/conftest.py @@ -39,6 +39,12 @@ # target is visible rather than tautological. URL = "https://example.test/a" +# The domain key the facts echo (#28). Deliberately unlike the default +# ``command_id`` in both prefix and shape: the two travel together on every fact, +# and an emit site that read the wrong attribute would be invisible if the tests +# gave them similar-looking values. +INFO_SOURCE_ID = "isrc-01JQ8Z" + def fetch_result( content: bytes = BODY, @@ -152,11 +158,15 @@ def command( url: str = URL, headers: dict[str, str] | None = None, timeout_seconds: float | None = None, + info_source_id: str = INFO_SOURCE_ID, ) -> ContentFetchCommand: """A decoded ``content.fetch`` command, as the handler receives it. ``headers`` / ``timeout_seconds`` default to ``None`` — the omitted-field shape, which the contract says must behave exactly as it did before #11. + + ``info_source_id`` has no ``None`` shape to default to: co-core 0.8.0 makes it + required, so every command the worker can ever decode carries one (#28). """ return ContentFetchCommand( occurred_at=datetime.now(UTC), @@ -164,6 +174,7 @@ def command( url=url, headers=headers, timeout_seconds=timeout_seconds, + info_source_id=info_source_id, ) @@ -210,9 +221,10 @@ def make_command( url: str = URL, headers: dict[str, str] | None = None, timeout_seconds: float | None = None, + info_source_id: str = INFO_SOURCE_ID, ) -> dict[str, str]: """A well-formed ``content.fetch`` wire frame.""" - return to_wire(command(command_id, url, headers, timeout_seconds)) + return to_wire(command(command_id, url, headers, timeout_seconds, info_source_id)) @pytest.fixture diff --git a/tests/worker/test_handler.py b/tests/worker/test_handler.py index ba8368e..87bab32 100644 --- a/tests/worker/test_handler.py +++ b/tests/worker/test_handler.py @@ -9,7 +9,15 @@ from src.core.errors import FailureReason, PermanentFetchError, TransientFetchError from src.storage.local import LocalBlobStore from src.storage.sweeper import BlobUsage -from tests.worker.conftest import BODY, URL, FakeFetcher, command, fetch_result, published_facts +from tests.worker.conftest import ( + BODY, + INFO_SOURCE_ID, + URL, + FakeFetcher, + command, + fetch_result, + published_facts, +) async def test_the_command_url_is_the_url_fetched(handler): @@ -47,6 +55,46 @@ async def test_a_successful_fetch_publishes_blob_available(handler, fake_redis, assert fact.command_id == "cmd-7" +async def test_the_domain_key_is_echoed_verbatim_onto_the_success_fact(handler, fake_redis): + """#28: copy ``info_source_id`` across, interpret nothing. + + Verbatim is the whole requirement — Replicator holds no domain state, so the + value is opaque here and the only way to get it wrong is to transform it. + Asserted against a value the handler could not have derived from anything + else on the command, so a site reading ``command_id`` by mistake fails + instead of coincidentally matching. + """ + await handler()(command("cmd-7", info_source_id="isrc-elsewhere")) + + (fact,) = await published_facts(fake_redis) + assert fact.info_source_id == "isrc-elsewhere" + assert fact.command_id == "cmd-7" + + +async def test_an_opaque_domain_key_is_not_normalized(handler, fake_redis): + """Whatever the issuer sent is what the fact carries, shape included. + + Replicator has no schema for this value and must not acquire one: trimming, + lower-casing, or rejecting an odd-looking id would all be *interpretation*, + and the charter's rule is that the mechanics layer never reads domain meaning + (``docs/contracts/replicator-boundaries.md``). co-core sets no ``min_length``, + so even the empty string is the issuer's business, not the fetcher's. + """ + await handler()(command(info_source_id=" Odd/Id:v2 ")) + + (fact,) = await published_facts(fake_redis) + assert fact.info_source_id == " Odd/Id:v2 " + + +async def test_the_default_command_carries_the_shared_domain_key(handler, fake_redis): + """Guards the fixture itself: a default of ``""`` would make the echo tests + above pass against a handler that hardcoded a blank.""" + await handler()(command()) + + (fact,) = await published_facts(fake_redis) + assert fact.info_source_id == INFO_SOURCE_ID + + async def test_the_fact_stream_defaults_to_content_blobs(handler, fake_redis): """The override below must not be able to move production off the real stream.""" await handler()(command()) diff --git a/tests/worker/test_handler_metadata.py b/tests/worker/test_handler_metadata.py index 33e16a2..2aa17d7 100644 --- a/tests/worker/test_handler_metadata.py +++ b/tests/worker/test_handler_metadata.py @@ -12,12 +12,19 @@ ``command.url`` for a missing ``final_url``, or the normalized ``application/octet-stream`` for a missing ``Content-Type``, would pass a type check and destroy the only thing these fields are for. + +``blob_expires_at`` (cannobserv#301, populated by #28) is the one field here that +describes the **store** rather than the fetch. It is grouped with the others +because it answers the same question — what can a consumer not recover on its +own — and its tests assert a *direction* rather than a value: the announced +horizon must never fall later than the blob's real one. """ from datetime import UTC, datetime, timedelta import pytest +from src.core.config import get_settings from src.worker.handler import MAX_HEADER_VALUE_LENGTH from tests.worker.conftest import URL, FakeFetcher, command, fetch_result, published_facts @@ -225,3 +232,54 @@ async def test_an_oversized_content_type_still_normalizes_to_a_media_type(handle (fact,) = await published_facts(fake_redis) assert fact.content_type_raw is None assert fact.media_type == "application/octet-stream" + + +async def test_the_blob_lifetime_is_announced(handler, fake_redis): + """A consumer cannot observe the TTL clock, so the fact carries the horizon. + + Its start point is the *last fetch reference* (``LocalBlobStore.store`` + ``os.utime``s on the content-addressed short-circuit), which is an event no + consumer sees. Deriving it from the issuer contract's MUST-7 TTL instead + would hard-code a retention policy the fetcher owns and start the clock in + the wrong place (cannobserv#301). + """ + await handler()(command()) + + (fact,) = await published_facts(fake_redis) + assert fact.blob_expires_at is not None + assert fact.blob_expires_at.tzinfo is not None + + +async def test_the_announced_horizon_never_outlives_the_real_one(handler, fake_redis): + """The direction is the whole guarantee, and it is one-way. + + The blob's real expiry is its mtime plus the TTL, and the mtime is stamped + *inside* ``store`` — after the handler reads the clock it derives this field + from. Any drift therefore lands with the announced horizon **earlier** than + the real one, so a consumer that re-fetches on it is early rather than + holding a dead ``blob_uri``. Deriving from ``occurred_at`` instead (stamped + after the store returns) would invert exactly this, by microseconds and in + the direction that lies. + """ + ttl = timedelta(seconds=get_settings().blob_ttl_seconds) + + await handler()(command()) + + (fact,) = await published_facts(fake_redis) + assert fact.fetched_at is not None + assert fact.blob_expires_at is not None + assert fact.fetched_at + ttl <= fact.blob_expires_at + assert fact.blob_expires_at <= datetime.now(UTC) + ttl + + +async def test_the_horizon_tracks_the_configured_ttl(handler, fake_redis, monkeypatch): + """Not a constant: an operator who shortens retention must not leave every + consumer holding a seven-day promise the sweep will not keep.""" + monkeypatch.setenv("REPLICATOR_BLOB_TTL_SECONDS", "60") + get_settings.cache_clear() + + await handler()(command()) + + (fact,) = await published_facts(fake_redis) + assert fact.blob_expires_at is not None + assert fact.blob_expires_at <= datetime.now(UTC) + timedelta(seconds=60) diff --git a/tests/worker/test_loop_dlq.py b/tests/worker/test_loop_dlq.py index 9a74468..016d4db 100644 --- a/tests/worker/test_loop_dlq.py +++ b/tests/worker/test_loop_dlq.py @@ -79,6 +79,10 @@ async def test_a_foreign_payload_type_is_dead_lettered(fake_redis, consumer, set size_bytes=1, media_type="text/html", url="https://example.test/a", + # Both required since co-core 0.8.0 (#28), so a foreign + # payload can no longer be built without them either. + command_id="cmd-that-actually-succeeded", + info_source_id="isrc-of-that-other-command", ) ), ) diff --git a/tests/worker/test_loop_facts.py b/tests/worker/test_loop_facts.py index 6f24480..0bf805d 100644 --- a/tests/worker/test_loop_facts.py +++ b/tests/worker/test_loop_facts.py @@ -222,6 +222,7 @@ async def test_a_foreign_payload_is_never_announced_even_when_it_echoes_a_comman media_type="text/html", url="https://example.test/a", command_id="cmd-that-actually-succeeded", + info_source_id="isrc-of-that-other-command", ) ), ) diff --git a/tests/worker/test_policy.py b/tests/worker/test_policy.py index 98af361..14fdbbd 100644 --- a/tests/worker/test_policy.py +++ b/tests/worker/test_policy.py @@ -195,6 +195,7 @@ def test_a_payload_from_another_stream_is_ignored(caplog): foreign = BlobAvailableEvent( occurred_at=now(), command_id="c1", + info_source_id="isrc-1", url="https://slow.test/a", blob_uri="file:///tmp/x.bin", content_fingerprint="f" * 64, diff --git a/tests/worker/test_reporter.py b/tests/worker/test_reporter.py index 7341b37..3f39913 100644 --- a/tests/worker/test_reporter.py +++ b/tests/worker/test_reporter.py @@ -18,6 +18,7 @@ REPORT = FailureReport( command_id="cmd-1", url="https://example.test/a", + info_source_id="isrc-01JQ8Z", reason=FailureReason.HTTP_STATUS, status_code=404, ) @@ -45,6 +46,28 @@ async def test_a_report_becomes_a_fetch_failed_fact(reporter, fake_redis): assert fact.status_code == 404 +async def test_the_domain_key_is_echoed_verbatim_onto_the_failure_fact(reporter, fake_redis): + """#28: the failure fact carries the domain key too, for the same reason the + success fact does — an issuer closing a pending entry should not have to + reach for its private ``command_id -> domain`` map to know *what* failed. + + Asserted on a value distinct from ``command_id``, so a reporter reading the + wrong attribute off the report fails rather than matching by coincidence. + """ + await reporter()( + FailureReport( + command_id="cmd-1", + url="https://example.test/a", + info_source_id="isrc-elsewhere", + reason=FailureReason.HTTP_STATUS, + ) + ) + + (fact,) = await decoded_facts(fake_redis, streams.CONTENT_BLOBS) + assert fact.info_source_id == "isrc-elsewhere" + assert fact.command_id == "cmd-1" + + async def test_every_fact_replicator_emits_today_is_terminal(reporter, fake_redis): """A report *is* a closure — the loop only builds one when it stops retrying. @@ -64,7 +87,9 @@ async def test_the_reason_lands_on_the_wire_as_its_token(reporter, fake_redis): Watcher branches on this string. A repr leaking into it would be a wire break that no local assertion on the enum itself would catch. """ - await reporter()(FailureReport(command_id="c", url="u", reason=FailureReason.TOO_LARGE)) + await reporter()( + FailureReport(command_id="c", url="u", info_source_id="i", reason=FailureReason.TOO_LARGE) + ) entry = (await fake_redis.xrange(streams.CONTENT_BLOBS))[0][1] assert b'"reason":"too_large"' in entry[b"payload"] @@ -87,7 +112,11 @@ async def test_the_envelope_key_is_per_emission_not_per_command(reporter, fake_r async def test_absent_context_is_omitted_rather_than_guessed(reporter, fake_redis): """No HTTP exchange, no status; not on the ceiling path, no attempt count.""" - await reporter()(FailureReport(command_id="c", url="u", reason=FailureReason.NOT_FETCHABLE)) + await reporter()( + FailureReport( + command_id="c", url="u", info_source_id="i", reason=FailureReason.NOT_FETCHABLE + ) + ) (fact,) = await decoded_facts(fake_redis, streams.CONTENT_BLOBS) assert fact.status_code is None @@ -97,7 +126,12 @@ async def test_absent_context_is_omitted_rather_than_guessed(reporter, fake_redi async def test_the_ceiling_path_reports_how_many_attempts_it_took(reporter, fake_redis): await reporter()( FailureReport( - command_id="c", url="u", reason=FailureReason.HANDLER_ERROR, attempts=5, detail="boom" + command_id="c", + url="u", + info_source_id="i", + reason=FailureReason.HANDLER_ERROR, + attempts=5, + detail="boom", ) ) diff --git a/uv.lock b/uv.lock index a8c8bb6..0fcf181 100644 --- a/uv.lock +++ b/uv.lock @@ -78,15 +78,15 @@ wheels = [ [[package]] name = "co-core" -version = "0.7.7" +version = "0.8.0" source = { registry = ".wheelhouse" } dependencies = [ { name = "dateparser" }, { name = "pydantic" }, ] -sdist = { path = "co_core-0.7.7.tar.gz" } +sdist = { path = "co_core-0.8.0.tar.gz" } wheels = [ - { path = "co_core-0.7.7-py3-none-any.whl" }, + { path = "co_core-0.8.0-py3-none-any.whl" }, ] [package.optional-dependencies] @@ -100,15 +100,15 @@ extract = [ [[package]] name = "co-core-aio" -version = "0.7.7" +version = "0.8.0" source = { registry = ".wheelhouse" } dependencies = [ { name = "co-core" }, { name = "httpx" }, ] -sdist = { path = "co_core_aio-0.7.7.tar.gz" } +sdist = { path = "co_core_aio-0.8.0.tar.gz" } wheels = [ - { path = "co_core_aio-0.7.7-py3-none-any.whl" }, + { path = "co_core_aio-0.8.0-py3-none-any.whl" }, ] [package.optional-dependencies] @@ -934,8 +934,8 @@ dev = [ [package.metadata] requires-dist = [ - { name = "co-core", extras = ["extract"], specifier = ">=0.7.7,<0.8" }, - { name = "co-core-aio", extras = ["bus"], specifier = ">=0.7.7,<0.8" }, + { name = "co-core", extras = ["extract"], specifier = ">=0.8.0,<0.9" }, + { name = "co-core-aio", extras = ["bus"], specifier = ">=0.8.0,<0.9" }, { name = "fastapi", specifier = ">=0.126.0,<1" }, { name = "pydantic", specifier = ">=2.9,<3" }, { name = "pydantic-settings", specifier = ">=2.5.0,<3" },