Jason WoltjeandClaude Fable 5
e11bc66223
ci/woodpecker/pr/ci Pipeline was successful
The embedded ack suggestion covered dead-lettered entries: quarantine is a render-time filter (0.6.14), so `ack.sh consumed --upto <observed_seq>` stepped the cursor past quarantined seqs and _record_last_consumed wrote consumed-hash witness rows for deliveries that never happened (live: mos-dt seq 68 buried under five successive digests; Finding A: a false 9d0f639f…@63 witness row). Fix, per the ruled disposition on #946: - digest.sh: rendered digest gains a QUARANTINED section (seq + class + HELD only — ids and locator values stay withheld, preserving the exclusion property); embedded ack clamped to min(observed_seq, min quarantined seq - 1) with a loud "# ACK CLAMPED (#946)" note; render --from-store syncs the store-owned quarantined.set via `store.sh quarantine-sync` (full replace — a gate fix self-heals stale quarantine); --from-file/--stdin never touch the set. - store.sh: consume REFUSES to cross an unconsumed quarantined seq; `--force-past-quarantine` is the ONLY way past, loud per-seq on stderr, and even the forced path never writes a consumed-hash witness for a quarantined seq; crossed seqs are pruned from the set after the cursor moves. New `quarantine-sync` (stdin seqs, full replace, invalid input is a loud no-op) and `quarantine-audit [--repair]` (sweeps consumed-hashes for rows provably contradicted by the dead-letter ledger; report exits 1; --repair removes only provably-false rows; the ledger is history and never modified; rows whose dead-letter evidence was pruned are unprovable and untouched). - ack.sh: plumbs --force-past-quarantine through to store consume; forced- path loudness is re-emitted on stderr while `CONSUMED <n>` stays clean. Tests: test-wake-store-ack.sh T13-T16 (34 assertions RED at base), test-wake-digest-quarantine.sh Q12-Q16 (10 RED at base; Q13/Q16 green-by-design controls). All nine wake suites green. version=0.6.15. Closes #946 Written-by: pepper (sb-it-1-dt) Co-Authored-By: Claude Fable 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01NsKce8iZuSuRnu3gVMCBKB
679 lines
30 KiB
Bash
Executable File
679 lines
30 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# store.sh — A2 of the wake canon (EPIC #892, W2): the THREE-CURSOR durable
|
|
# queue (store + drain lib).
|
|
#
|
|
# CONTRACT ANCHORS (docs/scratchpads/heartbeat-planning/CONVERGED-DESIGN.md):
|
|
# §1.2 three-cursor data-flow: observe -> durable store -> coalesce -> drain.
|
|
# §2.3 classes & durability: ALL classes durable; coalescing is optional.
|
|
# §2.4 cursor semantics: observed_seq is authoritative; SHAs are descriptors.
|
|
#
|
|
# Three cursors:
|
|
# observed_seq (store-owned SINGLE ALLOCATOR, #908) — monotonic int assigned by
|
|
# THIS lib at enqueue: next = observed_seq + 1, under an exclusive
|
|
# lock, committed IFF the durable write succeeds. Authoritative.
|
|
# There is exactly ONE allocator (this store); the detector and
|
|
# reconciler no longer allocate independently. Source SHAs are
|
|
# descriptors, NOT the cursor (§2.4).
|
|
# pending-inbox (durable) — append store; retains until CONSUMED; survives
|
|
# pane death / park / restart.
|
|
# consumed_seq (consumer-owned) — advances ONLY on a CONSUMED ack, over a
|
|
# CONTIGUOUS gapless prefix <=N. NEVER on sender exit-0 / paste.
|
|
#
|
|
# Coalesce (§1.2/§2.3): class=digest REPLACES the pending cumulative-state entry
|
|
# (newest subsumes prior). Non-coalescible classes (actionable, human, ...)
|
|
# APPEND. ALL classes are stored durably — durability is NEVER bypassed.
|
|
# Absent class => treated as `actionable` (fail-safe).
|
|
#
|
|
# HMAC: entries carry an `hmac` field left as an UNSIGNED PLACEHOLDER ("").
|
|
# Signing is W3/A5, explicitly NOT W2.
|
|
#
|
|
# Every state mutation is atomic write-tmp+rename (see _wake-common.sh).
|
|
#
|
|
# Operator-agnostic: state via XDG/env only; no operator paths/names/secrets.
|
|
set -uo pipefail
|
|
|
|
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
|
# shellcheck source=./_wake-common.sh disable=SC1091
|
|
. "$SCRIPT_DIR/_wake-common.sh"
|
|
|
|
STATE_DIR="$(wake_state_dir)"
|
|
|
|
_need_jq() {
|
|
command -v jq >/dev/null 2>&1 || {
|
|
echo "store.sh: jq is required" >&2
|
|
exit 3
|
|
}
|
|
}
|
|
|
|
usage() {
|
|
cat >&2 <<'EOF'
|
|
Usage: store.sh <command> [options]
|
|
|
|
Commands:
|
|
init Create/repair the XDG state layout.
|
|
enqueue [--class C] [opts] Record an observation into the durable
|
|
pending-inbox (coalesce or append) and
|
|
ALLOCATE its observed_seq. store.sh is the
|
|
SOLE allocator of observed_seq (#908):
|
|
under an exclusive enqueue lock it reads
|
|
the observed_seq cursor, sets
|
|
next = observed_seq + 1, writes the
|
|
pending record + observed.set + the cursor
|
|
as ONE transaction, and PRINTS next to
|
|
stdout. The cursor advances IFF the durable
|
|
write succeeds (no burn-before-enqueue).
|
|
--seq N LEGACY explicit observed_seq (optional). Normal callers
|
|
(detector, reconciler) OMIT this and let the store
|
|
allocate. Retained only for tests/tools that must place a
|
|
SPECIFIC seq (e.g. construct a gap). An explicit seq that
|
|
is <= consumed_seq is REFUSED loudly (anti-swallow): the
|
|
idempotent-ignore path can never silently drop a wake.
|
|
--class C digest|actionable|human|terminal-log|reaction.
|
|
Absent/empty => actionable (fail-safe).
|
|
--locators JSON JSON value for the entry's locators (default {}).
|
|
--emit-ts TS epoch seconds (default: now).
|
|
drain [--require-idle-cmd CMD] Emit the deliverable set (pending-inbox,
|
|
coalesced) as JSONL. WHAT to deliver;
|
|
the actual paste is out of scope.
|
|
If --require-idle-cmd is given and it
|
|
exits non-zero, emit nothing (not idle).
|
|
consume --upto N [--force-past-quarantine]
|
|
Advance consumed_seq over the contiguous
|
|
gapless prefix <=N; drop consumed
|
|
entries. Rejects a gap (cannot ack N
|
|
while N-1 is unconsumed). Cumulative &
|
|
idempotent. REFUSES to advance past a
|
|
QUARANTINED seq (#946): a quarantined
|
|
entry was dead-lettered at render and
|
|
never delivered in any digest, so an
|
|
ordinary ack may not record it consumed.
|
|
--force-past-quarantine is the ONLY way
|
|
past — loud per stepped-over seq, and
|
|
even then NO consumed-hash witness is
|
|
recorded for the quarantined entry.
|
|
quarantine-sync REPLACE the store-owned quarantined.set
|
|
with the observed_seqs read from stdin
|
|
(one per line; empty input CLEARS).
|
|
Called by digest.sh after each
|
|
authoritative store render — the set is
|
|
re-DERIVED per render, never accumulated,
|
|
so a fixed locator gate self-heals the
|
|
consume clamp (#944 recovery).
|
|
quarantine-audit [--repair] Report consumed-hashes rows that are
|
|
PROVABLY FALSE: the row matches a
|
|
dead-letter ledger entry on
|
|
(kind,id,observed_seq,observed_hash) at
|
|
or below consumed_seq — i.e. the recorded
|
|
consumption was of a quarantined,
|
|
never-delivered entry (#946 Finding A).
|
|
Report-only by default (exit 1 when any
|
|
found); --repair removes exactly those
|
|
rows (atomic, loud). The dead-letter
|
|
ledger itself is NEVER modified (it is
|
|
history). Rows whose dead-letter evidence
|
|
was pruned are NOT provable and are
|
|
never touched.
|
|
cursors Print observed_seq / consumed_seq / depth.
|
|
|
|
Environment:
|
|
WAKE_STATE_HOME override base (default ${XDG_STATE_HOME:-$HOME/.local/state}/mosaic/wake)
|
|
WAKE_AGENT per-agent queue namespace (default: default)
|
|
EOF
|
|
}
|
|
|
|
cmd_init() {
|
|
_wake_init_dir "$STATE_DIR" || {
|
|
echo "store.sh: failed to init $STATE_DIR" >&2
|
|
exit 1
|
|
}
|
|
# Maintenance/daemon-start stale-tmp reap (#927). Cleanup is OFF the per-enqueue
|
|
# hot path; `init` is a natural once-per-start maintenance point. It is
|
|
# age-scoped, so even if an operator runs `init` while an enqueue is in flight
|
|
# it can only reap demonstrably-orphaned (crash-left) tmps, never a live write.
|
|
_wake_clean_stale_tmp "$STATE_DIR"
|
|
echo "$STATE_DIR"
|
|
}
|
|
|
|
cmd_enqueue() {
|
|
_need_jq
|
|
local seq='' class='' locators='{}' emit_ts=''
|
|
while [ $# -gt 0 ]; do
|
|
case "$1" in
|
|
--seq)
|
|
seq="${2:-}"
|
|
shift 2
|
|
;;
|
|
--class)
|
|
class="${2:-}"
|
|
shift 2
|
|
;;
|
|
--locators)
|
|
locators="${2:-}"
|
|
shift 2
|
|
;;
|
|
--emit-ts)
|
|
emit_ts="${2:-}"
|
|
shift 2
|
|
;;
|
|
*)
|
|
echo "store.sh enqueue: unknown option '$1'" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
done
|
|
|
|
# --seq is OPTIONAL (#908): normal callers omit it and the store allocates.
|
|
# If supplied (legacy/explicit), it must be a non-negative integer.
|
|
local explicit=0
|
|
if [ -n "$seq" ]; then
|
|
explicit=1
|
|
case "$seq" in
|
|
*[!0-9]*)
|
|
echo "store.sh enqueue: --seq must be a non-negative integer" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
fi
|
|
|
|
# Absent class => actionable (fail-safe, §2.3).
|
|
[ -n "$class" ] || class="actionable"
|
|
case "$class" in
|
|
digest | actionable | human | terminal-log | reaction) : ;;
|
|
*)
|
|
echo "store.sh enqueue: unknown class '$class'" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
|
|
[ -n "$emit_ts" ] || emit_ts="$(date +%s)"
|
|
case "$emit_ts" in
|
|
*[!0-9]*)
|
|
echo "store.sh enqueue: --emit-ts must be epoch seconds" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
|
|
# Validate locators is well-formed JSON.
|
|
if ! printf '%s' "$locators" | jq -e . >/dev/null 2>&1; then
|
|
echo "store.sh enqueue: --locators must be valid JSON" >&2
|
|
exit 2
|
|
fi
|
|
|
|
# Ensure the layout exists. NB (#927): _wake_init_dir NO LONGER reaps stale
|
|
# tmps — that would race a concurrent enqueue's live in-flight write (deleting
|
|
# its tmp mid-rename -> spurious "durable pending write FAILED"). Stale-tmp
|
|
# reaping is a maintenance action (store.sh init / detector tick), never on this
|
|
# hot path.
|
|
_wake_init_dir "$STATE_DIR"
|
|
|
|
# --- serialize the whole allocate+write transaction (#908) ----------------
|
|
# A single store-side allocator means the read of observed_seq, the durable
|
|
# write, and the cursor bump MUST be one critical section: two concurrent
|
|
# enqueues that both read observed_seq=N would otherwise both allocate N+1 and
|
|
# collide. The lock (flock where present) makes concurrent enqueues get
|
|
# DISTINCT seqs.
|
|
_wake_lock_acquire "$STATE_DIR/.enqueue.lock" || {
|
|
echo "store.sh enqueue: could not acquire the enqueue lock ($STATE_DIR/.enqueue.lock)" >&2
|
|
exit 1
|
|
}
|
|
|
|
local consumed observed
|
|
consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
|
|
observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
|
|
|
|
if [ "$explicit" -eq 0 ]; then
|
|
# ALLOCATE: the store is the sole allocator. next = observed_seq + 1.
|
|
seq=$((observed + 1))
|
|
fi
|
|
|
|
# --- anti-swallow FAIL-LOUD (#908, arrow #3) ------------------------------
|
|
# observed_seq >= consumed_seq always (consumed advances only over the gapless
|
|
# observed prefix), so an ALLOCATED next = observed+1 is ALWAYS > consumed. The
|
|
# old "seq <= consumed => idempotent no-op" path is therefore UNREACHABLE for an
|
|
# allocation; if it is ever reached the store is inconsistent (corruption) — we
|
|
# FAIL LOUD, never a silent no-op. For a LEGACY explicit --seq, a seq inside the
|
|
# consumed prefix is likewise refused loudly so it can never silently swallow a
|
|
# fresh obligation (the migration-restart killer must be structurally impossible).
|
|
if [ "$seq" -le "$consumed" ]; then
|
|
_wake_lock_release
|
|
if [ "$explicit" -eq 1 ]; then
|
|
echo "store.sh enqueue: REFUSED explicit --seq $seq <= consumed_seq $consumed — anti-swallow (#908): a seq inside the consumed prefix is never a fresh obligation and must NEVER be silently ignored." >&2
|
|
else
|
|
echo "store.sh enqueue: FAIL LOUD (#908 allocation invariant) — computed observed_seq $seq <= consumed_seq $consumed (observed_seq cursor=$observed). The store is inconsistent; refusing to allocate rather than silently no-op." >&2
|
|
fi
|
|
exit 1
|
|
fi
|
|
|
|
# Build the durable entry. `hmac` is an UNSIGNED PLACEHOLDER (W3/A5 signs it).
|
|
local entry
|
|
entry="$(jq -cn \
|
|
--argjson seq "$seq" \
|
|
--arg class "$class" \
|
|
--argjson locators "$locators" \
|
|
--argjson emit_ts "$emit_ts" \
|
|
'{observed_seq:$seq, locators:$locators, class:$class, emit_ts:$emit_ts, hmac:""}')" || {
|
|
_wake_lock_release
|
|
echo "store.sh enqueue: failed to build entry" >&2
|
|
exit 1
|
|
}
|
|
|
|
# --- durable store write (coalesce vs append) -----------------------------
|
|
# digest: REPLACE the single pending cumulative-state entry (drop any existing
|
|
# digest lines, then append the newest). Newest subsumes prior.
|
|
# others: APPEND (never replaced).
|
|
# In BOTH cases the entry lands in the durable store.
|
|
#
|
|
# ORDERING (#908, arrow #1 — no burn-before-enqueue): the durable pending write
|
|
# comes FIRST and its failure ABORTS before the observed.set / observed_seq
|
|
# cursor writes. The cursor advances IFF the durable pending write succeeds, so
|
|
# a failed enqueue can never burn a seq that never reached the store.
|
|
local new_pending
|
|
if [ "$class" = "digest" ]; then
|
|
new_pending="$(jq -c 'select(.class != "digest")' "$STATE_DIR/pending.jsonl" 2>/dev/null || true)"
|
|
new_pending="$(printf '%s\n%s\n' "$new_pending" "$entry" | grep -v '^[[:space:]]*$' || true)"
|
|
else
|
|
new_pending="$(cat "$STATE_DIR/pending.jsonl" 2>/dev/null; printf '%s\n' "$entry")"
|
|
new_pending="$(printf '%s' "$new_pending" | grep -v '^[[:space:]]*$' || true)"
|
|
fi
|
|
if ! printf '%s\n' "$new_pending" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/pending.jsonl"; then
|
|
_wake_lock_release
|
|
echo "store.sh enqueue: durable pending write FAILED for seq $seq — observed_seq cursor NOT advanced (no burned seq, no interior gap)." >&2
|
|
exit 1
|
|
fi
|
|
|
|
# --- observed.set: record the seq in the live window (gap-detector SoT) ----
|
|
# Snapshot observed.set's prior content FIRST: if the final cursor write below
|
|
# fails AFTER this write advances observed.set, we roll observed.set back to this
|
|
# snapshot so observed.set can never be left cross-file inconsistent with the
|
|
# cursor (#917 — observed.set ahead of a cursor that never committed).
|
|
local prev_observed_set
|
|
prev_observed_set="$(cat "$STATE_DIR/observed.set" 2>/dev/null || true)"
|
|
if ! {
|
|
printf '%s' "$prev_observed_set"
|
|
printf '\n%s\n' "$seq"
|
|
} | grep -v '^[[:space:]]*$' | sort -n | uniq | _atomic_write "$STATE_DIR/observed.set"; then
|
|
_wake_lock_release
|
|
echo "store.sh enqueue: observed.set write FAILED for seq $seq — observed_seq cursor NOT advanced." >&2
|
|
exit 1
|
|
fi
|
|
|
|
# --- observed_seq cursor: max monotonic int (COMMIT of the allocation) -----
|
|
# Written LAST: only now, after the durable pending + observed.set writes
|
|
# succeeded, does the cursor advance. This is the atomic commit point of the
|
|
# allocation.
|
|
#
|
|
# #917 (defense-in-depth): this final write is GATED like the pending/observed.set
|
|
# writes (#908) — its failure is FAIL-LOUD, never silently swallowed (which would
|
|
# return success while the allocation stayed uncommitted). And because observed.set
|
|
# was already advanced just above, a cursor-write failure would leave observed.set
|
|
# ahead of the cursor (cross-file inconsistent); so on failure we ROLL observed.set
|
|
# BACK to its pre-write snapshot before failing loud. Net: observed.set and the
|
|
# cursor either BOTH advance or NEITHER does — never a stranded observed.set entry
|
|
# above the cursor. (The pending write may remain ahead, exactly as #908 already
|
|
# tolerates on the observed.set-failure path: the cursor is the sole COMMIT point,
|
|
# so an uncommitted pending entry is re-derived/reconciled, never consumed — the
|
|
# allocation is simply not committed.) On-disk format is UNCHANGED (read-compatible).
|
|
if [ "$seq" -gt "$observed" ]; then
|
|
if ! printf '%s' "$seq" | _atomic_write "$STATE_DIR/observed_seq"; then
|
|
printf '%s' "$prev_observed_set" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/observed.set" || true
|
|
_wake_lock_release
|
|
echo "store.sh enqueue: observed_seq cursor write FAILED for seq $seq — allocation NOT committed; rolled observed.set back so it stays consistent with the cursor (#917)." >&2
|
|
exit 1
|
|
fi
|
|
fi
|
|
|
|
_wake_lock_release
|
|
|
|
# Print the allocated observed_seq to stdout (sole-allocator contract): the
|
|
# detector/reconciler capture it for logging/locators/emit.
|
|
printf '%s\n' "$seq"
|
|
}
|
|
|
|
cmd_drain() {
|
|
local idle_cmd=''
|
|
while [ $# -gt 0 ]; do
|
|
case "$1" in
|
|
--require-idle-cmd)
|
|
idle_cmd="${2:-}"
|
|
shift 2
|
|
;;
|
|
*)
|
|
echo "store.sh drain: unknown option '$1'" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
done
|
|
|
|
# Drain returns the deliverable set ONLY when the pane is idle-at-prompt. The
|
|
# actual idle detection + paste is out of W2 scope (send-message.sh); the
|
|
# caller supplies an idle predicate. No predicate => emit unconditionally.
|
|
if [ -n "$idle_cmd" ]; then
|
|
if ! sh -c "$idle_cmd" >/dev/null 2>&1; then
|
|
return 0 # not idle-at-prompt: deliver nothing this cycle.
|
|
fi
|
|
fi
|
|
|
|
[ -f "$STATE_DIR/pending.jsonl" ] || return 0
|
|
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 qjson
|
|
[ -f "$STATE_DIR/pending.jsonl" ] || return 0
|
|
# #946: seqs in quarantined.set are EXCLUDED from the record — the trust
|
|
# boundary above says "existence implies durably enqueued+CONSUMED", but a
|
|
# quarantined entry was dead-lettered at render and NEVER delivered, so a row
|
|
# for it would witness a delivery that never happened. This holds even on the
|
|
# FORCED step-over path: the reconciler re-enumerating the state once is
|
|
# safe-but-noisy; a false witness silences it forever.
|
|
qjson="$(jq -nR -c '[inputs | select(length > 0) | tonumber? // empty]' "$STATE_DIR/quarantined.set" 2>/dev/null || true)"
|
|
[ -n "$qjson" ] || qjson='[]'
|
|
new_records="$(jq -c --argjson upto "$upto" --argjson quarantined "$qjson" '
|
|
(.observed_seq // -1) as $seq
|
|
| select($seq <= $upto)
|
|
| select(($quarantined | index($seq)) == null)
|
|
| 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='' force=0
|
|
while [ $# -gt 0 ]; do
|
|
case "$1" in
|
|
--upto)
|
|
upto="${2:-}"
|
|
shift 2
|
|
;;
|
|
--force-past-quarantine)
|
|
force=1
|
|
shift
|
|
;;
|
|
*)
|
|
echo "store.sh consume: unknown option '$1'" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
done
|
|
case "$upto" in
|
|
'' | *[!0-9]*)
|
|
echo "store.sh consume: --upto must be a non-negative integer" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
|
|
_need_jq
|
|
_wake_init_dir "$STATE_DIR"
|
|
|
|
local consumed observed
|
|
consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
|
|
observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
|
|
|
|
# Cumulative & idempotent: CONSUMED N implies all <=N. N already covered is a
|
|
# no-op success (the sender checks cursor-advanced-past-X, not per-wake).
|
|
if [ "$upto" -le "$consumed" ]; then
|
|
echo "$consumed"
|
|
return 0
|
|
fi
|
|
|
|
# Never consume beyond what was observed.
|
|
if [ "$upto" -gt "$observed" ]; then
|
|
echo "store.sh consume: cannot CONSUMED $upto beyond observed_seq $observed" >&2
|
|
exit 1
|
|
fi
|
|
|
|
# CONTIGUOUS gapless prefix: every seq in (consumed, upto] must have been
|
|
# observed. A missing interior seq is a GAP => reject (cannot ack N while N-1
|
|
# is unconsumed). observed.set is coalescing-immune, so a subsumed digest seq
|
|
# still counts as observed and does not read as a gap.
|
|
local k
|
|
k=$((consumed + 1))
|
|
while [ "$k" -le "$upto" ]; do
|
|
if ! grep -qxF "$k" "$STATE_DIR/observed.set" 2>/dev/null; then
|
|
echo "store.sh consume: gap at seq $k — cannot CONSUMED $upto while $k is unobserved/unconsumed" >&2
|
|
exit 1
|
|
fi
|
|
k=$((k + 1))
|
|
done
|
|
|
|
# --- #946 quarantine CLAMP -------------------------------------------------
|
|
# A quarantined seq was DEAD-LETTERED at render (no §2.1 hard locator): it was
|
|
# never delivered in any digest, so advancing consumed_seq past it would record
|
|
# consumption of an entry the consumer has never seen. The ordinary path
|
|
# REFUSES; --force-past-quarantine is the ONLY way past, and it is loud per
|
|
# stepped-over seq. digest.sh re-derives the set at each authoritative store
|
|
# render (quarantine-sync REPLACE), so a fixed locator gate self-heals this
|
|
# clamp without operator action.
|
|
local qfile="$STATE_DIR/quarantined.set" blocked='' q
|
|
if [ -s "$qfile" ]; then
|
|
while IFS= read -r q; do
|
|
case "$q" in '' | *[!0-9]*) continue ;; esac
|
|
if [ "$q" -gt "$consumed" ] && [ "$q" -le "$upto" ]; then
|
|
blocked="$blocked $q"
|
|
fi
|
|
done <"$qfile"
|
|
fi
|
|
if [ -n "$blocked" ]; then
|
|
if [ "$force" -eq 0 ]; then
|
|
echo "store.sh consume: REFUSED (#946 quarantine clamp) — quarantined seq(s):$blocked inside (consumed_seq=$consumed, upto=$upto] were dead-lettered at render and NEVER delivered in any digest. Advancing past them would record consumption of entries the consumer has never seen. Disposition them (see $STATE_DIR/dead-letter.jsonl) or step over EXPLICITLY with --force-past-quarantine." >&2
|
|
exit 1
|
|
fi
|
|
for q in $blocked; do
|
|
echo "store.sh consume: FORCED PAST QUARANTINE (#946) — stepping consumed_seq over quarantined seq $q (dead-lettered, NEVER delivered). No consumed-hash witness is recorded for it; the obligation stays visible ONLY in $STATE_DIR/dead-letter.jsonl." >&2
|
|
done
|
|
fi
|
|
|
|
# #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 |
|
|
grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/pending.jsonl" || true
|
|
awk -v c="$upto" 'NF && $1+0 > c' "$STATE_DIR/observed.set" 2>/dev/null |
|
|
_atomic_write "$STATE_DIR/observed.set" || true
|
|
printf '%s' "$upto" | _atomic_write "$STATE_DIR/consumed_seq"
|
|
# #946: on a forced step-over, PRUNE the stepped-over seqs from quarantined.set
|
|
# (they are inside the consumed prefix now; a stale entry would re-refuse the
|
|
# next consume forever). Mirrors the observed.set prune idiom above.
|
|
if [ -n "$blocked" ]; then
|
|
awk -v c="$upto" 'NF && $1+0 > c' "$qfile" 2>/dev/null |
|
|
_atomic_write "$qfile" || true
|
|
fi
|
|
echo "$upto"
|
|
}
|
|
|
|
# cmd_quarantine_sync — #946: REPLACE the store-owned quarantined.set with the
|
|
# observed_seqs read from stdin (one per line). Called by digest.sh after each
|
|
# AUTHORITATIVE full-set render (src=store): the set is re-DERIVED per render,
|
|
# never accumulated, so a fixed locator gate self-heals the consume clamp (the
|
|
# #944 recovery case — a cumulative-forever set would keep refusing acks on
|
|
# entries that now render). Empty input CLEARS the set. Foreign-data renders
|
|
# (--from-file/--stdin) never call this. Atomic write; invalid input is refused
|
|
# loudly with the set left untouched.
|
|
cmd_quarantine_sync() {
|
|
[ $# -eq 0 ] || {
|
|
echo "store.sh quarantine-sync: takes no options (observed_seqs on stdin, one per line)" >&2
|
|
exit 2
|
|
}
|
|
local seq list=''
|
|
while IFS= read -r seq; do
|
|
[ -n "$seq" ] || continue
|
|
case "$seq" in
|
|
*[!0-9]*)
|
|
echo "store.sh quarantine-sync: invalid observed_seq '$seq' — one non-negative integer per line; set left untouched" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
list="$list$seq"$'\n'
|
|
done
|
|
_wake_init_dir "$STATE_DIR"
|
|
if ! { printf '%s' "$list" | grep -v '^[[:space:]]*$' || true; } | sort -n | uniq | _atomic_write "$STATE_DIR/quarantined.set"; then
|
|
echo "store.sh quarantine-sync: quarantined.set write FAILED — the consume clamp may be stale for this lane" >&2
|
|
exit 1
|
|
fi
|
|
}
|
|
|
|
# cmd_quarantine_audit — #946 Finding A: the pre-#946 consume recorded
|
|
# consumed-hash rows for QUARANTINED (never-delivered) entries — false
|
|
# witnesses that silence the reconciler for keys whose only "consumption" was a
|
|
# dead-lettered entry (live: safe by population only where the key re-emitted
|
|
# and a later seq won the per-key max_by merge; keys that never recurred keep
|
|
# the false row forever).
|
|
#
|
|
# A row is PROVABLY FALSE iff the dead-letter ledger contains an entry matching
|
|
# it on (kind, id, observed_seq, observed_hash) AND row.observed_seq <=
|
|
# consumed_seq: the per-key max_by merge means the surviving row's provenance IS
|
|
# that quarantined entry (a healed row differs in seq/hash and never matches).
|
|
# PROVABILITY BOUND: a row whose dead-letter evidence was pruned/rotated away is
|
|
# NOT provable and is never touched — this audit only ever removes what the
|
|
# ledger can convict. The dead-letter ledger itself is history and is NEVER
|
|
# modified here.
|
|
cmd_quarantine_audit() {
|
|
local repair=0
|
|
while [ $# -gt 0 ]; do
|
|
case "$1" in
|
|
--repair)
|
|
repair=1
|
|
shift
|
|
;;
|
|
*)
|
|
echo "store.sh quarantine-audit: unknown option '$1'" >&2
|
|
exit 2
|
|
;;
|
|
esac
|
|
done
|
|
_need_jq
|
|
_wake_init_dir "$STATE_DIR"
|
|
local rec="$STATE_DIR/consumed-hashes.jsonl" dlf="$STATE_DIR/dead-letter.jsonl"
|
|
if [ ! -s "$rec" ]; then
|
|
echo "store.sh quarantine-audit: OK — no consumed-hashes record to audit"
|
|
return 0
|
|
fi
|
|
local dl
|
|
dl="$(jq -s -c '[.[] | {kind: (.locators.kind // ""), id: ((.locators.id // "") | tostring), observed_seq: (.observed_seq // -1), observed_hash: (.locators.observed_hash // "")}]' "$dlf" 2>/dev/null || true)"
|
|
[ -n "$dl" ] || dl='[]'
|
|
local consumed
|
|
consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
|
|
local false_rows
|
|
false_rows="$(jq -c --argjson dl "$dl" --argjson consumed "$consumed" '
|
|
. as $row
|
|
| select(($row.observed_seq // -1) <= $consumed)
|
|
| select(($dl | map(select(
|
|
.kind == ($row.kind // "")
|
|
and .id == (($row.id // "") | tostring)
|
|
and .observed_seq == ($row.observed_seq // -1)
|
|
and .observed_hash == ($row.observed_hash // "")
|
|
)) | length) > 0)
|
|
' "$rec" 2>/dev/null || true)"
|
|
if [ -z "$false_rows" ]; then
|
|
echo "store.sh quarantine-audit: OK — no provably-false consumed-hash rows (rows without surviving dead-letter evidence are not provable and were not judged)"
|
|
return 0
|
|
fi
|
|
local n row
|
|
n="$(printf '%s\n' "$false_rows" | grep -c . || true)"
|
|
while IFS= read -r row; do
|
|
[ -n "$row" ] || continue
|
|
if [ "$repair" -eq 1 ]; then
|
|
echo "store.sh quarantine-audit: REPAIR — removing FALSE WITNESS row $row (matches a dead-lettered, never-delivered entry at/below consumed_seq=$consumed)" >&2
|
|
else
|
|
echo "FALSE WITNESS — consumed-hashes row $row matches a dead-lettered, never-delivered entry at/below consumed_seq=$consumed (the recorded consumption never happened)"
|
|
fi
|
|
done <<<"$false_rows"
|
|
if [ "$repair" -eq 0 ]; then
|
|
echo "store.sh quarantine-audit: $n provably-false row(s) found — run with --repair to remove exactly these rows"
|
|
return 1
|
|
fi
|
|
local kept
|
|
kept="$(jq -c --argjson dl "$dl" --argjson consumed "$consumed" '
|
|
. as $row
|
|
| select(
|
|
(($row.observed_seq // -1) > $consumed)
|
|
or (($dl | map(select(
|
|
.kind == ($row.kind // "")
|
|
and .id == (($row.id // "") | tostring)
|
|
and .observed_seq == ($row.observed_seq // -1)
|
|
and .observed_hash == ($row.observed_hash // "")
|
|
)) | length) == 0)
|
|
)
|
|
' "$rec" 2>/dev/null || true)"
|
|
if ! { printf '%s\n' "$kept" | grep -v '^[[:space:]]*$' || true; } | _atomic_write "$rec"; then
|
|
echo "store.sh quarantine-audit: consumed-hashes rewrite FAILED — record left untouched" >&2
|
|
exit 1
|
|
fi
|
|
echo "store.sh quarantine-audit: repaired — removed $n provably-false row(s); dead-letter ledger untouched (history)"
|
|
}
|
|
|
|
cmd_cursors() {
|
|
_wake_init_dir "$STATE_DIR"
|
|
local observed consumed depth
|
|
observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
|
|
consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
|
|
depth="$(grep -cv '^[[:space:]]*$' "$STATE_DIR/pending.jsonl" 2>/dev/null || echo 0)"
|
|
printf 'observed_seq=%s\nconsumed_seq=%s\npending_depth=%s\n' "$observed" "$consumed" "$depth"
|
|
}
|
|
|
|
main() {
|
|
[ $# -ge 1 ] || {
|
|
usage
|
|
exit 2
|
|
}
|
|
local cmd="$1"
|
|
shift
|
|
case "$cmd" in
|
|
init) cmd_init "$@" ;;
|
|
enqueue) cmd_enqueue "$@" ;;
|
|
drain) cmd_drain "$@" ;;
|
|
consume) cmd_consume "$@" ;;
|
|
quarantine-sync) cmd_quarantine_sync "$@" ;;
|
|
quarantine-audit) cmd_quarantine_audit "$@" ;;
|
|
cursors) cmd_cursors "$@" ;;
|
|
-h | --help | help) usage ;;
|
|
*)
|
|
echo "store.sh: unknown command '$cmd'" >&2
|
|
usage
|
|
exit 2
|
|
;;
|
|
esac
|
|
}
|
|
|
|
main "$@"
|