diff --git a/AGENTS.md b/AGENTS.md index 7a4bd52..69cd76e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,7 +18,7 @@ TDD required. Red → Green → Refactor. No production code without a failing t Python 3.12, uv, hatchling src layout, pytest, ruff. -**co-core is pinned exactly (`==0.32.0`); Processor owns the pin** (spec D5, §5; since watcher#350 nothing else extracts). `processor_version` = `"+"`; a pin test makes every bump a deliberate, test-failing act. Never let a dependency bot or `uv lock --upgrade` move co-core. A bump gives Watcher notice (it re-baselines on `processor_version`). It is coordinated with Watcher only if it changes the contract: the `content.process` / `content.derived` models, `canonical_text` semantics, or `resolve_dispatch_essence`. Procedure: [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md#bumping-co-core). +**co-core is pinned exactly (`==0.37.0`); Processor owns the pin** (spec D5, §5; since watcher#350 nothing else extracts). `processor_version` = `"+"`; a pin test makes every bump a deliberate, test-failing act. Never let a dependency bot or `uv lock --upgrade` move co-core. A bump gives Watcher notice (it re-baselines on `processor_version`). It is coordinated with Watcher only if it changes the contract: the `content.process` / `content.derived` models, `canonical_text` semantics, or `resolve_dispatch_essence`. Procedure: [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md#bumping-co-core). **Cannobserv wheelhouse.** `co-core`, `co-core-aio`, `co-core-sync` resolve from `./.wheelhouse` (git-ignored) via `[tool.uv] find-links`. Populate it before any `uv` command: diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 67f7969..7e7bffb 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -447,7 +447,7 @@ On boot, the service preflights both buckets and exits non-zero if either is unr - **Drift:** `journalctl -u processor-drift` (§ The drift check). - **Liveness:** `journalctl -u processor -o cat | jq -cR 'fromjson? | select(.logger == "processor.liveness")'` (§ Liveness). - **Logs:** `journalctl -u processor`. `starting` names the `build`, `child_containment`, `landlock_abi` and `liveness`. Deploys: `journalctl -t processor-deploy`. Each outcome has `command_id`, `info_source_id`, `action` (`ack` / `strike` / `leave_pending` / `dead_letter`), `reason`, `detail`, and timings in `*_ms`. Each also has `input_digest` once it is valid, as bare hex like the command's. A complete fact adds `output_digest` (`sha256:`, as on the fact; the object is `blobs/.bin`; absent when `empty`), `output_size_bytes`, `empty` and `processor_version`, and so does a `leave_pending` whose publish failed after the store. A shadow command's output: `journalctl -u processor -o cat | jq -cR 'fromjson? | select(.command_id == "") | {action, reason, input_digest, output_digest}'` (`-R … fromjson?` skips systemd's own lines, which `-u` includes and plain `jq` aborts on). -- **Dead letters:** `processor_cli dlq list | show | drop ` (§ `/etc/processor/`). There is no replay. An entry failed to decode, was not a command, or is a command that raised outside the handler on every attempt (`reason` starts `gave up on attempt`): that one is a bug to fix. Before dead-lettering such a command, Processor published a terminal `extraction_error` whose `detail` starts `dead-lettered:` (best effort), so Watcher has closed it (#17). The journal's `dead-lettering` record says which: `failure_fact` is `published`, `skipped` (a fact had already gone out) or `refused`. On a command given up at the cap, `handle_skipped` is `true` when the record retries a give-up that did not finish (the failure fact refused transiently, or the dead-letter refused), with no re-run of the command and the original traceback (#28). The same entry logging that every ~11 min while it stays pending means `content.process.dlq` refuses it: `journalctl -u processor -o cat | jq -cR 'fromjson? | select(.handle_skipped) | {timestamp, command_id, message_id, reason}'`. +- **Dead letters:** `processor_cli dlq list | show | drop ` (§ `/etc/processor/`). There is no replay. An entry failed to decode, was not a command, or is a command that raised outside the handler on every attempt (`reason` starts `gave up on attempt`): that one is a bug to fix. Before dead-lettering such a command, Processor published a terminal `dead_lettered` whose `detail` starts `dead-lettered:` (best effort), so Watcher has closed it (#17; `extraction_error` before co-core 0.37.0, #21). The journal's `dead-lettering` record says which: `failure_fact` is `published`, `skipped` (a fact had already gone out) or `refused`. On a command given up at the cap, `handle_skipped` is `true` when the record retries a give-up that did not finish (the failure fact refused transiently, or the dead-letter refused), with no re-run of the command and the original traceback (#28). The same entry logging that every ~11 min while it stays pending means `content.process.dlq` refuses it: `journalctl -u processor -o cat | jq -cR 'fromjson? | select(.handle_skipped) | {timestamp, command_id, message_id, reason}'`. - **Lag / missing group:** the broker probe watches `processor.process` (broker#75). - **Strikes** are counted in memory. A restart resets them, so a poison command gets at most 3 more attempts. diff --git a/docs/plans/2026-10-17-21-dead-lettered.md b/docs/plans/2026-10-17-21-dead-lettered.md new file mode 100644 index 0000000..bc7a4cb --- /dev/null +++ b/docs/plans/2026-10-17-21-dead-lettered.md @@ -0,0 +1,58 @@ +--- +title: Dead-letter give-ups publish dead_lettered; co-core 0.32.0 → 0.37.0 (#21) +date: 2026-10-17 +status: per #21's hand-off from epic #24 (2026-10-17) and the v0.37.0 release comment +--- + +# dead_lettered + +## Problem + +Since #17 a command dead-lettered at the cap gets a terminal `extraction_error`, so Watcher blames the item's spec for Processor's fault. cannobserv v0.37.0 (`49d42aca`) adds `dead_lettered` to `ProcessingFailureReason` (cannobserv#536): terminal, no re-fetch, scoped to a command the processor dead-letters. Upstream's scope: +- **`handler.publish_gave_up` moves to it** (the consumer's cap: an escape past the handler, or a non-transient publish refusal). +- **The handler's own cap stays `extraction_error`** (every attempt timed out, crashed, or raised outside the parser). Upstream's accepted limit. +- After a refused ack, the fact that went out stands; no `dead_lettered` follows. + +## The catch-up reading (0.32.0 → 0.37.0) + +Checked in code (`git diff v0.32.0 v0.37.0 -- packages/`), not only the CHANGELOG: +- **Contract models** (`changes.py`): `ProcessingFailureReason` gains `"dead_lettered"`; `ProcessingFailedEvent`'s docstring documents it and the cap's `extraction_error`. `ContentProcessCommand`, `ProcessingCompleteEmit`, `ProcessingFailedEmit`'s fields and validators: unchanged. The rest is the watch pair (#563/#564), `FetchPolicyState.generation`, `WatchStatusState` levels: none of it read by Processor. +- **Bus**: `envelope.py` gains five event types and their key rules (additive); `streams.py` gains `INFO_WATCH_EVENTS` / `INFO_WATCH_CHECK` and comments. `dead_letter`, `exceptions`, `effects.bus`, `co_core_aio.bus`: unchanged. +- **Extract** (whole package), `canonical_text`, `spec_fingerprint`, `resolve_dispatch_essence`: unchanged. +- **Blob store** (`util.blobstore`, `util.hashing`, `co_core_sync.drivers.blobstore`): unchanged. +- **Dependencies**: each package's `pyproject.toml` moves only its version and the sibling floor (`>=0.37,<0.38`). +- **CHANGELOG** 0.33.0–0.36.0: Downstream lines are cli-only. 0.37.0: additive; no security entries in range. + +So the only code change is the token. + +## Approach + +**Pin.** `pyproject.toml` (three packages), `EXPECTED` in `tests/test_pin.py`, `scripts/build_wheelhouse.sh`'s default tag, `uv lock`. `LOCAL_GENERATION` stays 1, so `processor_version` = `0.37.0+1`. + +**Token.** `publish_gave_up` builds its fact with `"dead_lettered"`. `detail` keeps its `dead-lettered: ` prefix (the DLQ and the fact agree; Watcher stores `detail` as written). + +**Tests** (red first): +- `test_handler.py`: `test_publish_gave_up_publishes_one_terminal_extraction_error` → `…_dead_lettered`. +- `test_consumer.py`: give-ups through `publish_gave_up` (the escape cap, the transiently refused fact, the retried dead-letter, the failed dead-letter at the cap, the dead-letter refused every time) assert `dead_lettered`. The handler's cap (`test_three_strikes_then_a_terminal_failure`, `test_a_failed_ack_at_the_cap_keeps_the_count`) stays `extraction_error`. The refused-ack case still adds no fact. + +**Goldens.** Regenerate from co-core; expect no move (the extract package is byte-identical). A move means stop and explain. Real corpus passes unchanged. + +**Docs.** +- Spec: an amendment entry; §4 rows 313 and 315 and the #17 give-up prose; D5 "(0.32.0 today)"; §2's dependency row; §5's pin bullet plus a 0.32.0 → 0.37.0 sub-bullet; §7's pin-test line; the contract quick reference heading (and `dead_lettered` noted under `ProcessingFailureReason`); Open Question 2's range. +- Docstrings: `handler.py` module and `publish_gave_up`; `consumer.py` module. +- DEPLOYMENT.md: the dead-letter line (`extraction_error` → `dead_lettered`); the bump procedure's catch-up example. +- AGENTS.md: the pin line. + +**Off-repo (operator approval for each post).** Watcher notice on CannObserv/watcher#386 before the deploy: `processor_version` `0.37.0+1`, goldens moved: none, `dead_lettered` now published on the dead-letter path, an un-upgraded Watcher decodes it (`reason: str`). Then the bump note on #21. + +## Steps + +1. Red: pin test, handler/consumer token tests. +2. Green: wheelhouse at v0.37.0, pin, `uv lock`, the token. Regenerate goldens; record moved list. +3. Docs. +4. Full gate (`ruff check`, `ruff format --check`, `pytest` incl. integration on the scratch Redis, `uv lock --locked`); PR with `Refs #21`. +5. After FF merge: `git pull --ff-only`; Watcher notice (approved); `scripts/deploy.sh`; live checks: `starting` shows `processor_version: 0.37.0+1`, each source's digest unchanged across the bump, `NRestarts=0`. + +## Open questions + +None blocking. diff --git a/docs/specs/2026-09-29-processor-service-design.md b/docs/specs/2026-09-29-processor-service-design.md index 9f1200c..ff46361 100644 --- a/docs/specs/2026-09-29-processor-service-design.md +++ b/docs/specs/2026-09-29-processor-service-design.md @@ -63,6 +63,18 @@ on `co-processor`. gates the deploy (Section 5). - The corpus covers the XML essences, and a refusal is a golden (Section 7). +**Amended 2026-10-17 (co-core 0.37.0, processor#21):** +- The pin moves from 0.32.0 to **0.37.0** (Sections 2 and 5). It carries + cannobserv#536: `ProcessingFailureReason` gains `dead_lettered`, terminal, with + no re-fetch. +- A command dead-lettered at the cap now gets `dead_lettered`, not + `extraction_error`. The handler's own cap (every attempt timed out, crashed or + raised) stays `extraction_error` (Section 4). +- The new token changes the `content.derived` model, so Watcher gets notice + before the deploy, on watcher#386. The change is additive: an un-upgraded + Watcher decodes the token, since `reason` is a `str` on the consumer class + (Section 5). + --- ## Why a new service @@ -104,7 +116,7 @@ state and UI it should not have). | D2 | **Headless and stateless in v1**: no UI, no database. Idempotency is the write-if-absent store; correlation and dedupe are the issuer's (by `command_id`). | | D3 | **Own identities**: broker ACL user, GCS service account, bucket, tailnet tag. Observo's broker#62 `observo` user is stripped. | | D4 | **Own bucket** `gs://co-gcs-processor`, derived outputs only, so Watcher's read grant is a plain bucket-level `objectViewer`. | -| D5 | **co-core pinned exactly; Processor owns the pin** (0.32.0 today, processor#29). Bumps are deliberate (Section 5). Processor coordinates with Watcher only when a bump changes the contract: the `content.process` / `content.derived` models, `canonical_text` semantics, or `resolve_dispatch_essence`. It **gives Watcher notice whenever `processor_version` moves**, because Watcher re-baselines on it. Watcher still pins co-core for contracts and the bus (`==0.19.7` on 2026-10-10), and that pin no longer has to equal Processor's. *Amended 2026-10-10 (processor#47, after watcher#350). Until then: matched to Watcher, in lockstep, since both sides extracted during shadow. Watcher agreed to that on 2026-10-02 (watcher#325), and neither side deployed a bump until the parity corpus passed on it unchanged.* | +| D5 | **co-core pinned exactly; Processor owns the pin** (0.37.0 today, processor#21). Bumps are deliberate (Section 5). Processor coordinates with Watcher only when a bump changes the contract: the `content.process` / `content.derived` models, `canonical_text` semantics, or `resolve_dispatch_essence`. It **gives Watcher notice whenever `processor_version` moves**, because Watcher re-baselines on it. Watcher still pins co-core for contracts and the bus (`==0.19.7` on 2026-10-10), and that pin no longer has to equal Processor's. *Amended 2026-10-10 (processor#47, after watcher#350). Until then: matched to Watcher, in lockstep, since both sides extracted during shadow. Watcher agreed to that on 2026-10-02 (watcher#325), and neither side deployed a bump until the parity corpus passed on it unchanged.* | | D6 | **Extraction runs in a killable child process** with a timeout and an address-space limit, never a thread. | | D7 | Broker `NOPERM` and `OOM` are **transient**; a reclaim **re-runs** the extraction (no dedupe key). | | D8 | **v1 scope is #629's requirements only.** | @@ -142,7 +154,7 @@ processor-version pin on the command; no job API — the bus is the interface. | Thing | Value | |---|---| | Repo | `CannObserv/processor` — Python 3.12, uv, hatchling src layout | -| Dependencies | `co-core[extract]`, `co-core-aio[bus]`, `co-core-sync[gcs]`, all `==0.32.0` (Section 5), from the private CannObserv index | +| Dependencies | `co-core[extract]`, `co-core-aio[bus]`, `co-core-sync[gcs]`, all `==0.37.0` (Section 5), from the private CannObserv index | | CI | wheelhouse pulled through WIF as `co-pypi-reader` (org var `GCP_WIF_PROVIDER`), as in the other cohort repos | | VM | exe.dev `co-processor`, 8 GB, default `exeuntu` image, tag `processor` | | Service | systemd unit `processor`: `MemoryMax` below VM RAM, `Restart=on-failure`, `OOMPolicy=continue` (Section 4). Runs as its own user `processor` (no sudo, not in `docker`, no home), from a read-only release at `/srv/processor/live` that `scripts/deploy.sh` builds from a commit on `origin/main` (amended 2026-10-06, #2: the cohort's release standard, broker#22 / status#9) | @@ -310,9 +322,9 @@ Observo adopted in #395/#407) so the later plane can ingest them unchanged. | Bytes hash ≠ `input_digest` | `input_digest_mismatch`, terminal | yes | | Extractor raises, including `MemoryError` under `RLIMIT_AS`, and a spec co-core refuses (`UnsupportedExtractionError` since 0.32.0, cannobserv#523: `regex`, `jsonpath`, an empty or invalid `xpath`, an unknown algorithm, `xpath` on HTML / PDF / CSV, `css` on XML / PDF / CSV), amended 2026-10-12 | `extraction_error`, terminal | yes | | Child timeout or crash | — | no; the 3rd attempt publishes `extraction_error`, terminal, and acks | -| A non-transient exception escaping the handler (a bug, or an ack or dead-letter refused), amended 2026-09-30, 2026-10-02 and 2026-10-05 | at the cap, `extraction_error`, terminal, unless this entry's fact already went out (a fact published before a refused ack stands) | no; counted like a strike, and the 3rd attempt publishes, then dead-letters the entry with the exception as its reason. A dead-letter that does not land is retried alone, without running the command again (#28) | +| A non-transient exception escaping the handler (a bug, or an ack or dead-letter refused), amended 2026-09-30, 2026-10-02, 2026-10-05 and 2026-10-17 | at the cap, `dead_lettered`, terminal, unless this entry's fact already went out (a fact published before a refused ack stands) | no; counted like a strike, and the 3rd attempt publishes, then dead-letters the entry with the exception as its reason. A dead-letter that does not land is retried alone, without running the command again (#28) | | GCS 5xx / 429 / timeout / auth; broker `NOPERM`; broker `OOM command not allowed`; broker `MISCONF` / `BUSY` / `MASTERDOWN` / `TRYAGAIN` / `CLUSTERDOWN` / `NOREPLICAS` (amended 2026-09-30) | — | no; the entry is reclaimed | -| A non-transient publish failure (e.g. `WRONGTYPE`, or a bug in serialization), amended 2026-09-30, 2026-10-02 and 2026-10-05 | at the cap, `extraction_error`, terminal, best effort (the same refusal usually refuses it too) | no; counted like a strike, and the 3rd attempt dead-letters the entry (retried alone if it does not land, #28) | +| A non-transient publish failure (e.g. `WRONGTYPE`, or a bug in serialization), amended 2026-09-30, 2026-10-02, 2026-10-05 and 2026-10-17 | at the cap, `dead_lettered`, terminal, best effort (the same refusal usually refuses it too) | no; counted like a strike, and the 3rd attempt dead-letters the entry (retried alone if it does not land, #28) | - **`unsupported_media_type` is never emitted in v1** — `extractor_for_essence` is total (HTML for anything unknown). @@ -358,14 +370,21 @@ Observo adopted in #395/#407) so the later plane can ingest them unchanged. (`dlq_name(CONTENT_PROCESS)` = `content.process.dlq`). No replay — an undecodable frame has nothing to replay, and a command dead-lettered after escaping exceptions is a bug to fix. -- **A dead-lettered command still gets a fact** (amended 2026-10-02, #17). Before - dead-lettering a decodable command at the cap, Processor publishes - `processing_failed` with `terminal=true`, `reason=extraction_error`, and - `detail` = `dead-lettered: `. Without it Watcher would wait on - the command. Its reaper re-issues only once Processor answers a later command +- **A dead-lettered command still gets a fact** (amended 2026-10-02, #17, and + 2026-10-17, #21). Before dead-lettering a decodable command at the cap, + Processor publishes `processing_failed` with `terminal=true`, + `reason=dead_lettered`, and `detail` = `dead-lettered: `. + Without it Watcher would wait on the command. Its reaper re-issues only once Processor answers a later command (watcher#325, amended 2026-10-03), and its health reads a quiet Processor as down. Watcher closes the command instead: in shadow only an audit entry, after cutover the item's failure path. + - **The token** (cannobserv#536, co-core 0.37.0). Until 0.37.0 it was + `extraction_error`, which reads as a bad spec when the fault is + Processor's. `dead_lettered` is published only here. The handler's own cap + stays `extraction_error`: a timeout or crash can't be told from a document + that defeats the parser, and a raise outside the parser reads the same + (upstream's accepted limit). A Watcher that predates the token routes it + as it routes a give-up today; its processor-blaming route is watcher#386. - **No second failure fact.** It is skipped when this entry's fact already went out: a refused ack, or a dead-letter retried after the fact landed. That memory is per process: after a restart, a re-run that reaches the cap @@ -414,12 +433,19 @@ reclaim re-running (not re-publishing) the extraction. co-core. Since watcher#350 nothing on Watcher's side has to match. Bump it by hand only when Processor's own logic (config merging, dispatch) changes output in a way co-core's version cannot see. -- **co-core `==0.32.0`** since processor#29 (2026-10-12), the catch-up jump from +- **co-core `==0.37.0`** since processor#21 (2026-10-17). Before it, + `==0.32.0` since processor#29 (2026-10-12), the catch-up jump from 0.19.7, the version Watcher's `uv.lock` resolved at the cutover (Watcher's `pyproject.toml` allowed `>=0.19.6,<0.20`), so both sides reported `"0.19.7+1"` from the first shadow command. Since D5's 2026-10-10 amendment, Watcher's lock no longer constrains this pin. Processor needs `GcsBlobStore` (0.19.3+) and the `content.process` types (0.19.4). + - Across 0.32.0 → 0.37.0, the extract package, `canonical_text`, + `spec_fingerprint`, `resolve_dispatch_essence`, the stores and + `co_core_aio.bus` are unchanged in code. `ProcessingFailureReason` gains + `dead_lettered` (cannobserv#536). `envelope.py` and `streams.py` gain the + `info.watch-events` / `info.watch-check` streams and their payloads, which + Processor doesn't read. The goldens did not move. - Across 0.19.7 → 0.32.0, the bus, the stores, `canonical_text`, `spec_fingerprint` and the `content.process` / `content.derived` models are unchanged in code (a `changes.py` docstring and `streams.py` comments aside). @@ -543,7 +569,7 @@ TDD, red first. - Parity with Watcher's local path should be proven before shadow, not discovered in it. - **Pin test.** The lock's co-core equals the expected exact version and - `processor_version(LOCAL_GENERATION) == "0.32.0+1"` — a bump is a deliberate, + `processor_version(LOCAL_GENERATION) == "0.37.0+1"` — a bump is a deliberate, test-failing act. - **Shells — one test per Section 4 row.** Input/output stores: a local co-core store on a temp dir behind the same interface, plus fakes raising GCS 5xx / @@ -615,13 +641,13 @@ writing-plans for v1 there. Implementation happens on `co-processor`. --- -## Contract quick reference (co-core 0.19.4–0.32.0, unchanged across them but the refusal, added in 0.32.0) +## Contract quick reference (co-core 0.19.4–0.37.0, unchanged across them but the refusal, added in 0.32.0, and `dead_lettered`, in 0.37.0) | Need | Name | Module | |---|---|---| | Stream / group / DLQ | `CONTENT_PROCESS`, `CONTENT_DERIVED`, `group_name`, `dlq_name` | `co_core.pure.adapters.bus.streams` | | Decode | `from_wire` | `co_core.pure.adapters.bus.envelope` | -| Command / facts | `ContentProcessCommand`, `ProcessingCompleteEmit`, `ProcessingFailedEmit`, `ProcessingFailureReason` | `co_core.pure.models.changes` | +| Command / facts | `ContentProcessCommand`, `ProcessingCompleteEmit`, `ProcessingFailedEmit`, `ProcessingFailureReason` (`dead_lettered` since 0.37.0) | `co_core.pure.models.changes` | | Dispatch | `extractor_for_essence` | `co_core.pure.extract.dispatch` (needs the `extract` extra) | | Config | `extraction_config_from_spec` | `co_core.pure.extract.extraction_defaults` | | Refusal (0.32.0+) | `UnsupportedExtractionError` (a `ValueError`) | `co_core.pure.extract.errors` | @@ -656,7 +682,7 @@ echoed, reporting only. `ChangeRevision.processor_version` is never rewritten. `spec_fingerprint` gets the same treatment. 2. ~~**co-core-aio:** does `AsyncBusConsumer` expose a delivery count or a - max-deliveries hook?~~ **Answered, no (0.19.4 through 0.32.0).** It offers + max-deliveries hook?~~ **Answered, no (0.19.4 through 0.37.0).** It offers primitives only: `ensure_group`, `read`, `ack`, `claim_stale` / `claim_stale_page`, and `dead_letter`. It has no delivery count, no hook and no run loop. The in-memory counter stays, and Processor owns its loop. diff --git a/pyproject.toml b/pyproject.toml index 7eb4114..5bf7883 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,9 +8,9 @@ dependencies = [ # Exact, and Processor owns the pin (spec D5, §5). processor_version is # "+", so a bump moves every fact Watcher # compares; tests/test_pin.py makes one a deliberate, test-failing act. - "co-core[extract]==0.32.0", - "co-core-aio[bus]==0.32.0", - "co-core-sync[gcs]==0.32.0", + "co-core[extract]==0.37.0", + "co-core-aio[bus]==0.37.0", + "co-core-sync[gcs]==0.37.0", "pydantic>=2.12,<3", "pydantic-settings>=2.6,<3", # Imported directly: the consumer classifies redis-py's ResponseError / diff --git a/scripts/build_wheelhouse.sh b/scripts/build_wheelhouse.sh index 3d330bf..7e62088 100755 --- a/scripts/build_wheelhouse.sh +++ b/scripts/build_wheelhouse.sh @@ -8,7 +8,7 @@ set -euo pipefail root="$(cd "$(dirname "$0")/.." && pwd)" -tag="${1:-v0.32.0}" +tag="${1:-v0.37.0}" : "${GH_TOKEN_CANNOBSERV:?set GH_TOKEN_CANNOBSERV (read access to CannObserv/cannobserv)}" work="$(mktemp -d)" diff --git a/src/processor/consumer.py b/src/processor/consumer.py index 400eb15..e98c3b7 100644 --- a/src/processor/consumer.py +++ b/src/processor/consumer.py @@ -14,9 +14,9 @@ non-transient refusal of an ack or dead-letter) counts like a strike; at ``max_attempts`` the entry is dead-lettered with the exception as its reason. Transient escapes (broker, GCS) stay uncapped. Before that dead-letter, a - decodable command whose fact never went out gets a terminal ``extraction_error`` - (#17), so Watcher closes it instead of waiting; a refused one is logged and the - entry dead-lettered anyway. + decodable command whose fact never went out gets a terminal ``dead_lettered`` + (#17, #21), so Watcher closes it instead of waiting; a refused one is logged and + the entry dead-lettered anyway. - A strike count clears only once the entry's ack or dead-letter lands, so a failed one on the last attempt does not restart the count. A give-up at the cap that does not finish (the failure fact or the dead-letter refused) is retried alone (#28): diff --git a/src/processor/handler.py b/src/processor/handler.py index 4f231ab..d8dccfe 100644 --- a/src/processor/handler.py +++ b/src/processor/handler.py @@ -8,10 +8,13 @@ - ``leave_pending`` — infrastructure failed (``errors.is_transient``), the publish included: nothing published, the reclaim re-runs it. Uncapped. A non-transient publish failure escapes instead, and the consumer caps it: at the cap it - publishes ``publish_gave_up``'s terminal fact, then dead-letters (#17). + publishes ``publish_gave_up``'s terminal ``dead_lettered``, then dead-letters + (#17, #21). - ``strike`` — the child timed out or crashed, or something unexpected raised: nothing published, counted toward ``max_attempts``; on the last attempt the - handler publishes ``extraction_error`` instead and returns ``ack``. + handler publishes ``extraction_error`` instead and returns ``ack``. Not + ``dead_lettered``: a timeout or crash can't be told from a document that defeats + the parser (cannobserv#536's scope). A download that fails its checksum (``DataCorruption``) is ``input_unreadable``. @@ -274,14 +277,15 @@ async def _read_input(command: ContentProcessCommand, deps: Deps, timer: _Timer) async def publish_gave_up(command: ContentProcessCommand, deps: Deps, detail: str) -> None: - """Publish a terminal ``extraction_error`` for a command about to be dead-lettered. + """Publish a terminal ``dead_lettered`` for a command about to be dead-lettered. Without a fact, Watcher waits on the command: its reaper re-issues only once a later command is answered, and its health reads the silence as Processor down - (#17, #20). + (#17, #20). ``dead_lettered`` (cannobserv#536, #21) says the fault is Processor's, + not the spec's or the bytes'. Raises whatever the publish raises; the consumer decides. """ - fact = _failed(command, deps, "extraction_error", detail) + fact = _failed(command, deps, "dead_lettered", detail) await deps.publish(BusPublish(topic=CONTENT_DERIVED, fields=to_wire(fact))) diff --git a/tests/fixtures/parity/goldens.json b/tests/fixtures/parity/goldens.json index 36b8e9f..ae3ff80 100644 --- a/tests/fixtures/parity/goldens.json +++ b/tests/fixtures/parity/goldens.json @@ -1,5 +1,5 @@ { - "co_core": "0.32.0", + "co_core": "0.37.0", "generated_by": "scripts/gen_parity_goldens.py (co-core's pure extract API, direct; #47)", "goldens": { "csv": { diff --git a/tests/test_consumer.py b/tests/test_consumer.py index f48c670..54549ba 100644 --- a/tests/test_consumer.py +++ b/tests/test_consumer.py @@ -245,6 +245,8 @@ async def test_a_foreign_event_is_dead_lettered(admin, bus, stores) -> None: async def test_three_strikes_then_a_terminal_failure(admin, bus, stores) -> None: + # The handler's own cap stays extraction_error: a timeout can't be told from a + # document that defeats the parser. dead_lettered is the DLQ path's (#21). async def timeout(*_a, **_k) -> ChildResult: return ChildResult(kind="timeout", detail="no result after 60s") @@ -463,7 +465,7 @@ async def broken_handle(*_args, **_kwargs): assert (await group_info(admin))["pending"] == 0 (fact,) = await facts(admin) assert isinstance(fact, ProcessingFailedEvent) - assert (fact.command_id, fact.reason, fact.terminal) == ("cmd-1", "extraction_error", True) + assert (fact.command_id, fact.reason, fact.terminal) == ("cmd-1", "dead_lettered", True) assert fact.detail.startswith("dead-lettered: gave up on attempt 3: KeyError") @@ -537,7 +539,7 @@ async def flaky_publish(effect): await consumer.step() # still at the cap assert await admin.xlen(dlq_name(CONTENT_PROCESS)) == 1 - assert [f.reason for f in await facts(admin)] == ["extraction_error"] + assert [f.reason for f in await facts(admin)] == ["dead_lettered"] async def test_a_failure_fact_is_published_once_when_the_dead_letter_is_retried( @@ -567,7 +569,7 @@ async def flaky_dead_letter(self, *args, **kwargs): await consumer.step() assert await admin.xlen(dlq_name(CONTENT_PROCESS)) == 1 - assert [f.reason for f in await facts(admin)] == ["extraction_error"] + assert [f.reason for f in await facts(admin)] == ["dead_lettered"] async def test_a_refused_ack_at_the_cap_adds_no_failure_fact( @@ -705,7 +707,7 @@ async def flaky_dead_letter(self, *args, **kwargs): assert meta.reason.startswith("gave up on attempt 3: KeyError") assert (await group_info(admin))["pending"] == 0 (fact,) = await facts(admin) - assert fact.reason == "extraction_error" + assert fact.reason == "dead_lettered" assert fact.detail == f"dead-lettered: {meta.reason}" # the DLQ and the fact agree @@ -767,7 +769,7 @@ async def refused_dead_letter(self, *_args, **_kwargs): assert calls == [1, 2, 3] assert (await group_info(admin))["pending"] == 1 - assert [f.reason for f in await facts(admin)] == ["extraction_error"] + assert [f.reason for f in await facts(admin)] == ["dead_lettered"] dead = [r for r in caplog.records if r.getMessage() == "dead-lettering"] assert [(r.levelno, r.attempt, r.handle_skipped) for r in dead] == [ (logging.ERROR, 3, False), diff --git a/tests/test_handler.py b/tests/test_handler.py index c068ffe..c3b9048 100644 --- a/tests/test_handler.py +++ b/tests/test_handler.py @@ -415,12 +415,13 @@ async def test_a_publish_failure_record_names_the_stored_output(h: Harness) -> N # --- the consumer's cap (#17) ----------------------------------------------------------- -async def test_publish_gave_up_publishes_one_terminal_extraction_error(h: Harness) -> None: +async def test_publish_gave_up_publishes_one_terminal_dead_lettered(h: Harness) -> None: + # The processor's fault, not the spec's: cannobserv#536's token, not extraction_error. command = h.message(store=False).payload await publish_gave_up(command, h.deps, "dead-lettered: gave up on attempt 3: " + "x" * 2000) fact = h.only_failure() - assert fact.reason == "extraction_error" + assert (fact.reason, fact.terminal) == ("dead_lettered", True) assert fact.occurred_at == NOW assert fact.detail.startswith("dead-lettered: gave up on attempt 3: x") assert len(fact.detail) == 1000 # the same cap as every failure fact diff --git a/tests/test_pin.py b/tests/test_pin.py index 404c624..49fad86 100644 --- a/tests/test_pin.py +++ b/tests/test_pin.py @@ -15,7 +15,7 @@ from processor.processors.extract import LOCAL_GENERATION, PROCESSOR_VERSION -EXPECTED = "0.32.0" +EXPECTED = "0.37.0" ROOT = Path(__file__).resolve().parent.parent PINNED = ("co-core", "co-core-aio", "co-core-sync") diff --git a/uv.lock b/uv.lock index ec105c5..78ba189 100644 --- a/uv.lock +++ b/uv.lock @@ -268,14 +268,14 @@ wheels = [ [[package]] name = "co-core" -version = "0.32.0" +version = "0.37.0" source = { registry = ".wheelhouse" } dependencies = [ { name = "dateparser" }, { name = "pydantic" }, ] wheels = [ - { path = "co_core-0.32.0-py3-none-any.whl" }, + { path = "co_core-0.37.0-py3-none-any.whl" }, ] [package.optional-dependencies] @@ -289,14 +289,14 @@ extract = [ [[package]] name = "co-core-aio" -version = "0.32.0" +version = "0.37.0" source = { registry = ".wheelhouse" } dependencies = [ { name = "co-core" }, { name = "httpx" }, ] wheels = [ - { path = "co_core_aio-0.32.0-py3-none-any.whl" }, + { path = "co_core_aio-0.37.0-py3-none-any.whl" }, ] [package.optional-dependencies] @@ -306,7 +306,7 @@ bus = [ [[package]] name = "co-core-sync" -version = "0.32.0" +version = "0.37.0" source = { registry = ".wheelhouse" } dependencies = [ { name = "co-core" }, @@ -314,7 +314,7 @@ dependencies = [ { name = "urllib3" }, ] wheels = [ - { path = "co_core_sync-0.32.0-py3-none-any.whl" }, + { path = "co_core_sync-0.37.0-py3-none-any.whl" }, ] [package.optional-dependencies] @@ -867,9 +867,9 @@ dev = [ [package.metadata] requires-dist = [ - { name = "co-core", extras = ["extract"], specifier = "==0.32.0" }, - { name = "co-core-aio", extras = ["bus"], specifier = "==0.32.0" }, - { name = "co-core-sync", extras = ["gcs"], specifier = "==0.32.0" }, + { name = "co-core", extras = ["extract"], specifier = "==0.37.0" }, + { name = "co-core-aio", extras = ["bus"], specifier = "==0.37.0" }, + { name = "co-core-sync", extras = ["gcs"], specifier = "==0.37.0" }, { name = "google-api-core", specifier = ">=2.27,<3" }, { name = "google-auth", specifier = ">=2,<3" }, { name = "google-cloud-storage", specifier = ">=3,<4" },