From 9e81ffd7fc0f7ba95b08c9ee4e07d324e5a03aee Mon Sep 17 00:00:00 2001 From: "jason.woltje" Date: Sun, 26 Jul 2026 15:48:40 +0000 Subject: [PATCH] fix(wake): #932 stop reconciler re-enumerating already-CONSUMED detector state (#935) Co-authored-by: jason.woltje Co-committed-by: jason.woltje --- .../framework/tools/wake/_wake-common.sh | 6 ++ .../mosaic/framework/tools/wake/manifest.txt | 29 ++++++- .../mosaic/framework/tools/wake/reconcile.sh | 47 +++++++++- packages/mosaic/framework/tools/wake/store.sh | 51 +++++++++++ .../tools/wake/test-wake-reconcile.sh | 86 +++++++++++++++++++ .../tools/wake/test-wake-store-ack.sh | 30 +++++++ 6 files changed, 244 insertions(+), 5 deletions(-) diff --git a/packages/mosaic/framework/tools/wake/_wake-common.sh b/packages/mosaic/framework/tools/wake/_wake-common.sh index 80e2dd04..d753c18b 100755 --- a/packages/mosaic/framework/tools/wake/_wake-common.sh +++ b/packages/mosaic/framework/tools/wake/_wake-common.sh @@ -36,6 +36,12 @@ wake_state_dir() { # {observed_seq, locators, class, emit_ts, hmac}. # ack-ledger.jsonl — local-write-only ack ledger (RECEIVED / CONSUMED). # ack-sync.state — background-sync bookkeeping (last shipped / outage flag). +# consumed-hashes.jsonl — #932 store-owned last-consumed record: one object per +# (kind,id), {kind,id,observed_hash,observed_seq}, written at +# consume-truncation. The reconciler's THIRD accounting source +# (a consumed state matches neither the truncated inbox nor the +# reconciler's own seen-ledger). ADDITIVE / lazily created; +# older code ignores it (on-disk read-compat preserved). # _wake_tmp_glob DIR — the glob used for atomic-write temp files, so readers can # ignore in-flight/crashed writes. A crash leaves one of these; it is NEVER the diff --git a/packages/mosaic/framework/tools/wake/manifest.txt b/packages/mosaic/framework/tools/wake/manifest.txt index b8793213..af939ddd 100644 --- a/packages/mosaic/framework/tools/wake/manifest.txt +++ b/packages/mosaic/framework/tools/wake/manifest.txt @@ -195,8 +195,35 @@ # tolerates on its observed.set-failure path. ON-DISK FORMAT UNCHANGED # (read-compatible; a store written by older code reads identically). Only # store.sh (+ test-wake-store-ack.sh T11) changed. +# 0.6.10 #932 reconciler RE-ENUMERATION of already-CONSUMED detector-observed +# state (wake-pilot finding #7 — safe-but-noisy G2a alarm-hygiene). +# After consume-truncation a consumed state matched NO accounting +# record (inbox truncated; the reconciler's seen-ledger only covers +# its OWN enumerations; the detector hash-file is correctly +# DISTRUSTED) -> the reconciler treated it as UNACCOUNTED and +# re-enumerated it: one DUPLICATE orientation wake + one SPURIOUS +# rc=1 CRITICAL per detector-active window per cycle (functionally +# safe — no lost obligation — but cry-wolf erosion of real alarms at +# fleet scale). FIX: (1) store.sh records the last-consumed +# observed_hash per (kind,id) at consume-truncation into a NEW +# store-owned durable record consumed-hashes.jsonl (atomic write; +# ADDITIVE — existing on-disk format unchanged/read-compatible; #908 +# allocator untouched); (2) reconcile.sh adds a THIRD accounting +# source alongside the inbox and its seen-ledger: a detector-observed +# state whose observed_hash MATCHES the store's recorded last-consumed +# hash is ACCOUNTED (not re-enumerated — no dup wake, no spurious +# CRITICAL). TRUST BOUNDARY: the 3rd check consults ONLY the +# store-written record (its existence implies the state was durably +# enqueued+consumed, so it structurally cannot exhibit the §5 +# hash-advance-without-enqueue swallow signature); trusting DETECTOR +# hash-files STAYS REJECTED. G3 is NOT weakened: only states the store +# RECORDED as consumed are suppressed — a genuinely-unaccounted state +# (enqueued-but-unconsumed, still in the inbox, OR a real gap) still +# re-enumerates + alarms. Changed: store.sh, reconcile.sh, +# _wake-common.sh (doc), test-wake-reconcile.sh (R10/R11), +# test-wake-store-ack.sh (T12). component=wake -version=0.6.9 +version=0.6.10 # Watch-list schema this component consumes, and the INCLUSIVE range of # schema_version values it supports. A wake-watch-list.json whose schema_version diff --git a/packages/mosaic/framework/tools/wake/reconcile.sh b/packages/mosaic/framework/tools/wake/reconcile.sh index e13a9a95..691f323f 100755 --- a/packages/mosaic/framework/tools/wake/reconcile.sh +++ b/packages/mosaic/framework/tools/wake/reconcile.sh @@ -47,10 +47,14 @@ # RETIRED; --allow-enumerate / WAKE_RECONCILE_ALLOW_ENUMERATE remain accepted # as deprecated no-ops for caller compatibility. The G3 accounting invariants # (0-UNACCOUNTED, vacuous-pass prevention) are unaffected and still enforced. -# * "accounted" is judged against observed_seq/inbox (§4/G3 wording) PLUS the -# reconciler's OWN durable reconciled-state ledger — NOT the detector's -# hash-file. Judging by the detector baseline would let a silent first-seen -# baseline swallow a pre-existing obligation (the exact hole G3 closes). +# * "accounted" is judged against THREE sources: observed_seq/inbox (§4/G3 +# wording), the reconciler's OWN durable reconciled-state ledger, and (#932) +# the STORE's last-consumed record (consumed-hashes.jsonl, store-written at +# consume-truncation) — NOT the detector's hash-file. Judging by the detector +# baseline would let a silent first-seen baseline swallow a pre-existing +# obligation (the exact hole G3 closes). The 3rd source suppresses ONLY states +# the store RECORDED as consumed (structurally cannot be hash-advance-without- +# enqueue), so a truly-unaccounted state still re-enumerates (G3 teeth intact). # * enumeration uses the non-coalescible fail-safe class `actionable` (§2.3) # so two distinct pre-existing obligations can never coalesce into one. # @@ -367,6 +371,35 @@ _inbox_has() { [ "${n:-0}" -gt 0 ] } +# _store_consumed_has KIND ID HASH — #932 THIRD accounting source. True iff the +# STORE's last-consumed record (consumed-hashes.jsonl, written by store.sh at +# consume-truncation) records HASH as the last-consumed observed_hash for +# (kind,id). After consume truncates the pending prefix, a consumed state matches +# NEITHER _inbox_has (inbox truncated) NOR the reconciler's own seen-ledger (it +# only covers the reconciler's OWN enumerations) — so without this check the +# reconciler re-enumerates the just-consumed state: one DUPLICATE orientation +# wake + one SPURIOUS rc=1 CRITICAL per detector-active window per cycle (#932). +# +# TRUST BOUNDARY (load-bearing — do NOT violate): this consults ONLY the +# store-written record under STATE_DIR — NEVER a detector-owned hash-file. The +# store record is trustworthy BY CONSTRUCTION: it is written only at CONSUME of a +# durably-enqueued entry, so its existence implies the state WAS durably enqueued, +# and it structurally cannot exhibit the §5 swallow-hole signature (hash-advance- +# WITHOUT-enqueue). Trusting DETECTOR hash-files STAYS REJECTED (it reopens §5). +# +# G3 is NOT weakened: this suppresses ONLY states the store has RECORDED as +# consumed. A genuinely-unaccounted state (durably enqueued but not yet consumed +# — still in the inbox — OR a real gap) has no matching consumed record, so it is +# still UNACCOUNTED and still re-enumerates + alarms. +_store_consumed_has() { + local kind="$1" id="$2" h="$3" n + [ -f "$STATE_DIR/consumed-hashes.jsonl" ] || return 1 + n="$(jq -s --arg k "$kind" --arg id "$id" --arg h "$h" \ + '[ .[] | select(.kind==$k and .id==$id and .observed_hash==$h) ] | length' \ + "$STATE_DIR/consumed-hashes.jsonl" 2>/dev/null || echo 0)" + [ "${n:-0}" -gt 0 ] +} + cmd_reconcile() { local enumerate=1 allow_enumerate=0 while [ $# -gt 0 ]; do @@ -444,6 +477,12 @@ cmd_reconcile() { # Or already reconciled to this exact state before (durable ledger)? elif [ -f "$ledger" ] && [ "$(tr -d '[:space:]' <"$ledger")" = "$curhash" ]; then accounted=1 + # Or the STORE recorded this exact state as ALREADY CONSUMED (#932 — the THIRD + # accounting source; a consume-truncated state matches neither the inbox nor + # the seen-ledger, so without this it would spuriously re-enumerate). Consults + # ONLY the store-written record, never a detector hash-file (§5 stays closed). + elif _store_consumed_has "$kind" "$id" "$curhash"; then + accounted=1 fi if [ "$accounted" -eq 1 ]; then diff --git a/packages/mosaic/framework/tools/wake/store.sh b/packages/mosaic/framework/tools/wake/store.sh index 49572419..ba20df06 100755 --- a/packages/mosaic/framework/tools/wake/store.sh +++ b/packages/mosaic/framework/tools/wake/store.sh @@ -326,6 +326,51 @@ cmd_drain() { grep -v '^[[:space:]]*$' "$STATE_DIR/pending.jsonl" 2>/dev/null || true } +# _record_last_consumed UPTO — #932: at consume-truncation, record the +# last-consumed observed_hash per (kind,id) into a STORE-OWNED durable record +# (consumed-hashes.jsonl). This is the record the reconciler consults as its +# THIRD accounting source: after consume truncates the pending prefix, a consumed +# state matches NEITHER the inbox (truncated) NOR the reconciler's seen-ledger +# (which only covers the reconciler's OWN enumerations), so the reconciler used +# to treat it as UNACCOUNTED and re-enumerate it — one duplicate orientation wake +# + one spurious rc=1 CRITICAL per detector-active window per cycle (safe but +# noisy; G2a alarm-hygiene at fleet scale). +# +# Called BEFORE the pending prefix is dropped (it reads the entries being +# consumed). For each entry with observed_seq <= UPTO that carries full source +# locators (kind,id,observed_hash), keep the HIGHEST-seq hash per (kind,id) as +# that source's last-consumed state; merge with any prior record (monotonic — +# consume only advances, so a newer consume's seq always wins its key). +# +# TRUST BOUNDARY (load-bearing): this record is written ONLY here, at CONSUME of a +# durably-enqueued entry, so its existence implies the state WAS durably +# enqueued+consumed. It therefore structurally cannot exhibit the §5 swallow-hole +# signature (hash-advance-WITHOUT-enqueue). It is a STORE record — the detector's +# own hash-file stays DISTRUSTED and is never consulted for accounting. +# +# ADDITIVE: a new file consumed-hashes.jsonl; the existing on-disk format is +# unchanged and read-compatible (older code ignores this file). Atomic write. +# Best-effort: a failure here degrades to the pre-#932 safe-but-noisy behaviour +# (the reconciler re-enumerates the consumed state once — no lost obligation), +# never a failed consume. +_record_last_consumed() { + local upto="$1" existing new_records merged + [ -f "$STATE_DIR/pending.jsonl" ] || return 0 + new_records="$(jq -c --argjson upto "$upto" ' + select((.observed_seq // -1) <= $upto) + | select((.locators.kind // "") != "" and (.locators.id // "") != "" and (.locators.observed_hash // "") != "") + | {kind:.locators.kind, id:.locators.id, observed_hash:.locators.observed_hash, observed_seq:.observed_seq} + ' "$STATE_DIR/pending.jsonl" 2>/dev/null || true)" + [ -n "$new_records" ] || return 0 + existing="$(cat "$STATE_DIR/consumed-hashes.jsonl" 2>/dev/null || true)" + merged="$(printf '%s\n%s\n' "$existing" "$new_records" | grep -v '^[[:space:]]*$' | + jq -s -c 'group_by([.kind, .id]) | .[] | max_by(.observed_seq)' 2>/dev/null || true)" + [ -n "$merged" ] || return 0 + if ! printf '%s\n' "$merged" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/consumed-hashes.jsonl"; then + echo "store.sh consume: WARN — could not record last-consumed hashes (#932); the reconciler may re-enumerate the consumed state once (safe-but-noisy), no lost obligation." >&2 + fi +} + cmd_consume() { local upto='' while [ $# -gt 0 ]; do @@ -381,6 +426,12 @@ cmd_consume() { k=$((k + 1)) done + # #932: record the last-consumed observed_hash per (kind,id) BEFORE the pending + # prefix is dropped (this reads the entries about to be truncated), so the + # reconciler can recognise an already-consumed state as ACCOUNTED instead of + # re-enumerating it. Additive, store-owned, best-effort (never fails consume). + _record_last_consumed "$upto" + # Advance the consumer cursor and drop the now-consumed prefix from the # durable store (retain-until-CONSUMED is satisfied). jq -c "select(.observed_seq > $upto)" "$STATE_DIR/pending.jsonl" 2>/dev/null | diff --git a/packages/mosaic/framework/tools/wake/test-wake-reconcile.sh b/packages/mosaic/framework/tools/wake/test-wake-reconcile.sh index 2735b01d..e197c896 100755 --- a/packages/mosaic/framework/tools/wake/test-wake-reconcile.sh +++ b/packages/mosaic/framework/tools/wake/test-wake-reconcile.sh @@ -19,6 +19,10 @@ # observed_seq allocator, so co-feeding one store yields DISTINCT, # contiguous seqs (no aliasing). Formerly the fail-closed dual- # allocator refusal; #908 dissolves the hazard at the root. (§4/G3, #908) +# R10 a CONSUMED detector state is ACCOUNTED via the store's last-consumed +# record — NOT re-enumerated (no dup wake, no spurious rc=1 CRITICAL). (#932) +# R11 pilot repro: consumed state SUPPRESSED while a distinct unconsumed/gap +# state STILL re-enumerates + flags (G3 teeth intact; no over-suppression). (#932) # # Uses FAKE/STUB sources only (no live network). Isolated per test. # shellcheck disable=SC2030,SC2031 @@ -295,6 +299,88 @@ EOF "$STORE" consume --upto 3 >/dev/null 2>&1 || fail_msg "R9: CONSUMED 3 must succeed over the gapless co-fed prefix" ) && ok +echo "== R10: #932 — a CONSUMED detector state is ACCOUNTED via the store's last-consumed record (no dup wake, no spurious rc=1 CRITICAL) ==" +( + WAKE_STATE_HOME="$(fresh_state r10)" + export WAKE_STATE_HOME + unset WAKE_AGENT WAKE_RECONCILE_ALLOW_ENUMERATE + stub="$TMP_ROOT/r10stub" + mkdir -p "$stub" + make_stub "$stub" + export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh" + printf 'V0\n' >"$stub/repo_r1" + wl="$TMP_ROOT/r10.json" + cat >"$wl" <<'EOF' +{ "schema_version": 1, + "repos": [ { "id": "r1" } ], + "watches": [ { "lane": "L", "sources": [ { "kind": "repo", "id": "r1" } ] } ] } +EOF + export WAKE_WATCH_LIST="$wl" + # Detector baselines (silent) then a real delta -> one pending inbox entry (seq 1). + "$DET" poll-once >/dev/null 2>&1 || fail_msg "R10: detector baseline failed" + printf 'V1\n' >"$stub/repo_r1" + "$DET" poll-once >/dev/null 2>&1 || fail_msg "R10: detector delta failed" + [ "$(depth)" = "1" ] || fail_msg "R10: detector delta should leave one pending entry, got $(depth)" + # CONSUME it: consumed_seq advances, the pending prefix is truncated. The store + # now records r1's last-consumed observed_hash. The source's CURRENT state is + # UNCHANGED (still V1) — exactly the pilot's just-consumed byte-match window. + "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "R10: CONSUMED 1 must succeed" + [ "$(depth)" = "0" ] || fail_msg "R10: consume must truncate the pending prefix (depth 0), got $(depth)" + # RE-RECONCILE. On BASE this RE-ENUMERATES the just-consumed state (inbox + # truncated + seen-ledger never covered a detector enqueue) -> UNACCOUNTED=1, + # rc=1 (spurious CRITICAL), depth 1 (duplicate wake). After #932 the store's + # last-consumed record ACCOUNTS it -> clean. + out="$("$RECON" reconcile 2>&1)" + rc=$? + [ "$rc" -eq 0 ] || fail_msg "R10: a consumed state must be ACCOUNTED (exit 0, no spurious rc=1 CRITICAL) [$out]" + echo "$out" | grep -qi 'UNACCOUNTED=0' || fail_msg "R10: a consumed state must be 0 unaccounted (no re-enumeration) [$out]" + [ "$(depth)" = "0" ] || fail_msg "R10: reconciler must NOT re-enumerate a consumed state (depth still 0 = no duplicate wake), got $(depth)" +) && ok + +echo "== R11: #932 — pilot repro: consumed state SUPPRESSED while a distinct unconsumed/gap state STILL re-enumerates (G3 teeth intact; no blanket suppression) ==" +( + WAKE_STATE_HOME="$(fresh_state r11)" + export WAKE_STATE_HOME + unset WAKE_AGENT WAKE_RECONCILE_ALLOW_ENUMERATE + stub="$TMP_ROOT/r11stub" + mkdir -p "$stub" + make_stub "$stub" + export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh" + printf 'V0\n' >"$stub/repo_r1" + # r2 is a genuine unaccounted GAP: present from the start so the detector's + # FIRST-SEEN baseline is SILENT (never enqueues it), leaving it unaccounted and + # unconsumed — it has NO last-consumed record, so it MUST still re-enumerate. + printf 'GAP-STATE\n' >"$stub/repo_r2" + wl="$TMP_ROOT/r11.json" + cat >"$wl" <<'EOF' +{ "schema_version": 1, + "repos": [ { "id": "r1" }, { "id": "r2" } ], + "watches": [ { "lane": "L", "sources": [ { "kind": "repo", "id": "r1" }, { "kind": "repo", "id": "r2" } ] } ] } +EOF + export WAKE_WATCH_LIST="$wl" + # Baseline BOTH sources silently (no enqueue on first-seen). r2 stays at this + # state forever -> never enqueued = genuine gap. + "$DET" poll-once >/dev/null 2>&1 || fail_msg "R11: detector baseline failed" + # r1: a real delta -> enqueued (seq 1) -> CONSUMED (store records its hash). + printf 'V1\n' >"$stub/repo_r1" + "$DET" poll-once >/dev/null 2>&1 || fail_msg "R11: detector delta (r1) failed" + "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "R11: CONSUMED 1 must succeed" + [ "$(depth)" = "0" ] || fail_msg "R11: store should be empty before the reconcile, got depth $(depth)" + out="$("$RECON" reconcile 2>&1)" + rc=$? + # r2 is genuinely unaccounted -> the pass STILL FLAGS (G3 teeth intact). + [ "$rc" -ne 0 ] || fail_msg "R11: a genuine gap (r2) must still FLAG (non-zero) [$out]" + # After #932: ONLY r2 is unaccounted (r1 suppressed by the store record). On + # BASE both r1 and r2 re-enumerate (UNACCOUNTED=2) -> this assertion is red-first. + echo "$out" | grep -qi 'UNACCOUNTED=1' || fail_msg "R11: exactly ONE source (the gap r2) must be unaccounted; the consumed r1 must be suppressed [$out]" + # Prove precisely WHICH state re-enumerated: the gap r2 IS enumerated, the + # consumed r1 is NOT. (base re-enumerates r1 too -> the r1-absent assertion is + # red-first; the r2-present assertion holds both before and after = no over-suppression.) + en(){ "$STORE" drain | jq -s --arg id "$1" '[ .[] | select(.locators.id==$id) ] | length'; } + [ "$(en r2)" -ge 1 ] || fail_msg "R11: the genuine gap r2 MUST be re-enumerated into the store (no blanket suppression) [$out]" + [ "$(en r1)" -eq 0 ] || fail_msg "R11: the CONSUMED r1 must NOT be re-enumerated (its byte-matched hash is accounted by the store record) [$out]" +) && ok + echo if [ -s "$FAILFILE" ]; then echo "wake reconcile harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2 diff --git a/packages/mosaic/framework/tools/wake/test-wake-store-ack.sh b/packages/mosaic/framework/tools/wake/test-wake-store-ack.sh index 4b3f5ce9..f39600ec 100755 --- a/packages/mosaic/framework/tools/wake/test-wake-store-ack.sh +++ b/packages/mosaic/framework/tools/wake/test-wake-store-ack.sh @@ -17,6 +17,8 @@ # T9 burn-before-enqueue: a failed durable write must NOT advance the # observed_seq cursor (no burned seq / no interior gap) (#908 arrow 1) # T10 concurrency: two concurrent enqueues get DISTINCT seqs (lock) (#908) +# T12 consume records the last-consumed observed_hash per (kind,id) into the +# store-owned record (additive; monotonic last-seq wins; lazily created) (#932) # # Isolated: every test runs against a fresh WAKE_STATE_HOME temp dir. set -uo pipefail @@ -575,6 +577,34 @@ echo "== T11: final observed_seq cursor write FAILURE — fail loud + observed.s fi ) && ok +echo "== T12: #932 — consume records the last-consumed observed_hash per (kind,id) into the store-owned record (additive; monotonic; keyed by kind,id) ==" +( + WAKE_STATE_HOME="$(fresh_state t12)" + export WAKE_STATE_HOME + unset WAKE_AGENT + rec="$WAKE_STATE_HOME/default/consumed-hashes.jsonl" + # Two source states for the SAME (kind,id): the store must record only the + # LAST-consumed (highest-seq) hash for that key; plus a distinct (kind,id). + s1="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r1","observed_hash":"HASH-A"}')" + s2="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r1","observed_hash":"HASH-B"}')" + s3="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r2","observed_hash":"HASH-C"}')" + [ "$s1 $s2 $s3" = "1 2 3" ] || fail_msg "T12: contiguous allocation expected 1 2 3, got '$s1 $s2 $s3'" + # BEFORE consume there is no record (additive / lazily created). + [ ! -f "$rec" ] || fail_msg "T12: the last-consumed record must not exist before any consume" + "$STORE" consume --upto 3 >/dev/null 2>&1 || fail_msg "T12: CONSUMED 3 must succeed over the gapless prefix" + # AFTER consume the store-owned record exists and holds ONE entry per (kind,id). + [ -f "$rec" ] || fail_msg "T12: consume must write the store-owned last-consumed record ($rec)" + nkeys="$(jq -s '[ .[] | {kind,id} ] | unique | length' "$rec" 2>/dev/null || echo 0)" + [ "$nkeys" = "2" ] || fail_msg "T12: the record must hold exactly one entry per (kind,id) — 2 keys, got $nkeys" + # repo/r1 must record the LAST-consumed hash (HASH-B at seq 2), never HASH-A. + r1h="$(jq -sr '[ .[] | select(.kind=="repo" and .id=="r1") ] | .[0].observed_hash' "$rec" 2>/dev/null)" + [ "$r1h" = "HASH-B" ] || fail_msg "T12: repo/r1 must record the LAST-consumed hash HASH-B (seq 2), got '$r1h'" + r2h="$(jq -sr '[ .[] | select(.kind=="repo" and .id=="r2") ] | .[0].observed_hash' "$rec" 2>/dev/null)" + [ "$r2h" = "HASH-C" ] || fail_msg "T12: repo/r2 must record HASH-C, got '$r2h'" + # ADDITIVE: existing on-disk state files are unchanged/consistent post-consume. + "$STORE" cursors | grep -q 'consumed_seq=3' || fail_msg "T12: consumed_seq must be 3 after CONSUMED 3" +) && ok + echo if [ -s "$FAILFILE" ]; then echo "wake store/ack harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2