Plan: chain-v2 append transport — durable staging, per-shard drainer, FTI shard-order primitive (#1207, epic &73)

On this page
NOTE

Implements ADR-014 Amendment 5 C3 under the Amendment 6 corrections, with this plan’s own bindings ratified as Amendment 7. Parent rollout: ADR-014 chain-v2 (Step 5). Review state: v8 — full rework after an external review rejected v7 (~20 blocking + ~15 material findings; the §Appendix maps every finding to its landing section); prior history one architect round
four internal contextless rounds. All file:line verified against main == de4dcd86.

Status

Step Description Status

0

0a tracker reconciliation (#1207 ACs rewritten; #1205 gains the backlog-degrade AC; #1285 filed for the canon float boundary; #1279 go/no-go additions) → 0b this plan-commit MR (plan + nav + ADR-014 Amendment 7 + parent fixes) → 0c external review of the committed artifact → sign-off.

Done (2026-07-31) — 0a 2026-07-30; 0b MR !1049 merged (8d0e7214); 0c discharged by the in-plan-mode external review (the reviewer ran against the full plan text before approval; sign-off 2026-07-31)

1

MR-1 shared primitive (feature/1207-chain-append-primitive, Relates to #1207): canopy_common::chain_append (tx-typed single-shard primitive, refusal classifier, topology fetch + routing-version fence, FTI multi-shard carve-out with chunking) + ChainSource decode in canopy-chain + test-lib chain seeder + tanf/medicaid seam wiring (dormant) + suites.

Done (2026-07-31) — !1050 merged (impl 54970080, merge 86f04578); 144 unit + 4 seam + 13 integration tests green; full battery passed

2

MR-2 audit staging + drainer (feature/1207-audit-staging-drainer, Closes #1207): staging + dedup-index migration, consumer/ingest staging modes, bounded rotating router + drainer with savepoint poison isolation, backlog health, cargo xtask perf chain-drain + sustained-ingress evidence, docs.

Done (2026-07-31) — !1051 merged (impl 03686f11, OpenAPI snapshot 2fb6a79a, test fix 41bc2aa9, merge e236648c); 32 integration + config-domain unit tests green; full battery passed; D10 perf matrix: every 300/s cell holds backlog slope ≤ 0 (drainer keeps exact pace), 600/s cells put single-node capacity at ~440–490 chained/s

3

Post-merge docs close-out: close #1207 (closing comment, SHAs); parent Step 5 → Done; this plan → Archive + nav; throughput cells posted on #1279. (Docs-only, committed direct to main per git-workflow.)

Done (2026-07-31) — this commit; #1207 closed, cells posted on #1279

Epic &73
Issue #1207 (critical) — blocked by #1246 (Done) + #1236 (Done); blocks #1279
Branches feature/1207-append-transport-plan (this MR), then per the Status table

Local cargo xtask validate runs before each MR’s first push.

Step 0 — lifecycle (commit first, then review)

  1. 0a — tracker reconciliation (done 2026-07-30):

    • #1207’s AC checklist rewritten to match its own 2026-07-27 chain-v2 UPDATE: original AC2 ("checkpointed verify passes over batched appends") moved to #1205/#1206 (they own verification); original AC3 restated as sustained-ingress evidence (backlog slope ≤ 0 at ≥ 300 events/s, method in D10); the UPDATE’s pins became explicit ACs. "Nonzero backlog degrades status" stays OWNED BY #1205 — its description now carries that AC explicitly (wire the raw staged > 0 from this plan’s snapshot into the C6 machine).

    • #1285 filed: the canopy-chain float-boundary defect surfaced by external review (canon.rs:48 accepts every float AFTER integer checks — decimal/exponent source text rounds through f64 upstream of validation; adjudication + KATs). #1207 does not block on it: post-parse floats are deterministic end-to-end; the ambiguity is pre-parse.

    • #1279 gained: all-replica config/version attestation, startup capability probe, residual-backlog check, roll-forward/backout rehearsal, the v1-arm + seam-enum deletion as a separate post-cutover cleanup MR, and its stale parent-step reference noted.

  2. 0b — this plan-commit MR (Relates to #1207, docs-only): this page (nav-linked, Active — repo rules require the review-ready plan to be the in-repo artifact)
    ADR-014 Amendment 7 + parent plan Step 5 → In progress + the parent’s C4 epoch-state summary corrected to Amendment 6’s vocabulary + architecture.adoc ADR-index line
    CHANGELOG.adoc.

  3. 0c — external review runs against the COMMITTED artifact. Implementation MRs start only after it passes and the user signs off.

ADR-014 Amendment 7 content (short; the Amendment-6 precedent — contract-surface decisions a child makes are ratified formally before implementation):

  • Routing id BOUND: direct ingest routes on the server-minted envelope event_id; #-queue events on the publisher-minted envelope id. Both are the payload’s hashed event_id — hash-bound, immutable, equality structural (D1 CHECK).

  • Event uniqueness: one chained row per event_id per family table, enforced by a UNIQUE expression index on the live v2 table — the permanent replay identity (staging rows are dequeued; inbox rows expire at 7 days).

  • "Leased per-shard drainer" discharged structurally: ownership IS the claimed-row + head row locks inside one transaction — no lease table/TTL, so no steal/renew races; release on failure is bounded by the pinned transaction/session timeouts (D4), not wall-clock lease expiry. Satisfies Amendment 5’s drainer invariants (no overlapping ownership; O(batch) restart).

  • Ingest 202 redefinition ("accepted + durably staged", with an {event_id} receipt body), recorded with its dormancy gate (#1279 flips it).

Context (recon facts — verified file:line)

Contract (Amendment 5 C3 adr-014-fti-audit-hash-chain.adoc:445-458; Amendment 6 :574-695):

  • Writers chain from chain_heads via the shipped SECURITY DEFINER fns, never the newest live row; validate + canonicalize BEFORE locking.

  • Each shard in its OWN tx (head FOR UPDATE, contiguous seq, hash+insert+advance same tx); N per-shard. Only multi-head-one-tx path: the FTI determination carve-out (compute all shards first, lock heads ascending).

  • Direct ingest enqueues into the SAME durable staging transport (202 = durably staged); staged events inbox-stamped before ack; nonzero backlog must degrade status (#1205 owns the state machine — 0a made that explicit).

  • Sharding loses global order; no drain-order contract exists anywhere in this design.

Substrate surface (merged #1246, dormant):

  • chain_head_lock(family, epoch, shard) → (instance, last_seq, last_hash) — epoch FOR SHARE → head FOR UPDATE, re-checks state='active' AND epoch = current_epoch (security migration 20260910000000_chain_v2_substrate.sql:392-425; lock order shared with the future closer, comment :388-391).

  • chain_append_rows_audit/fti(epoch int, shard smallint, expected_routing_version smallint, rows jsonb[]) RETURNS bigint — batch 1..=500; baked source literal; per-row contiguity/watermark/interior-linkage/32-byte-hash/closed-payload-set/pinned-unhashed checks; single head UPDATE; refusals RAISE EXCEPTION 'chain-v2: …' (P0001) — but the INSERT’s derived-column casts ((v_payload→>'event_id')::uuid audit, security :555; (v_payload→>'id')::uuid fti, tanf :350; timestamps both) can ALSO raise SQLSTATE class 22 on poison (D5 handles both). Row element: {chain_seq, previous_hash(hex), event_hash(hex), canonical_event_payload, unhashed} (audit unhashed ⊆ {ip_address}; fti ⊆ {request_id,ip_address,success}).

  • Proven caller flow + jsonb[] binding (services/canopy-tanf/tests/ chain_v2_substrate_test.rs:623-750): head-lock → hash in Rust → SELECT chain_append_rows_fti($1,$2,$3,(SELECT COALESCE(array_agg(e ORDER BY ord), ARRAY[]::jsonb[]) FROM jsonb_array_elements($4) WITH ORDINALITY AS t(e,ord))). Eight concurrent writers serialize on the head lock first-try; shards independent.

  • Discovery (SELECT to _app): chain_topology, chain_epochs, chain_sources, chain_status_v. No direct chain_heads SELECT.

  • crates/canopy-chain: payload builders (build() → Result<Value, ChainError> — the I-JSON walk), ChainEnvelope + event_hash, shard_for (algorithm v1 — the ONLY compiled routing), newtypes, versions::{EVENT_HASH_FORMULA_VERSION=2, ROUTING_VERSION=1}.

v1 write paths (stay production until #1279):

  • Security ingress A: MQ # subscriber (main.rs:57-71, durable, max_attempts=5→DLQ; the handler deliberately ignores the inbox tx — #456 comment main.rs:45-54, to be rewritten as a mode split; the #1094 block :36-44 stays) → parse_event → insert_audit_event (store/mod.rs:83-175: own tx, pg_advisory_xact_lock(1)). Newer-schema envelopes PARK in the inbox (subscriber.rs:923) — a quarantine #1205’s status must surface (0a note). Inbox rows are NOT permanent: processed rows are reaped after 7 days (crates/canopy-mq/src/inbox_drainer.rs:3), and admin replay skips processed rows (replay.rs:97) — hence the chain-side dedup index (D1).

  • Security ingress B: POST /v1/security/audit/ingest (api/mod.rs:255-288), server-mints EventEnvelopeId::new() at :265, 202.

  • Field bridge: ParsedAuditEvent (event_parsing.rs:12-35) → AuditPayloadBuilder 1:1; household_id string→UUID-or-NULL (store/mod.rs:105-110); ip_address = the audit unhashed key; oversize-integer metadata fails build() → retry→DLQ (fail-closed).

  • FTI: insert_fti_chain_entry (fti_audit.rs:334-387, constant advisory lock, single row); call sites tanf determine.rs:918, medicaid determine.rs:829-836 (N-entry loop inside one tx, followed by a trailing status UPDATE — the v1 append is NOT last today). FTI routing id = caller-minted row id (FtiAuditEntryId, v7; Amendment 6 field table). A retried determination mints fresh ids — FTI has no replay-identity problem.

  • Precedents: finalize_saga_enabled config gating (services/canopy-applications/src/config.rs:282-311); OutboxDrainer health shape (outbox_drainer.rs:328-448); service lib target for test imports (services/canopy-applications/src/lib.rs); claim-index-must-match-claim-query (20260815000000_event_outbox_claim_order_idx.sql:4-17); perf harness home = xtask (xtask/src/cmd/perf.rs; standards:testing.adoc — NOT #[ignore]`d nextest tests, per `testing-discipline).

D1 — staging store + permanent replay identity (canopy_security only; FTI stages nothing)

One new migration (20260930000000_chain_append_staging.sql; then touch crates/canopy-test-lib/src/db.rs). Three objects:

(a) chain_append_staging — a queue, NOT chained data; ownership stays with the migration identity (never canopy_chain_owner_security):

CREATE TABLE chain_append_staging (
    event_id     UUID PRIMARY KEY CHECK (uuid_extract_version(event_id) = 7),
    canonical_event_payload JSONB NOT NULL
        CHECK (jsonb_typeof(canonical_event_payload) = 'object'),
    payload_digest BYTEA NOT NULL CHECK (octet_length(payload_digest) = 32),
    unhashed     JSONB NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(unhashed) = 'object'),
    chain_instance_id UUID,                -- router-stamped, all three together
    chain_epoch  INT  CHECK (chain_epoch >= 0),
    shard_id     SMALLINT CHECK (shard_id >= 0),
    attempts     INT NOT NULL DEFAULT 0 CHECK (attempts >= 0),
    parked_at    TIMESTAMPTZ,
    park_reason  TEXT,
    staged_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    -- routing id = hashed id, STRUCTURAL:
    CHECK ((canonical_event_payload->>'event_id') = event_id::text),
    -- stamps travel together; (epoch, NULL shard) unrepresentable:
    CHECK ((chain_instance_id IS NULL) = (chain_epoch IS NULL)
       AND (chain_epoch IS NULL) = (shard_id IS NULL)),
    CHECK ((parked_at IS NULL) = (park_reason IS NULL))
);
CREATE INDEX chain_append_staging_claim_idx
    ON chain_append_staging (chain_instance_id, chain_epoch, shard_id, staged_at, event_id)
    WHERE parked_at IS NULL;
CREATE INDEX chain_append_staging_unrouted_idx
    ON chain_append_staging (staged_at, event_id)          -- matches the router ORDER BY
    WHERE parked_at IS NULL AND shard_id IS NULL;
CREATE INDEX chain_append_staging_parked_idx
    ON chain_append_staging (parked_at) WHERE parked_at IS NOT NULL;  -- stats/ops

payload_digest = SHA-256 of the canonical bytes, computed at build time (already produced for validation) — the divergent-replay discriminator.

(b) Restricted grants + guarded dequeue (accepted audit data must not be freely mutable/deletable by the runtime):

GRANT SELECT, INSERT ON chain_append_staging TO canopy_security_app;
GRANT UPDATE (chain_instance_id, chain_epoch, shard_id, attempts, parked_at, park_reason)
    ON chain_append_staging TO canopy_security_app;      -- routing/park state ONLY
-- NO DELETE grant. Dequeue is a SECURITY DEFINER fn owned by the migration identity:
--   chain_staging_dequeue(p_event_ids uuid[]) RETURNS bigint  (DELETE … = ANY; returns count)
-- EXECUTE to canopy_security_app. payload/digest/unhashed/staged_at/event_id are
-- IMMUTABLE to the runtime by column-grant omission.

Unpark is an OPERATOR action under the maintenance/migration credential — a pinned runbook procedure in security-operations.adoc (inspect park_reason, record a ticket reference, UPDATE … SET parked_at = NULL, park_reason = NULL, attempts = 0), never a runtime API. Parked rows are an auditor-visible quarantine, gauged and queryable.

(c) Permanent replay identity — chain-side UNIQUE dedup index (same migration; external review proved staging-DELETE + 7-day inbox retention leaves NO durable replay identity, so a >7-day-late redelivery would double-chain):

CREATE UNIQUE INDEX audit_events_v2_event_id_uq
    ON audit_events_v2 (((canonical_event_payload->>'event_id')));

One chained row per event_id, permanent, pruned WITH the data (no unbounded receipt table). The drainer’s prevalidation (D4) consults it BEFORE locking: already-chained + equal digest → silently dequeue (late exact replay = no-op); already-chained + different digest → park fail-closed (divergent replay, observable). The index also backstops as a 23505 during append. Residual, documented: after #1208 archival moves rows out of the live table (years later) the live index no longer covers them — replay horizon is days, archive horizon is years; #1208 gains an AC note. Applying an index to a dormant-but-chain-owned table is flagged in Open decisions.

Staging lifecycle: INSERT → stamp → dequeue-in-append-tx (exactly-once handoff; no janitor — only parked rows persist). Admission bound: CHAIN_STAGING_MAX_DEPTH (default 500_000); the consumer checks the sampled depth atomic before staging and errors (nack → requeue; the durable broker remains the overflow home, exactly as today) with a debounced warn; ingest returns 503. Payload size is bounded by the existing axum body limit on the ingest route and broker message limits on the bus path (documented, not new machinery).

D2 — ingress modes (dormant behind config)

Pinned signatures (security-local, chain_staging/stage.rs):

pub struct StagedAuditRow {
    pub event_id: Uuid,                  // staging PK, carried explicitly
    pub payload: serde_json::Value,      // built closed-set 11-key payload
    pub payload_digest: [u8; 32],        // SHA-256 of canonical bytes (computed in build)
    pub unhashed: serde_json::Value,     // {"ip_address": …} or {}
}
/// event_id from parsed.event_id (envelope id on the bus path; freshly-minted id on the
/// ingest path). Payload via AuditPayloadBuilder (household parse-or-NULL mirroring v1;
/// metadata null→{}). Digest from the canonical bytes build() already produces.
pub fn build_staged_audit_row(parsed: &ParsedAuditEvent)
    -> Result<StagedAuditRow, canopy_chain::ChainError>;

/// INSERT … ON CONFLICT (event_id) DO NOTHING, then, if conflicted, SELECT the existing
/// payload_digest: equal → Ok(Staged::Duplicate) (idempotent replay); different →
/// Err(StageError::DivergentReplay) — FAIL CLOSED, observable. Takes &mut PgConnection
/// (the two-statement body reborrows &mut *conn): the consumer passes &mut **tx (inbox
/// tx — atomicity is the point); the ingest handler acquires first
/// (`let mut conn = state.db.inner().acquire().await?`). An executor generic can't serve
/// both statements on the tx path (a &mut connection isn't Copy).
pub async fn stage_audit_event(
    conn: &mut sqlx::PgConnection, row: &StagedAuditRow) -> Result<Staged, StageError>;

  • Consumer (flag off = v1 byte-for-byte): flag on → parse_event → build_staged_audit_row (Err → nack ×5 → DLQ; poison filtered at intake) → depth check (D1 admission bound) → stage_audit_event(&mut **tx, …) on the inbox tx — commit stamps inbox + stages atomically, then ack. That is "durably staged + inbox-stamped before ack" verbatim. DivergentReplay → error → retry ×5 → DLQ + error! (never silently dropped). Rewrite the main.rs:45-54 #456 comment as the mode split.

  • Ingest (flag off = v1 sync append): flag on → mint EventEnvelopeId::new() (the Amendment-7 binding) → build → stage via an acquired connection → 202 with body {"event_id": …} — the receipt; semantics documented on the api page: a client retry WITHOUT its own idempotency key mints a new event (intentional — replay identity belongs to the envelope id, not the body). Build failure → ApiError::UnprocessableEntity ONLY for the client-traceable I-JSON class (ChainError::UnsafeInteger); any other builder failure is a 500. utoipa responses(…) gains 422 (J5). Flag threading: the existing SecurityConfig-in-state pattern.

  • CHANGELOG Changed: 202 redefinition, dormant until #1279.

D3 — router (bounded; shard stamping as a drainer sub-step)

Staged rows arrive unstamped. Per pass, AT MOST CHAIN_ROUTE_MAX_BATCHES iterations (unbounded "route until short" starves draining under sustained ingress), each one tx:

  • Claim WHERE parked_at IS NULL AND shard_id IS NULL ORDER BY staged_at, event_id LIMIT $w FOR UPDATE SKIP LOCKED ($w = CHAIN_ROUTE_BATCH_SIZE) — an EXACT match for chain_append_staging_unrouted_idx including the ORDER BY (the claim predicate must imply the partial-index predicate; this repo already paid for violating that once — 20260815000000_event_outbox_claim_order_idx.sql:4-17).

  • Compute shard_for(event_id, shard_count) in Rust; one batched UPDATE stamping ALL THREE: SET chain_instance_id = $i, chain_epoch = $e, shard_id = u.shard (instance from this pass’s fetch_topology — an instance reset must not consume stale stamps; the D1 CHECK makes partial stamps unrepresentable); commit.

Why not stage-time stamping: the ack path must never wait on topology discovery, and restamp machinery would be needed for discovery-gap rows regardless. Replica races are benign (deterministic values); SKIP LOCKED prevents blocking. Rollover/reset restamping is #1280/#1279 mechanics — preconditions there (quiesce + drained staging) mean stale-stamp rows cannot exist in #1207’s steady state; if they somehow did, the drainer’s exact-stamp claim never touches them and they sit visibly in the backlog gauges.

D4 — per-shard drain (bounded, rotating, validate-before-lock, savepoint isolation)

Scheduler (fixed-order drain-to-empty can starve shard N and never yield): per pass — pre-flight fetch_topology (Missing/NotActive → debounced warn, sleep) → router (≤ R batches) → visit shards starting at offset pass_counter % shard_count, rotating; each shard gets AT MOST CHAIN_DRAIN_MAX_BATCHES_PER_SHARD drain txs per pass. Remaining backlog waits for the next tick (250ms) — bounded latency, no starvation. The REAL pass loop (drainer_pass) is a testable function.

Every drain/route tx opens with SET LOCAL lock_timeout = $L; SET LOCAL statement_timeout = $S (config, D8) and the pool sets idle_in_transaction_session_timeout — PG releases a wedged/partitioned session’s locks only when the server notices, so release is BOUNDED by these timeouts (the wedge test proves it); Amendment 7’s lease wording says exactly that.

drain_shard = ONE tx:

  1. Claim: SELECT event_id, canonical_event_payload, payload_digest, unhashed FROM chain_append_staging WHERE chain_instance_id = $1 AND chain_epoch = $2 AND shard_id = $3 AND parked_at IS NULL ORDER BY staged_at, event_id LIMIT $n FOR UPDATE SKIP LOCKED.

  2. Prevalidate BEFORE the head lock (C3 validate-before-lock): per row — canonical_bytes(payload) re-validates; derived-cast prevalidation (every field the SQL INSERT casts: uuid/timestamp/unhashed types) so deterministic 22xxx cast poison never reaches the append; digest recomputed == stored digest. Failures → in-tx UPDATE … SET parked_at/park_reason/attempts (no exception, claim locks held — no separate-tx race). Chain-dedup probe: SELECT the already-chained matches among the claim via the dedup index — the index serves the LOOKUP only (the chain table carries no digest column); the probe fetches each matched row’s canonical_event_payload and recomputes its digest (the same canonicalize-then-SHA the C2 semantics guarantee deterministic). Equal-digest hits → mark for silent dequeue (late exact replay = no-op); different-digest → park (divergent replay, fail closed).

  3. Append the survivors via the MR-1 primitive on THIS tx: SAVEPOINT batch; append_rows_on_shard(tx, Audit, &target, shard, &rows). On Refused{Row}/cast-class (rare — prevalidated): ROLLBACK TO batch, then per-row SAVEPOINT r; append one; on refusal ROLLBACK TO r + park in-tx (a PG exception aborts the tx — savepoints are the only valid isolation; row/head locks survive ROLLBACK TO SAVEPOINT, so claims and serialization hold). ONE pinned special case: on 23505 against the dedup index (a replay that raced past the probe), re-run the digest-compare — equal → silent-dequeue set; different → park. On Environment/Invariant: rollback everything, dispatch per D5. Inside the primitive: chain_head_lock → instance fence (lock-returned instance must equal target.instance, else Invariant) → hash in Rust from the locked head (hash exactly the claim-returned JSONB values — the C2 refetch rule) → chain_append_rows_audit(epoch, shard, ROUTING_VERSION, rows) — the COMPILED constant (D7 fence).

  4. Count checks: appended == surviving rows.len(), else Invariant + rollback.

  5. Dequeue: chain_staging_dequeue($appended ++ $silent_dedup); returned count must equal the list length, else Invariant + rollback.

  6. COMMIT — append + head advance + parks + dequeue atomic; any failure rolls back the whole segment; rows stay staged.

Deadlock-freedom: SKIP LOCKED never waits on staging rows; head acquisition uses `chain_head_lock’s pinned epoch→head order; one head per tx; same-shard replicas pipeline (hashes computed only after the lock returns the post-commit head — the substrate’s 8-writer proof).

Spawn: ChainDrainer::spawn(pool, cfg) (OutboxDrainer shape, homed in the service), from main.rs, only when the flag is on. The stats sampler runs ALWAYS (flag-off included — dormancy must not hide residual staging after a failed trial). Test importability: canopy-security is bin-only — MR-2 adds src/lib.rs (the canopy-applications precedent) exposing pub mod chain_staging; main.rs consumes the lib. Ingest-path tests follow the existing tests/audit_ingest_test.rs harness.

D5 — refusal classification + dispatch

classify_refusal in canopy_common::chain_append takes the SQLSTATE and message (string matching alone is inadequate — the append INSERT can raise deterministic cast errors):

pub fn classify_refusal(sqlstate: &str, message: &str) -> Option<RefusalClass>;

  • SQLSTATE class 22 (data exception) or 23 raised BY an append statement → Row (deterministic poison; the 23505 dedup special case is pinned in D4). Prevalidation makes these near-unreachable; classification is the belt.

  • P0001 with chain-v2: ` prefix → the matcher table below. `classify_refusal receives the FORMATTED runtime message (PG substitutes every % before sqlx surfaces it); THREE messages carry MID-message parameters, matched by contains on a parameter-free fragment; the rest by starts_with on the stem after the prefix.

Class starts_with stems contains fragments (mid-param)

Row (the row’s fault)

payload keys diverge from the closed (one stem, both families) · non-string required payload field · optional payload field must be string or null (audit-only) · unknown unhashed ingress key · data_elements_accessed (both fti literals :312/:320) · resource_id must be string or null (fti :331)

—

Environment (no row involved)

no audit topology · no fti topology · no topology for family · routing_version mismatch

not the active current epoch (:412) · not registered for this instance (:471)

Invariant (caller bug / substrate divergence)

non-contiguous seq · interior linkage break · empty batch · batch exceeds max 500 · event_hash not 32 bytes (caller-computed — a firing means caller divergence; parking would quarantine an innocent row)

at or below archived_through (:499)

(9 distinct per-row-check literal texts across both fn bodies — audit 5, three shared verbatim with fti, fti 4 unique; 8 → Row, event_hash not 32 bytes → Invariant.)

  • None fail-safe: chain-v2: `-prefixed P0001 with no table match → `Invariant (loud, never park, never retry-hot). Everything else → ChainAppendError::Database (transient: warn + capped backoff 250ms→5s).

Refused carries {class, sqlstate, message} (don’t discard the SQLSTATE).

Dispatch (per call-site; the class names the fault domain). Drainer, exhaustive over ChainAppendError:

  • Refused{Row} → savepoint isolation + in-tx park (D4); single-shot (deterministic — retrying is theater).

  • Refused{Environment} → idle-skip the pass, debounced warn.

  • Refused{Invariant} | UnsupportedFamily → rollback, loud error!, capped backoff, never park.

  • Chain(e) during PER-ROW hashing → that row’s fault (the drainer knows which row) → park it; Chain(e) anywhere else → Invariant treatment.

  • Database → warn + backoff; rows released by rollback.

FTI seam: no staging, never parks — every Refused propagates and aborts the caller’s determination tx (fail-closed).

D6 — backlog primitives (the #1205 hand-off surface)

ChainStagingHealth: atomics {staged, oldest_staged_age_secs, parked, last_sample_at, last_pass_at, sample_errors}; thresholds pre-validated by ChainDrainConfig::from_config (D8 startup-error contract, not OutboxHealth’s env-clamp). snapshot() → ChainStagingSnapshot:

  • staged raw count — the input #1205’s C6 "nonzero degrades" wiring consumes (its AC as of 0a).

  • degraded_reason: parked > 0 · depth/age over threshold · stale sample (now − last_sample_at > 3× interval, or sample_errors climbing — one good sample must not leave stale-green) · drainer stalled (flag on and now − last_pass_at > 10× tick).

  • Sampler runs ALWAYS (even dormant — sampled:false only before the first sample ever); one aggregate every 30s: SELECT count() FILTER (WHERE parked_at IS NULL), count() FILTER (WHERE parked_at IS NOT NULL), min(staged_at) FILTER (WHERE parked_at IS NULL) FROM chain_append_staging (bounded by the admission cap; parked_idx covers ops queries; autovacuum churn expectation documented in data-models).

  • Inbox parks (subscriber.rs:923) and DLQ depth never enter staging — recorded at 0a as #1205 status inputs, NOT re-implemented here.

D7 — the shared primitive + FTI carve-out (MR-1)

Home crates/canopy-common/src/chain_append.rs (canopy-common gains the planned canopy-chain runtime dep; no cycle). MR-1 also adds ONE tiny additive canopy-chain item: ChainSource decode (DB text → enum; none exists today). Module doc pins the refetched-value hashing rule (audit callers pass PG-refetched values; FTI builder output is normalization-stable by construction — strings/arrays/null only, no floats — which the substrate’s refetch-recompute tests already prove).

pub enum RefusalClass { Row, Environment, Invariant }

#[derive(Debug, thiserror::Error)]
pub enum ChainAppendError {
    Database(#[from] sqlx::Error),
    Chain(#[from] canopy_chain::ChainError),
    Refused { class: RefusalClass, sqlstate: String, message: String },
    UnsupportedFamily { family: ChainFamily },   // EleGrant until #1248 — typed, no panic
}

pub fn classify_refusal(sqlstate: &str, message: &str) -> Option<RefusalClass>;  // D5

/// HTTP mapping for the FTI seam. Environment → 503 with a FIXED public string
/// ("audit chain unavailable" — ApiError::ServiceUnavailable exposes its message,
/// error.rs:136, so detail goes to logs only). Row/Invariant/Chain/UnsupportedFamily →
/// 500. NEVER 422 here (422 is the ingest endpoint's client-data contract, D2).
impl From<ChainAppendError> for ApiError { … }

pub struct StagedChainRow { pub payload: serde_json::Value, pub unhashed: serde_json::Value }
pub struct ShardAppendOutcome { pub appended: u32, pub last_seq: i64, pub last_hash: EventHash }

/// TX-TYPED (a bare connection would autocommit lock/append separately — the C3 unit is
/// a transaction). Head-lock → instance fence (lock-returned instance == target.instance,
/// else Refused{Invariant}) → envelope + event_hash per row (interior linkage from the
/// locked head) → chain_append_rows_{family} passing the COMPILED
/// versions::ROUTING_VERSION (never the fetched DB value — see fetch_topology). Envelope
/// source = target.source (registry-derived; a caller-constant mismatch class is
/// unrepresentable). Returned count checked == rows.len(). Never partial.
pub async fn append_rows_on_shard(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, family: ChainFamily,
    target: &ActiveTopology, shard: ShardId, rows: &[StagedChainRow],
) -> Result<ShardAppendOutcome, ChainAppendError>;

pub struct ActiveTopology {
    pub instance: ChainInstanceId, pub epoch: Epoch,
    pub shard_count: ShardCount,   pub source: ChainSource,   // registry-derived
    // no routing_version field — see the fence below
}
pub enum TopologyStatus { Missing, NotActive { state: String }, Active(ActiveTopology) }

/// Discovery: chain_topology ⋈ chain_epochs ⋈ chain_sources. Fetch, no cache; call on the
/// SAME tx for in-tx fence consistency (FTI), or per pass (drainer).
/// ROUTING-VERSION FENCE (the naive fence is circular — passing the fetched value back to
/// SQL validates nothing): the binary compiles exactly ONE routing algorithm
/// (versions::ROUTING_VERSION = 1). fetch_topology COMPARES the DB's
/// chain_epochs.routing_version against the compiled constant and returns
/// Refused{Environment, "routing version skew: db=…, binary=…"} on mismatch — a v2-DB
/// deployment can never accept placements computed by a v1 binary. The append then passes
/// the compiled constant, and the SQL fence re-checks in-lock.
/// chain_sources: exactly one row per (family, instance); zero/multiple ⇒
/// Refused{Environment} (fail-closed).
pub async fn fetch_topology<'e, E: sqlx::PgExecutor<'e>>(executor: E, family: ChainFamily)
    -> Result<TopologyStatus, ChainAppendError>;

pub struct FtiChainAccessV2 { /* id, accessed_by, accessed_at, purpose_code,
    data_elements_accessed, originating_system, action, resource_type, resource_id,
    request_id, ip_address, success */ }
impl From<&FtiAuditEntry> for FtiChainAccessV2 { … }

/// Build + group, fallible (build() returns Result): validates ALL payloads FIRST (C3),
/// computes shard_for(entry.id), returns BTreeMap — ascending iteration IS the C3
/// lock-order rule. Proptest surface.
pub fn build_fti_shard_rows(entries: &[FtiChainAccessV2], count: ShardCount)
    -> Result<BTreeMap<ShardId, Vec<StagedChainRow>>, ChainAppendError>;

/// The C3 FTI carve-out, tx-typed: fetch_topology on the tx (must be Active — a
/// determination cannot proceed without its Pub-1075 chain row; fail-closed);
/// build_fti_shard_rows; per shard ASCENDING: append_rows_on_shard, CHUNKED at 500 (a
/// shard group may exceed the SQL max) — sequential chunks inside the same tx, linkage
/// continuing from each ShardAppendOutcome.last_hash. Call LAST before commit.
pub async fn append_fti_entries_v2(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, entries: &[FtiChainAccessV2],
) -> Result<(), ChainAppendError>;

Call-site wiring (dormant seam). Per service: config chain_v2_append_enabled (default false, # DORMANT until #1279 yaml comment); new src/fti_chain.rs:

#[derive(Debug, Clone, Copy)]                 // Copy: Axum Extension requires Clone
pub enum ChainAppendMode { V1Advisory, V2Sharded }
pub async fn append_determination_chain_entries(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, mode: ChainAppendMode,
    entries: &[FtiAuditEntry],
) -> Result<(), ApiError>;

  • The seam sits at TODAY’S call position (tanf :918; medicaid :829-836) so the V1 arm is genuinely byte-identical (a reorder would change v1 lock timing). Consequence, documented: medicaid’s V2 arm holds its head locks across the one trailing single-row status UPDATE — bounded (no second head, no other locks), deadlock-free, accepted; v1’s own position already violated the fti_audit "call LAST" doc the same way.

  • Mode threads handler → determine → (medicaid) persist_determinations. Lint reality (verified): both determine`s carry `#[expect(clippy::too_many_arguments)] (still covered); persist_determinations (6→7 params) stays AT the threshold — do NOT add an expect (it would trip unfulfilled_lint_expectations under -D warnings).

  • Medicaid collects its loop-built entries into a Vec for the one seam call; both services' load-bearing advisory-lock comments (tanf :884-885, medicaid :827-828) are rewritten for the mode split (J5).

  • Seam arms are unit-tested IN-PLACE (#[cfg(test)] in src/fti_chain.rs — bin crates run src unit tests; no tanf/medicaid lib target needed). The heavy integration suites drive canopy_common::chain_append + the SQL directly (all importable).

  • Deadlock proof obligation: two concurrent multi-shard txs, opposite natural orders, both commit — the primitive’s ascending sort is the only reason.

Test-lib seeder (crates/canopy-test-lib/src/chain.rs, + pure canopy-chain dep): SeededChain { instance, family, source, shard_count } with METHOD genesis_hash(&self, shard) → EventHash (needs instance/family/epoch — a free fn can’t)
seed_chain_v2(pool, family, source, shard_count, activate). Pays off in #1205/#1206/#1208/#1279.

D8 — configuration

Env var Default Domain (out-of-domain = STARTUP ERROR)

CANOPY_SECURITY__CHAIN_V2_APPEND_ENABLED

false

bool

CANOPY_SECURITY__CHAIN_DRAIN_BATCH_SIZE

500

1..=500 (SQL fn max; substrate pinned "revisited by #1207 with throughput evidence")

CANOPY_SECURITY__CHAIN_DRAIN_TICK_MS

250

10..=60000

CANOPY_SECURITY__CHAIN_DRAIN_MAX_BATCHES_PER_SHARD

4

1..=64

CANOPY_SECURITY__CHAIN_ROUTE_BATCH_SIZE

1024

1..=10000

CANOPY_SECURITY__CHAIN_ROUTE_MAX_BATCHES

4

1..=64

CANOPY_SECURITY__CHAIN_DRAIN_LOCK_TIMEOUT_MS

5000

100..=60000

CANOPY_SECURITY__CHAIN_DRAIN_STATEMENT_TIMEOUT_MS

30000

1000..=300000

CANOPY_SECURITY__CHAIN_STAGING_MAX_DEPTH

500000

≥1000

CANOPY_SECURITY__CHAIN_STAGING_ALERT_DEPTH

10000

≥1

CANOPY_SECURITY__CHAIN_STAGING_ALERT_AGE_SECS

300

≥1

CANOPY_TANF__CHAIN_V2_APPEND_ENABLED

false

bool

CANOPY_MEDICAID__CHAIN_V2_APPEND_ENABLED

false

bool

Validated by ChainDrainConfig::from_config (the FinalizeSagaConfig pattern), unit-tested. AC pins recorded in configuration-reference: dwell/flush (partial batch appends next tick; per-shard work bounded per pass; no shutdown flush — staging is durable, next boot resumes in O(one claim)); park policy (classification-driven, single-shot, never a numeric threshold); prefetch-32 relationship (prefetch bounds staging INGRESS in-flight per consumer; drain batching is independent; imbalance accumulates in staging where it is measured — with the admission cap bounding the DB and pushing true overflow back to the durable broker, which is where it lives today).

D9 — dormancy + the #1279 handshake

Staged rollout mandated by Amendment 6, NOT back-compat: at cutover direct DML becomes forbidden, so v2 writers must exist dormant BEFORE it, and v1 must keep writing until it. No dual-format readers, no legacy formula, no shims; the v1 arms die in #1279’s cleanup MR.

  • Flag off (default): v1 paths byte-identical (regression-tested); no drainer; staging empty; the sampler STILL runs (residue from any aborted trial stays visible).

  • #1279 go-live: flip three config keys + activate epochs + swap identities — no code deploy. Startup capability probe, all-replica config attestation, and backout rehearsal are #1279 go/no-go items (added at 0a).

  • Pre-activation window (flag on, epoch installing): audit path safe by construction (ingestion stages, router can’t stamp, drainer idle-skips); FTI V2 arm FAILS CLOSED (determination aborts, 503) — acceptable only because #1279 is quiesced downtime; the runbook orders tanf/medicaid flips after epoch activation.

D10 — throughput evidence (xtask perf harness; feeds #1279)

The [ignore]`d-test approach is out — `testing-discipline bans [ignore] and the perf home is the xtask harness. New arm: cargo xtask perf chain-drain (release build, devstack Postgres):

  • Seeds a dedicated schema via seed_chain_v2 (activated epoch), then runs a SUSTAINED producer (staging inserts through the real build_staged_audit_row/stage_audit_event path at a target rate) concurrently with real drainer_pass workers — measuring the ACTUAL AC: steady-state chained/s and backlog slope over the window, not drain-of-a-finite-pile.

  • Matrix: shard_count ∈ {2, 8} × workers ∈ {1, 2} (worker = one sequential-shard drainer = one replica; labeled as such) × sustained rate ∈ {300, 600}/s. Pinned method: 30s warmup, 120s measurement window, 3 repetitions, payload mix 1KiB typical / 64KiB p99, pool = 10, report mean ± spread per cell.

  • Pass = backlog slope ≤ 0 at 300/s in every cell (the AC’s "sustained ingest ≥ hundreds of events/s without divergence"); print all cells. MR description records numbers + method
    exact command; Step 3 posts them on #1279 (shard-count effect separable from worker count by the labeling).

  • In-battery tests stay correctness-only (no wall-clock asserts).

Scope — explicitly OUT

  • C6 status machine / DTOs / citation coverage → #1205/#1206 (0a added the backlog-wiring AC to #1205; this plan ships the primitives).

  • Epoch closure/rollover + stale-epoch/instance restamp queries + their indexes → #1280.

  • Archive/purge → #1208/#1247 (0a adds the dedup-index replay-horizon note to #1208).

  • ele_grant → #1248 (UnsupportedFamily, typed).

  • Cutover execution, identity swap, v1 deletion + seam-enum cleanup MR, binding shard-count, attestation/backout → #1279. Anchor authority → #1278.

  • canopy-chain float-boundary defect → #1285.

Verification (test inventory — mapped to invariants and the review’s failure modes)

canopy-common unit: classify_refusal_pins_the_shipped_message_set (every distinct literal from both fn bodies as FORMATTED messages — sample values substituted for every %, incl. the three mid-param cases — plus SQLSTATE-22 and 23505 arms; unknown → None); unclassified_chain_v2_refusal_fails_safe_as_invariant; build_fti_shard_rows_sorts_ascending_and_partitions + proptest …_partition_is_total_and_stable; fti_access_maps_unhashed_ingress_exactly; chain_append_error_maps_to_api_status (Environment→503 fixed redacted string, everything else 500, never 422); ele_grant_family_returns_unsupported_family; fti_chunking_splits_over_500_preserving_linkage.

tanf/medicaid tests/chain_v2_append_test.rs (canopy-common + canopy-chain + test-lib only — no service imports; seam arms unit-tested in src): fti_v2_single_entry_appends_and_refetch_recompute_verifies; fti_v2_multi_shard_opposite_arrival_orders_do_not_deadlock; fti_v2_no_active_epoch_refuses_and_aborts_tx; fetch_topology_derives_source_validates_routing_version_and_refuses_absent_registry (incl. DB routing_version bumped ⇒ Refused{Environment} — the compiled-constant fence); fti_v2_multi_shard_refusal_is_atomic_across_shards (corrupt rows injected on one shard via a direct append_rows_on_shard call with a hand-built StagedChainRow); medicaid adds fti_v2_n_entries_one_tx_groups_across_shards and fti_v2_shard_group_over_500_chunks_in_order.

security tests/chain_append_staging_test.rs: stage_and_inbox_stamp_commit_atomically_before_ack; redelivery_is_single_staging_row; divergent_replay_same_id_fails_closed (staged variant AND already-chained variant); late_replay_after_chaining_dequeues_silently (the dedup-index path); admission_cap_nacks_consumer_and_503s_ingest; oversize_integer_metadata_errs_for_dlq_and_stages_nothing; ingest_v2_returns_event_id_receipt_and_does_not_chain_synchronously; ingest_v2_unsafe_integer_is_422_other_builder_errors_500; flag_off_paths_are_byte_identical_v1; household_id_non_uuid_drops_to_null_in_payload; staging_ddl_rejects_partial_stamps_and_id_mismatch.

security tests/chain_append_drainer_test.rs (driving the real drainer_pass/drain_shard via the new lib target): router_stamps_instance_epoch_shard_matching_shard_for; drain_appends_verifies_dequeues_and_counts_match; drain_batches_n_rows_per_head_lock; float_metadata_survives_refetch_recanonicalize; mid_drain_crash_commits_no_partial_segment_and_resumes; wedged_session_releases_within_timeouts (second session holds a claim, pg_terminate_backend + timeout budget — the bounded-liveness proof); shard_drains_proceed_while_another_head_is_held; concurrent_drainers_same_shard_pipeline_without_forks; sustained_staging_does_not_starve_draining_and_all_shards_advance (the scheduler test — continuous producer; every shard’s head advances; router work bounded per pass); savepoint_poison_isolation_parks_exactly_the_offender_under_concurrency (poison via direct SQL among healthy rows, TWO drainers running — claims retained across ROLLBACK TO SAVEPOINT); cast_poison_22xxx_parks_not_hot_loops; preflight_not_active_idle_skips_without_parking (superuser UPDATE chain_epochs SET state='installing'; no reverse fn exists); drainer_environment_dispatch_never_parks (flip state after pre-flight, call drain_shard with the stale target); invariant_dispatch_is_loud_and_rowsafe (dispatch-level: constructed Refused{Invariant} + count-mismatch arm — a "corrupt head ⇒ linkage refusal" test is WRONG: the primitive chains FROM whatever head the lock returns); instance_fence_refuses_foreign_instance; staging_grants_match_the_matrix (post-cutover _app shape: INSERT ok, payload UPDATE denied, DELETE denied, dequeue fn EXECUTE ok); stale_sampler_degrades_snapshot; staging_stats_and_health_degrade_and_recover; unpark_runbook_procedure_restores_row (maintenance-credential UPDATE per the runbook).

Gates, both MRs: full local battery before first push; SPDX; no unwrap/expect outside tests; proptest on grouping math. Quality budgets — B3a (production serde_json::Value, zero-headroom blocking ratchet; canopy-common/-security NOT exempt): every new qualified occurrence carries // STRUCTURAL-VALUE: <reason> (the canon.rs:15 precedent — these fields ARE structurally-untyped canonical JSON), B3a lock delta stated in the MR description; B3b for test imports.

Files touched (by MR)

0b (this MR): this page + nav.adoc + ADR-014 Amendment 7 + parent plan (Step 5
epoch-state text) + architecture.adoc ADR index + CHANGELOG.adoc.

MR-1: crates/canopy-chain/src/types.rs (ChainSource decode — additive); crates/canopy-common/{src/chain_append.rs (new), src/lib.rs, Cargo.toml}; crates/canopy-test-lib/{src/chain.rs (new), src/lib.rs, Cargo.toml}; services/canopy-tanf/{src/fti_chain.rs (new), src/config.rs, src/main.rs, src/api/handlers.rs, src/determine.rs} + config/canopy-tanf/default.yaml + medicaid mirror; services/canopy-{tanf,medicaid}/tests/chain_v2_append_test.rs (new); docs — configuration-reference.adoc (tanf/medicaid rows), api/canopy-{tanf,medicaid}.adoc (determine 503 arm + OpenAPI responses), CHANGELOG.adoc; own Status row.

MR-2: services/canopy-security/migrations/20260930000000_chain_append_staging.sql (staging + grants + dequeue fn + dedup index); crates/canopy-test-lib/src/db.rs (touch); services/canopy-security/src/lib.rs (new)
src/chain_staging/{stage,router,drainer,health}.rs (new) + main.rs + api/mod.rs
config.rs + Cargo.toml + config/canopy-security/default.yaml; xtask/src/cmd/perf.rs (the chain-drain arm); services/canopy-security/tests/{chain_append_staging_test.rs, chain_append_drainer_test.rs} (new); docs — configuration-reference.adoc (security rows + prefetch/admission), api/canopy-security.adoc (ingest 202/receipt/422), data-models/canopy-security.adoc (staging + dedup index + lifecycle + parked-rows surface + stats/autovacuum note), security-operations.adoc (unpark runbook + staging ops), CHANGELOG.adoc; own Status row.

Step 3 (docs MR): close #1207, parent Step 5 → Done, plan → Archive + nav, throughput cells on #1279.

Sequencing & review-risk notes

0b → MR-1 → MR-2 → Step 3, strictly. The dedup index touches a dormant chain-owned table (flagged in Open decisions); everything else avoids chain-owned objects. Pre-empt in MR descriptions: the main.rs comment reversal; the structural-lease bounded-liveness wording; SQLSTATE+message classification (shipped/immutable string set, test-pinned); the B3a delta.

Open decisions for sign-off

  1. The chain-side UNIQUE dedup index on audit_events_v2 (D1c) — an additive index on a dormant, chain-owned table, replacing an unbounded receipts table as the permanent replay identity. Recorded in Amendment 7.

  2. Structural lease with pinned timeouts (no lease table/TTL; release bounded by lock/statement/idle-in-tx timeouts) — recommended.

  3. Seam wiring now, at today’s call position (V1 byte-identical; medicaid V2 holds head locks across one trailing single-row UPDATE — bounded, documented).

  4. Throughput via cargo xtask perf chain-drain (sustained-ingress slope method) — replaces the rejected #[ignore] rig.

  5. Admission cap semantics (nack→broker on cap; the broker stays the overflow home as today) — the alternative, unbounded DB staging, was rejected by review.

Appendix — external-review disposition (finding → landing section)

Lifecycle reversed → §Step 0 (commit-first) · AC reconciliation contradictory → 0a (AC2→1205/#1206, AC3→D10 slope method, backlog→#1205 AC) · scheduler starvation → D4 (bounded + rotating + scheduler test) · API doesn’t enforce tx → D7 (tx-typed) · routing-version fence circular → D7 fetch_topology (compiled-constant compare) · topology identity fencing → D1 (instance stamp + CHECK) + D4/D7 (instance fence) · replay identity after dequeue → D1c (dedup index) + D4.2 (probe) · same-ID/different-content → D2 (DivergentReplay fail-closed) + D4.2 (digest compare) · staging mutability → D1b (column grants + SECURITY DEFINER dequeue) · routing-id equality → D1 CHECK · poison-isolation algorithm → D4.3 (savepoints, in-tx parks) · deterministic cast poison → D4.2 (prevalidation) + D5 (SQLSTATE-22 → Row; per-row Chain() → park) · lease liveness → D4 (SET LOCAL timeouts + wedge test; Amendment 7 wording) · backlog-status owner → 0a (#1205 AC) + D6 (raw staged) · stale-green health → D6 (sample freshness, pass liveness, always-on sampler; inbox-park/DLQ noted to #1205) · throughput doesn’t test AC → D10 (sustained slope method) · [ignore] policy → D10 (xtask perf arm) · unbounded staging → D1 (admission cap) + D8 · validation under head lock → D4.2 (prevalidate pre-lock) · canon float defect → #1285 · count checks → D4.4/D4.5 · park recovery → D1b (runbook) + test · cutover fencing → D9 + 0a (#1279 items) · FTI >500 → D7 (chunking) · medicaid byte-identical → D7 (seam at today’s position) · test implementability → D7 (seam unit tests in src; suites via canopy-common) + D4 (security lib) + Verification (wedge test; dispatch-level invariant test replacing the wrong corrupt-head test) · DDL invalid states → D1 CHECKs · router width/order → D3 + D8 (CHAIN_ROUTE*) + D1 index (staged_at, event_id) · stats index/budget → D1 (parked_idx) + D6 (pinned aggregate) · ingest receipt → D2 ({event_id} body) · error/API surface → D5 (sqlstate field) + D7 (fixed 503 string; 422 scope; ChainSource decode; Copy on the mode enum) · genesis_hash signature → D7 (SeededChain method) · status/bookkeeping sequencing → §Status (per-MR rows; Step 3 archives) · doc/tracker scope → §Files + 0a/0b · verification omissions → §Verification (all named) · review-state honesty → the NOTE header.

Edit this page · default