jason.woltje
28f022d9c0
Co-authored-by: jason.woltje <jason@diversecanvas.com> Co-committed-by: jason.woltje <jason@diversecanvas.com>
270 lines
13 KiB
Bash
Executable File
270 lines
13 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# test-wake-store-ack.sh — RED-FIRST invariant harness for W2 (EPIC #892):
|
|
# three-cursor durable store (store.sh, A2) + RECEIVED/CONSUMED ack-wrapper
|
|
# (ack.sh, A4).
|
|
#
|
|
# Each test asserts ONE CONVERGED-DESIGN invariant and is designed to go RED if
|
|
# that invariant regresses:
|
|
# T1 three-cursor advancement (§1.2/§2.4)
|
|
# T2 digest-coalesce-REPLACE vs actionable-APPEND (§1.2/§2.3)
|
|
# T3 contiguous-prefix CONSUMED / gap rejection (§1.2/§2.2)
|
|
# T4 wake_id DELIVERY-dedup (never re-action) (§2.2)
|
|
# T5 atomic write-tmp+rename survives crash mid-write (§1.2)
|
|
# T6 durability survives restart (retain until CONSUMED) (§2.3)
|
|
# T7 ack local-write NEVER blocks on network (§2.2)
|
|
#
|
|
# Isolated: every test runs against a fresh WAKE_STATE_HOME temp dir.
|
|
set -uo pipefail
|
|
|
|
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
|
STORE="$SCRIPT_DIR/store.sh"
|
|
ACK="$SCRIPT_DIR/ack.sh"
|
|
|
|
command -v jq >/dev/null 2>&1 || {
|
|
echo "SKIP: jq not available" >&2
|
|
exit 0
|
|
}
|
|
|
|
TMP_ROOT="$(mktemp -d)"
|
|
trap 'rm -rf "$TMP_ROOT"' EXIT
|
|
|
|
# Failures are recorded to a marker FILE, not a shell var: each test runs in a
|
|
# subshell (for env isolation) and a subshell cannot mutate a parent variable,
|
|
# so a var-based counter would silently swallow failures (the exact bug this
|
|
# harness must never have).
|
|
FAILFILE="$TMP_ROOT/failures"
|
|
: >"$FAILFILE"
|
|
pass=0
|
|
fail_msg() {
|
|
echo " FAIL: $*" >&2
|
|
echo "x" >>"$FAILFILE"
|
|
}
|
|
ok() { pass=$((pass + 1)); }
|
|
|
|
# jq_any FILE FILTER — true iff any JSONL record in FILE matches FILTER. Uses
|
|
# slurp (-s), NOT `jq -e select` (which reflects only the LAST input's value
|
|
# over a multi-line JSONL and would misreport presence).
|
|
jq_any() {
|
|
local file="$1" filter="$2"
|
|
[ -f "$file" ] || return 1
|
|
[ "$(jq -s "[.[] | select($filter)] | length" "$file" 2>/dev/null || echo 0)" -gt 0 ]
|
|
}
|
|
# stream_any — same, but reads JSONL from stdin.
|
|
stream_any() {
|
|
local filter="$1"
|
|
[ "$(jq -s "[.[] | select($filter)] | length" 2>/dev/null || echo 0)" -gt 0 ]
|
|
}
|
|
|
|
# fresh_state NAME — echoes a fresh isolated state home for export.
|
|
fresh_state() {
|
|
local d="$TMP_ROOT/$1"
|
|
rm -rf "$d"
|
|
mkdir -p "$d"
|
|
printf '%s' "$d"
|
|
}
|
|
|
|
echo "== T1: three-cursor advancement =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t1)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
"$STORE" enqueue --seq 1 --class actionable --locators '{"repo":"r","issue":1}' >/dev/null
|
|
"$STORE" enqueue --seq 2 --class actionable --locators '{"repo":"r","issue":2}' >/dev/null
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'observed_seq=2' || fail_msg "T1: observed_seq should be 2 after enqueue 1,2 [$cur]"
|
|
echo "$cur" | grep -q 'consumed_seq=0' || fail_msg "T1: consumed_seq must NOT advance on enqueue (only on CONSUMED ack) [$cur]"
|
|
echo "$cur" | grep -q 'pending_depth=2' || fail_msg "T1: pending depth should be 2 [$cur]"
|
|
# consumed_seq advances only on CONSUMED.
|
|
"$STORE" consume --upto 2 >/dev/null
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'consumed_seq=2' || fail_msg "T1: consumed_seq should be 2 after CONSUMED 2 [$cur]"
|
|
echo "$cur" | grep -q 'pending_depth=0' || fail_msg "T1: pending drained after CONSUMED [$cur]"
|
|
) && ok
|
|
|
|
echo "== T2: digest coalesce-REPLACE vs actionable APPEND =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t2)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
"$STORE" enqueue --seq 1 --class digest --locators '{"head":"aaa"}' >/dev/null
|
|
"$STORE" enqueue --seq 2 --class actionable --locators '{"issue":7}' >/dev/null
|
|
"$STORE" enqueue --seq 3 --class digest --locators '{"head":"bbb"}' >/dev/null
|
|
"$STORE" enqueue --seq 4 --class human --locators '{"from":"peer"}' >/dev/null
|
|
out="$("$STORE" drain)"
|
|
ndigest="$(printf '%s\n' "$out" | jq -c 'select(.class=="digest")' | grep -c . || true)"
|
|
[ "$ndigest" = "1" ] || fail_msg "T2: digest must COALESCE to a single pending entry, got $ndigest"
|
|
head="$(printf '%s\n' "$out" | jq -r 'select(.class=="digest") | .locators.head')"
|
|
[ "$head" = "bbb" ] || fail_msg "T2: newest digest must REPLACE prior (expected bbb, got $head)"
|
|
nactionable="$(printf '%s\n' "$out" | jq -c 'select(.class=="actionable")' | grep -c . || true)"
|
|
nhuman="$(printf '%s\n' "$out" | jq -c 'select(.class=="human")' | grep -c . || true)"
|
|
[ "$nactionable" = "1" ] || fail_msg "T2: actionable must APPEND (never replaced), got $nactionable"
|
|
[ "$nhuman" = "1" ] || fail_msg "T2: human must APPEND / stay durable, got $nhuman"
|
|
# Durability never bypassed: all classes present in the durable store.
|
|
total="$(printf '%s\n' "$out" | grep -c . || true)"
|
|
[ "$total" = "3" ] || fail_msg "T2: durable store should hold digest+actionable+human = 3, got $total"
|
|
) && ok
|
|
|
|
echo "== T3: contiguous-prefix CONSUMED (reject ack N while N-1 unconsumed) =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t3)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
# Gap: observe seq 1 and 3, but NOT 2.
|
|
"$STORE" enqueue --seq 1 --class actionable --locators '{}' >/dev/null
|
|
"$STORE" enqueue --seq 3 --class actionable --locators '{}' >/dev/null
|
|
if "$STORE" consume --upto 3 >/dev/null 2>&1; then
|
|
fail_msg "T3: CONSUMED 3 must be REJECTED while seq 2 is a gap (unconsumed)"
|
|
fi
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'consumed_seq=0' || fail_msg "T3: rejected CONSUMED must NOT advance the cursor [$cur]"
|
|
# Also reject via the ack wrapper.
|
|
if "$ACK" consumed --upto 3 --no-sync >/dev/null 2>&1; then
|
|
fail_msg "T3: ack.sh CONSUMED 3 must be REJECTED over a gap"
|
|
fi
|
|
# Fill the gap; now the contiguous prefix is ackable.
|
|
"$STORE" enqueue --seq 2 --class actionable --locators '{}' >/dev/null
|
|
"$ACK" consumed --upto 3 --no-sync >/dev/null 2>&1 || fail_msg "T3: CONSUMED 3 should succeed once gap at 2 is filled"
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'consumed_seq=3' || fail_msg "T3: consumed_seq should be 3 after contiguous prefix filled [$cur]"
|
|
# Cumulative + can't regress: CONSUMED 2 after 3 is an idempotent no-op.
|
|
"$ACK" consumed --upto 2 --no-sync >/dev/null 2>&1 || fail_msg "T3: cumulative CONSUMED 2 (<=3) should be an idempotent success"
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'consumed_seq=3' || fail_msg "T3: cursor must not regress below 3 [$cur]"
|
|
) && ok
|
|
|
|
echo "== T4: wake_id DELIVERY-dedup (dup delivery = re-RECEIVE, never re-action) =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t4)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
"$STORE" enqueue --seq 1 --class digest --locators '{}' >/dev/null
|
|
r1="$("$ACK" received --wake-id WID-abc)"
|
|
[ "$r1" = "RECEIVED" ] || fail_msg "T4: first delivery should be RECEIVED, got '$r1'"
|
|
r2="$("$ACK" received --wake-id WID-abc)"
|
|
[ "$r2" = "DUP" ] || fail_msg "T4: duplicate delivery of same wake_id should be DUP (re-RECEIVE), got '$r2'"
|
|
# RECEIVED (delivery) must NEVER advance consumed_seq (delivery != consumption).
|
|
cur="$("$STORE" cursors)"
|
|
echo "$cur" | grep -q 'consumed_seq=0' || fail_msg "T4: RECEIVED must not advance consumed_seq [$cur]"
|
|
) && ok
|
|
|
|
echo "== T5: atomic write-tmp+rename survives crash mid-write =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t5)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
"$STORE" enqueue --seq 1 --class actionable --locators '{"issue":1}' >/dev/null
|
|
STATE_DIR="$WAKE_STATE_HOME/default"
|
|
before="$(cat "$STATE_DIR/pending.jsonl")"
|
|
# Simulate a crash mid-write: a leftover atomic-write temp file with GARBAGE
|
|
# that was never renamed over the live file.
|
|
printf '{"observed_seq": 999, TRUNCATED-GARBAGE' >"$STATE_DIR/.wake.tmp.crash12"
|
|
# The live store must be untouched and still valid JSON.
|
|
after="$(cat "$STATE_DIR/pending.jsonl")"
|
|
[ "$before" = "$after" ] || fail_msg "T5: live pending.jsonl changed by a crash temp file"
|
|
drained="$("$STORE" drain)"
|
|
printf '%s\n' "$drained" | jq -e . >/dev/null 2>&1 || fail_msg "T5: drain returned non-JSON — garbage temp corrupted the store"
|
|
printf '%s\n' "$drained" | stream_any '.observed_seq==1' || fail_msg "T5: committed entry (seq 1) lost after simulated crash"
|
|
printf '%s\n' "$drained" | grep -q 999 && fail_msg "T5: uncommitted garbage (seq 999) leaked into the live store"
|
|
# A subsequent real mutation must succeed and clean the stale temp.
|
|
"$STORE" enqueue --seq 2 --class actionable --locators '{"issue":2}' >/dev/null || fail_msg "T5: enqueue after crash temp failed"
|
|
[ -e "$STATE_DIR/.wake.tmp.crash12" ] && fail_msg "T5: stale crash temp not cleaned by next mutation"
|
|
d2="$("$STORE" drain)"
|
|
[ "$(printf '%s\n' "$d2" | grep -c .)" = "2" ] || fail_msg "T5: store not healthy after crash+recovery (expected 2 entries)"
|
|
) && ok
|
|
|
|
echo "== T5b: _atomic_write is tmp+rename (killing the writer mid-write leaves the target intact) =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t5b)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
# shellcheck disable=SC1091
|
|
. "$SCRIPT_DIR/_wake-common.sh"
|
|
d="$WAKE_STATE_HOME/atomic"
|
|
mkdir -p "$d"
|
|
target="$d/file"
|
|
printf 'GOOD\n' | _atomic_write "$target"
|
|
# Feed a writer via a FIFO but NEVER send EOF, then kill it mid-write. With
|
|
# true tmp+rename the partial bytes land in a temp file and the rename never
|
|
# runs, so `target` keeps its committed content. A direct (non-atomic) write
|
|
# would have streamed the garbage straight into `target`.
|
|
fifo="$d/fifo"
|
|
mkfifo "$fifo"
|
|
( _atomic_write "$target" <"$fifo" ) &
|
|
wpid=$!
|
|
exec 9>"$fifo"
|
|
printf 'PARTIAL-GARBAGE-NO-EOF' >&9
|
|
sleep 0.3
|
|
pkill -9 -P "$wpid" 2>/dev/null || true
|
|
kill -9 "$wpid" 2>/dev/null || true
|
|
exec 9>&-
|
|
wait "$wpid" 2>/dev/null || true
|
|
got="$(cat "$target" 2>/dev/null)"
|
|
[ "$got" = "GOOD" ] || fail_msg "T5b: target corrupted by a killed mid-write (got '$got') — write is not tmp+rename atomic"
|
|
) && ok
|
|
|
|
echo "== T6: durability survives restart (retain until CONSUMED) =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t6)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
# "Session 1": enqueue then vanish (no in-memory state carried).
|
|
"$STORE" enqueue --seq 1 --class human --locators '{"from":"peer"}' >/dev/null
|
|
"$STORE" enqueue --seq 2 --class actionable --locators '{"issue":9}' >/dev/null
|
|
# "Session 2": a brand-new process (this subshell invocation of store.sh) must
|
|
# see the persisted entries — nothing was held in memory.
|
|
d="$("$STORE" drain)"
|
|
[ "$(printf '%s\n' "$d" | grep -c .)" = "2" ] || fail_msg "T6: entries not retained across restart (expected 2)"
|
|
printf '%s\n' "$d" | stream_any '.class=="human"' || fail_msg "T6: a human message was lost across restart (durability bypassed)"
|
|
# Retained UNTIL consumed: after CONSUMED they are released.
|
|
"$STORE" consume --upto 2 >/dev/null
|
|
d2="$("$STORE" drain)"
|
|
n2="$(printf '%s\n' "$d2" | grep -c . || true)"
|
|
[ "$n2" = "0" ] || fail_msg "T6: entries should be released only AFTER CONSUMED (still $n2 pending)"
|
|
) && ok
|
|
|
|
echo "== T7: ack local-write NEVER blocks on network =="
|
|
(
|
|
WAKE_STATE_HOME="$(fresh_state t7)"
|
|
export WAKE_STATE_HOME
|
|
unset WAKE_AGENT
|
|
"$STORE" enqueue --seq 1 --class digest --locators '{}' >/dev/null
|
|
# A network-shaped shim on PATH: if the SYNCHRONOUS ack path ever invoked the
|
|
# network directly, this marker would appear before ack returns.
|
|
bin="$TMP_ROOT/t7bin"
|
|
mkdir -p "$bin"
|
|
for netcmd in curl wget nc ssh; do
|
|
cat >"$bin/$netcmd" <<EOF
|
|
#!/usr/bin/env bash
|
|
touch "$WAKE_STATE_HOME/NET_CALLED_SYNC"
|
|
exit 0
|
|
EOF
|
|
chmod +x "$bin/$netcmd"
|
|
done
|
|
# Background sync deliberately SLOW: proves the ack does not wait for it.
|
|
export WAKE_ACK_SYNC_CMD="sleep 3; touch '$WAKE_STATE_HOME/SYNC_DONE'"
|
|
start="$(date +%s)"
|
|
out="$(PATH="$bin:$PATH" "$ACK" consumed --upto 1 --wake-id WID-1)"
|
|
end="$(date +%s)"
|
|
elapsed=$((end - start))
|
|
[ "$elapsed" -lt 2 ] || fail_msg "T7: ack path blocked ${elapsed}s — must return without waiting on the background sync"
|
|
echo "$out" | grep -q 'CONSUMED 1' || fail_msg "T7: CONSUMED not reported [$out]"
|
|
# Local write is durable IMMEDIATELY (no network needed to record the ack).
|
|
STATE_DIR="$WAKE_STATE_HOME/default"
|
|
jq_any "$STATE_DIR/ack-ledger.jsonl" '.type=="CONSUMED" and .upto==1' ||
|
|
fail_msg "T7: CONSUMED not written to the local ledger synchronously"
|
|
consumed_now="$("$STORE" cursors | sed -n 's/consumed_seq=//p')"
|
|
[ "$consumed_now" = "1" ] || fail_msg "T7: consumed_seq not advanced by the local ack write (got '$consumed_now')"
|
|
# The synchronous path must NOT have touched the network.
|
|
[ -e "$WAKE_STATE_HOME/NET_CALLED_SYNC" ] && fail_msg "T7: a network command was invoked on the SYNCHRONOUS ack path"
|
|
# And the slow sync must still be in flight (proves it was truly backgrounded).
|
|
[ -e "$WAKE_STATE_HOME/SYNC_DONE" ] && fail_msg "T7: background sync finished synchronously — it was not detached"
|
|
true
|
|
) && ok
|
|
|
|
echo
|
|
if [ -s "$FAILFILE" ]; then
|
|
echo "wake store/ack harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2
|
|
exit 1
|
|
fi
|
|
echo "wake store/ack harness: all invariants passed ($pass groups)"
|