jason.woltje
13e6ce5e5c
Co-authored-by: jason.woltje <jason@diversecanvas.com> Co-committed-by: jason.woltje <jason@diversecanvas.com>
169 lines
8.2 KiB
Bash
Executable File
169 lines
8.2 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# _wake-common.sh — shared state resolution + atomic-write primitive for the
|
|
# wake/heartbeat durable queue (W2 of the wake canon, EPIC #892).
|
|
#
|
|
# CONTRACT ANCHORS (docs/scratchpads/heartbeat-planning/CONVERGED-DESIGN.md):
|
|
# §1.2 three-cursor durable queue; ALL state XDG, atomic write-tmp+rename.
|
|
# §2.3 durability is NEVER bypassed.
|
|
#
|
|
# This file is sourced by store.sh and ack.sh. It defines NO top-level actions;
|
|
# sourcing it is side-effect-free except for setting readonly path vars.
|
|
#
|
|
# Operator-agnostic (framework firewall): state location is derived purely from
|
|
# XDG / env. No operator paths, names, or secrets appear here.
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# State layout (XDG). §1.2: "all state XDG".
|
|
# base = ${WAKE_STATE_HOME:-${XDG_STATE_HOME:-$HOME/.local/state}/mosaic/wake}
|
|
# agent = ${WAKE_AGENT:-default} (per-agent queue namespace)
|
|
# STATE_DIR = <base>/<agent>
|
|
# ---------------------------------------------------------------------------
|
|
wake_state_dir() {
|
|
local base agent
|
|
base="${WAKE_STATE_HOME:-${XDG_STATE_HOME:-$HOME/.local/state}/mosaic/wake}"
|
|
agent="${WAKE_AGENT:-default}"
|
|
printf '%s/%s' "$base" "$agent"
|
|
}
|
|
|
|
# File names within STATE_DIR.
|
|
# observed_seq — highest observed_seq ever recorded (max monotonic int).
|
|
# consumed_seq — consumer cursor: top of the contiguous consumed prefix.
|
|
# observed.set — observed seqs in the live window (> consumed_seq), one int
|
|
# per line; the gap-detector's source of truth. Immune to
|
|
# coalescing (coalescing removes a pending ENTRY, never the
|
|
# fact that its seq was observed).
|
|
# pending.jsonl — the durable pending-inbox: one entry JSON object per line,
|
|
# {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).
|
|
|
|
# _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
|
|
# live file (only rename promotes content), so it can never corrupt a read.
|
|
_wake_tmp_prefix='.wake.tmp.'
|
|
|
|
# _atomic_write TARGET (content on stdin)
|
|
# §1.2 / task: EVERY state mutation is atomic write-tmp+rename. Write a temp
|
|
# file in the SAME directory (so mv is a same-filesystem atomic rename), fsync
|
|
# is best-effort, then rename over the target. A crash before the rename leaves
|
|
# the old target fully intact and a stale .wake.tmp.* that readers ignore.
|
|
_atomic_write() {
|
|
local target="$1" dir tmp
|
|
dir="$(dirname "$target")"
|
|
[ -d "$dir" ] || mkdir -p "$dir"
|
|
tmp="$(mktemp "$dir/${_wake_tmp_prefix}XXXXXX")" || return 1
|
|
if ! cat >"$tmp"; then
|
|
rm -f "$tmp"
|
|
return 1
|
|
fi
|
|
# Best-effort durability of the temp file before the rename. Not fatal if the
|
|
# platform lacks it — the rename atomicity is the load-bearing guarantee.
|
|
sync "$tmp" 2>/dev/null || true
|
|
if ! mv -f "$tmp" "$target"; then
|
|
rm -f "$tmp"
|
|
return 1
|
|
fi
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Enqueue serialization (single store-side observed_seq allocator, #908).
|
|
#
|
|
# store.sh is the SOLE allocator of observed_seq: it reads the observed_seq
|
|
# cursor, computes next = observed_seq + 1, and writes the pending record +
|
|
# observed.set + the cursor as ONE transaction. That read-modify-write MUST be
|
|
# serialized so two concurrent enqueues cannot read the same cursor and collide
|
|
# on a seq. These helpers take an exclusive lock for the duration of the
|
|
# transaction. flock (Linux/CI) is the load-bearing mechanism; where flock is
|
|
# absent the open still succeeds so a single-threaded enqueue is unaffected (the
|
|
# concurrency guarantee then degrades — the concurrency test SKIPs without flock,
|
|
# exactly as the detector's single-instance test already does).
|
|
#
|
|
# The lock uses a FIXED fd (8) within one store.sh process. Each store.sh
|
|
# invocation is its own process (the detector/reconciler call it as a
|
|
# subprocess), so fd 8 is always free here and never clashes with the detector
|
|
# run-loop lock (fd 9, a DIFFERENT process).
|
|
# ---------------------------------------------------------------------------
|
|
_wake_lock_acquire() {
|
|
# _wake_lock_acquire LOCKFILE — open fd 8 on LOCKFILE and take an exclusive
|
|
# (blocking) lock. Returns non-zero if the lock cannot be taken.
|
|
local lf="$1" dir
|
|
dir="$(dirname "$lf")"
|
|
[ -d "$dir" ] || mkdir -p "$dir" 2>/dev/null || true
|
|
# Open the lock fd. If the file does not yet exist and the dir is writable it
|
|
# is created; if the dir is read-only but the file exists, opening it O_WRONLY
|
|
# still succeeds (write perm on the file, not the dir).
|
|
# NB: a command-less `exec` redirection persists for the WHOLE shell, so we must
|
|
# NOT append `2>/dev/null` here (it would permanently silence the caller's
|
|
# stderr and swallow every later fail-loud diagnostic). Redirect only fd 8.
|
|
exec 8>"$lf" || return 1
|
|
if command -v flock >/dev/null 2>&1; then
|
|
flock 8 || return 1
|
|
fi
|
|
return 0
|
|
}
|
|
|
|
_wake_lock_release() {
|
|
# _wake_lock_release — drop the enqueue lock (closing fd 8 releases the flock).
|
|
# The `2>/dev/null` is SCOPED to the brace group (suppressing a "bad fd" close
|
|
# error) — it must NOT sit on a bare `exec`, where the redirection would persist
|
|
# for the whole shell and silence every later fail-loud diagnostic.
|
|
{ exec 8>&-; } 2>/dev/null || true
|
|
}
|
|
|
|
# _wake_clean_stale_tmp DIR — reap ORPHANED atomic-write temp files left by a
|
|
# crash mid-write (a crash before the rename leaves a .wake.tmp.* that no reader
|
|
# ever promotes). These are never the live store.
|
|
#
|
|
# AGE-SCOPED (#927): only tmp files whose mtime is older than
|
|
# ${WAKE_TMP_STALE_MIN:-5} minutes are removed. A LIVE in-flight atomic write's
|
|
# tmp is at most milliseconds old (mktemp -> cat -> sync -> rename all complete
|
|
# well under a second), so it can NEVER match this age filter. That is what makes
|
|
# the cleanup safe even if it ever overlaps a concurrent enqueue's atomic write:
|
|
# it deletes only DEMONSTRABLY-orphaned tmps, never another process's live write.
|
|
#
|
|
# This is why #927 is fixed: an UNCONDITIONAL delete of every .wake.tmp.* (the
|
|
# old behaviour) clobbered a concurrent enqueue's in-flight tmp -> spurious
|
|
# "durable pending write FAILED". It is ALSO no longer invoked from the per-
|
|
# enqueue hot path (see _wake_init_dir); it runs only at maintenance / daemon-
|
|
# start (store.sh init) and the detector poll tick, where accumulation is bounded
|
|
# once per pass rather than raced on every enqueue.
|
|
_wake_clean_stale_tmp() {
|
|
local dir="$1" min="${WAKE_TMP_STALE_MIN:-5}"
|
|
[ -d "$dir" ] || return 0
|
|
case "$min" in '' | *[!0-9]*) min=5 ;; esac
|
|
find "$dir" -maxdepth 1 -name "${_wake_tmp_prefix}*" -type f -mmin "+$min" -delete 2>/dev/null || true
|
|
}
|
|
|
|
# _wake_read_int FILE DEFAULT — read a single integer from FILE, or DEFAULT.
|
|
_wake_read_int() {
|
|
local file="$1" def="$2" val
|
|
if [ -f "$file" ]; then
|
|
val="$(tr -d '[:space:]' <"$file")"
|
|
case "$val" in
|
|
'' | *[!0-9]*) printf '%s' "$def" ;;
|
|
*) printf '%s' "$val" ;;
|
|
esac
|
|
else
|
|
printf '%s' "$def"
|
|
fi
|
|
}
|
|
|
|
# _wake_init_dir STATE_DIR — ensure the state layout exists; idempotent.
|
|
#
|
|
# #927: this runs on the HOT enqueue path (cmd_enqueue calls it BEFORE taking the
|
|
# enqueue lock) as well as on consume/cursors/ack. It must therefore NEVER touch
|
|
# another process's tmp files: the old _wake_clean_stale_tmp call here deleted a
|
|
# concurrent enqueue's LIVE in-flight tmp mid-write -> spurious durable-write
|
|
# abort. Stale-tmp reaping is now an explicit maintenance action (store.sh init /
|
|
# detector poll tick), NOT a side effect of ensuring the layout. Keep this
|
|
# function limited to creating the dir + seeding the cursor files.
|
|
_wake_init_dir() {
|
|
local dir="$1"
|
|
mkdir -p "$dir" || return 1
|
|
[ -f "$dir/observed_seq" ] || printf '0' | _atomic_write "$dir/observed_seq"
|
|
[ -f "$dir/consumed_seq" ] || printf '0' | _atomic_write "$dir/consumed_seq"
|
|
[ -f "$dir/observed.set" ] || printf '' | _atomic_write "$dir/observed.set"
|
|
[ -f "$dir/pending.jsonl" ] || printf '' | _atomic_write "$dir/pending.jsonl"
|
|
[ -f "$dir/ack-ledger.jsonl" ] || printf '' | _atomic_write "$dir/ack-ledger.jsonl"
|
|
}
|