fix(wake): #908 unify observed_seq on a single store-side allocator (dissolve detector-private-counter seam) (#915)

Co-authored-by: jason.woltje <jason@diversecanvas.com> Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-26 06:02:02 +00:00
parent 2378665eaf
commit e2ec927b1c
8 changed files with 487 additions and 157 deletions
--- a/packages/mosaic/framework/tools/wake/store.sh
+++ b/packages/mosaic/framework/tools/wake/store.sh
@@ -8,8 +8,11 @@
 #   §2.4  cursor semantics: observed_seq is authoritative; SHAs are descriptors.
 #
 # Three cursors:
-#   observed_seq  (detector-owned; this lib RECORDS/stores it) — monotonic int
-#                 assigned at observation. Authoritative. Source SHAs are
+#   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.
@@ -48,9 +51,23 @@ Usage: store.sh <command> [options]

 Commands:
  init                                   Create/repair the XDG state layout.
-  enqueue --seq N --class C [opts]       Record an observation into the durable
-                                         pending-inbox (coalesce or append).
-      --seq N            observed_seq (detector-assigned monotonic int). Required.
+  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 {}).
@@ -109,12 +126,18 @@ cmd_enqueue() {
    esac
  done

-  case "$seq" in
-  '' | *[!0-9]*)
-    echo "store.sh enqueue: --seq must be a non-negative integer" >&2
-    exit 2
-    ;;
-  esac
+  # --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"
@@ -142,13 +165,42 @@ cmd_enqueue() {

  _wake_init_dir "$STATE_DIR"

-  local consumed
-  consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
+  # --- 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
+  }

-  # An observation already inside the consumed prefix is idempotently ignored
-  # (§1.2: cursor never regresses; a re-observed already-consumed seq is a no-op).
+  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
-    return 0
+    _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).
@@ -159,6 +211,7 @@ cmd_enqueue() {
    --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
  }
@@ -168,6 +221,11 @@ cmd_enqueue() {
  #         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)"
@@ -176,20 +234,35 @@ cmd_enqueue() {
    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
-  printf '%s\n' "$new_pending" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/pending.jsonl"
+  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) ----
-  {
+  if ! {
    cat "$STATE_DIR/observed.set" 2>/dev/null
    printf '%s\n' "$seq"
-  } | grep -v '^[[:space:]]*$' | sort -n | uniq | _atomic_write "$STATE_DIR/observed.set"
+  } | 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 -------------------------------
-  local observed
-  observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
+  # --- 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.
  if [ "$seq" -gt "$observed" ]; then
    printf '%s' "$seq" | _atomic_write "$STATE_DIR/observed_seq"
  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() {