CalvinBackup/gstack
1
0
Fork 0
mirror of https://github.com/garrytan/gstack.git synced 2026-09-11 07:29:00 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(brain-sync): per-record spool dir — the enqueue/drain race dies structurally

Browse Source 
Producers appended lines to .brain-queue.jsonl while the drain re-read and
os.replace'd it; the in-code comment admitted a lockless append between the
re-read and the replace was lost. Locks and rename-rotation designs were both
reviewed and rejected (each retained a tail race); the shipped design is a
maildir-style spool: one FILE per record in .brain-queue.d/ (tmp + atomic
rename), the drain snapshots filenames, processes, and deletes exactly what
it snapshotted. Writer and drainer never share an inode — nothing to race.

Semantics: at-least-once (a crash between process and unlink re-drains;
downstream content-hash dedup absorbs duplicates); retained (privacy-held)
records keep their files; unparseable records are kept + warned, never
destroyed. Legacy .brain-queue.jsonl migrates atomically on the next drain
(crash-leftover .migrating files recovered too); status/drop-queue count both
surfaces; discover-new writes spool records and advances its cursor
per-record-written. The preamble's queue-depth line switches to spool count
in this wave's template block.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-08-17 10:47:26 -07:00
co-authored by Claude Fable 5
parent e57b2798fd
commit 6df30370b3
4 changed files with 470 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/gstack-brain-sync
+233 -106
View File
@@ -20,6 +20,13 @@
set -uo pipefail
GSTACK_HOME="${GSTACK_HOME:-$HOME/.gstack}"
# Maildir-style spool: one FILE per record, <epoch>-<pid>-<uniq>.json.
# Writers (gstack-brain-enqueue, --discover-new) create records via tmp-file
# + atomic rename; the drain deletes exactly the files it snapshotted. No
# shared inode between writer and drainer → no append/rewrite race.
QUEUE_DIR="$GSTACK_HOME/.brain-queue.d"
# Legacy single-file queue: kept ONLY for migration. Pre-spool writers
# appended lines here; migrate_legacy_queue converts them to spool files.
QUEUE="$GSTACK_HOME/.brain-queue.jsonl"
ALLOWLIST="$GSTACK_HOME/.brain-allowlist"
PRIVACY_MAP="$GSTACK_HOME/.brain-privacy-map.json"
@@ -120,7 +127,84 @@ sys.exit(0)
"
}
# Compute matched allowlisted, privacy-filtered path set from queue.
# True (0) if the spool holds at least one record file.
spool_has_records() {
local f
for f in "$QUEUE_DIR"/*.json; do
[ -e "$f" ] && return 0
done
return 1
}
# Convert one legacy queue file's lines into spool record files (tmp +
# os.replace, one file per line). Reads the file TWICE before unlinking: a
# pre-rename writer can still append through its already-open fd after our
# rename, and those appends land in the renamed file — the second pass
# catches them (the tail race the shared-file design could never close).
# Unparseable lines migrate as-is; finalize_queue keeps + warns on them.
convert_legacy_file() {
local legacy="$1"
python3 - "$legacy" "$QUEUE_DIR" <<'PYEOF' 2>/dev/null || true
import os, sys, time
legacy, spool = sys.argv[1:3]
def read_lines(path):
try:
with open(path) as f:
return [l.rstrip("\r\n") for l in f if l.strip()]
except (FileNotFoundError, OSError):
return []
seq = 0
def write_spool(line):
global seq
seq += 1
tmp = os.path.join(spool, f".tmp-{os.getpid()}-m{seq}")
with open(tmp, "w") as f:
f.write(line + "\n")
os.replace(tmp, os.path.join(spool, f"{int(time.time())}-{os.getpid()}-m{seq}.json"))
written = set()
for _pass in (1, 2): # second read closes the pre-rename-fd tail race
for line in read_lines(legacy):
if line not in written: # identical duplicates collapse, as the old rewrite did
write_spool(line)
written.add(line)
os.unlink(legacy)
PYEOF
}
# Legacy migration (transition window only). If the single-file queue holds
# records, atomically rename it aside and convert each line to a spool file.
# A concurrent OLD writer that recreates a fresh legacy file after the rename
# simply gets migrated on the NEXT drain — nothing is lost, only deferred one
# boundary. Runs inside the run lock, before the drain reads the spool.
migrate_legacy_queue() {
local migrating="$QUEUE.migrating"
# Crash leftover: a prior migration renamed but died before unlink. Some of
# its lines may already exist as spool files — re-converting duplicates is
# safe (at-least-once; the drain dedups paths per snapshot and downstream
# content-hash dedup absorbs re-syncs). Losing the file would not be. If
# the conversion itself fails, the file stays for the next run (never rm a
# non-empty .migrating file outside convert_legacy_file's own unlink).
if [ -f "$migrating" ]; then
if [ -s "$migrating" ]; then
mkdir -p "$QUEUE_DIR" 2>/dev/null || return 0
convert_legacy_file "$migrating"
else
rm -f "$migrating" 2>/dev/null || true
fi
fi
if [ -s "$QUEUE" ]; then
mkdir -p "$QUEUE_DIR" 2>/dev/null || return 0
mv -f "$QUEUE" "$migrating" 2>/dev/null || return 0
convert_legacy_file "$migrating"
fi
return 0
}
# Compute matched allowlisted, privacy-filtered path set from the spool.
# Output: newline-delimited relative paths that should be staged.
#
# #2549: every non-staged queue entry is CLASSIFIED, never silently discarded.
@@ -132,13 +216,20 @@ sys.exit(0)
# attempt, no allowlist glob, not on disk) and are removed WITH a counted
# status — the old behavior truncated the whole queue and reported every one
# of these, including privacy holds, as "no allowlisted changes".
#
# Spool snapshot ($3): the sorted list of spool record filenames read here is
# written to the snapshot manifest, one filename per line. finalize_queue
# deletes exactly the manifest's files and never touches records created
# after this listing — a concurrent enqueue is a separate file by
# construction, so it simply rides to the next drain.
compute_paths_to_stage() {
local mode="$1"
local class_file="${2:-}"
python3 - "$GSTACK_HOME" "$QUEUE" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" <<'PYEOF'
local snapshot_file="${3:-}"
python3 - "$GSTACK_HOME" "$QUEUE_DIR" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" "$snapshot_file" <<'PYEOF'
import sys, json, os, fnmatch, glob
gstack_home, queue, allowlist_path, privacy_path, skip_path, mode, class_file = sys.argv[1:8]
gstack_home, spool_dir, allowlist_path, privacy_path, skip_path, mode, class_file, snapshot_file = sys.argv[1:9]
def load_lines(path):
try:
@@ -164,23 +255,38 @@ privacy_map = load_privacy_map(privacy_path)
# discover_new — otherwise an explicitly-skipped file gets committed.
skip_lines = {s.replace(os.sep, "/") for s in load_lines(skip_path)}
# Read queue; collect unique file paths.
queue_paths = set()
# Snapshot the spool: sorted (= chronological, filenames are epoch-first)
# list of record files at read time. Records that appear after this listing
# belong to the NEXT drain. Files we cannot read stay OUT of the manifest so
# finalize never deletes a record this drain didn't actually consume.
try:
with open(queue) as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
p = obj.get("file")
if isinstance(p, str):
queue_paths.add(p)
except json.JSONDecodeError:
continue
except FileNotFoundError:
pass
snapshot = sorted(n for n in os.listdir(spool_dir) if n.endswith(".json"))
except (FileNotFoundError, NotADirectoryError):
snapshot = []
queue_paths = set()
consumed = []
for name in snapshot:
try:
with open(os.path.join(spool_dir, name)) as f:
line = f.readline().strip()
except OSError:
continue
consumed.append(name)
if not line:
continue
try:
obj = json.loads(line)
p = obj.get("file")
if isinstance(p, str):
queue_paths.add(p)
except json.JSONDecodeError:
continue # unparseable record: finalize keeps + warns
if snapshot_file:
with open(snapshot_file, "w") as f:
for name in consumed:
f.write(name + "\n")
def path_matches_any(path, globs):
for pattern in globs:
@@ -241,22 +347,26 @@ for p in final:
PYEOF
}
# #2549: surgical queue rewrite — replaces every whole-queue truncation
# (`: > "$QUEUE"`). Re-reads the LIVE queue at rewrite time (a writer may have
# enqueued while we were staging/pushing — those entries must survive; the old
# truncation destroyed them) and keeps every line whose file is either
# retained (privacy/mode-held) or not part of this drain at all. Atomic
# tmp+mv in the same directory. Dropped-path detail goes to a 0600 sidecar so
# the status line can stay content-free (counts only).
rewrite_queue() {
local paths_file="$1" # staged (drained) paths, one per line
local class_file="$2" # classification JSON from compute_paths_to_stage
# Fail-open by design (a failed rewrite self-corrects next run: re-stage →
# Finalize the drain: delete exactly the spool record files this drain
# consumed (per the snapshot manifest), keeping retained (privacy/mode-held)
# and unparseable records queued. The predecessor (a shared-file queue
# rewrite) had a lockless-append race between its live re-read and the
# os.replace; with one file per record that race class is structurally gone —
# a concurrent enqueue is a separate file the snapshot never listed, so
# finalize cannot touch it. Crash semantics are at-least-once: a drain that
# dies before finalize leaves its spool files in place and the next run
# re-drains them; downstream content-hash dedup absorbs the duplicates.
# Dropped-path detail goes to a 0600 sidecar so the status line can stay
# content-free (counts only).
finalize_queue() {
local snapshot_file="$1" # spool filenames this drain consumed, one per line
local class_file="$2" # classification JSON from compute_paths_to_stage
# Fail-open by design (a failed finalize self-corrects next run: re-stage →
# nothing-to-commit), but say so — a silent failure here would let the
# subsequent "ok/idle" status claim a drain that did not happen.
python3 - "$QUEUE" "$paths_file" "$class_file" "$GSTACK_HOME/.brain-sync-drops.json" <<'PYEOF' || echo "BRAIN_SYNC: warning: queue rewrite failed — entries retained; next run re-drains" >&2
python3 - "$QUEUE_DIR" "$snapshot_file" "$class_file" "$GSTACK_HOME/.brain-sync-drops.json" <<'PYEOF' || echo "BRAIN_SYNC: warning: queue finalize failed — entries retained; next run re-drains" >&2
import json, os, sys, time
queue, paths_file, class_file, drops_file = sys.argv[1:5]
spool_dir, snapshot_file, class_file, drops_file = sys.argv[1:5]
def lines(path):
try:
@@ -265,7 +375,6 @@ def lines(path):
except FileNotFoundError:
return []
staged = set(lines(paths_file))
try:
with open(class_file) as f:
classified = json.load(f)
@@ -275,36 +384,31 @@ retained = set(classified.get("retained", []))
dropped = set()
for group in (classified.get("dropped", {}) or {}).values():
dropped.update(group)
processed = staged | dropped
kept = []
seen_lines = set()
unparseable = 0
# LIVE re-read narrows (not fully closes) the concurrent-append window: the
# lockless enqueue can still land on the old inode between this read and the
# os.replace below. Vastly better than the old whole-queue truncation.
for line in lines(queue):
if line in seen_lines:
continue # identical duplicate lines collapse on rewrite
for name in lines(snapshot_file):
full = os.path.join(spool_dir, name)
try:
p = json.loads(line).get("file")
with open(full) as f:
rec = f.readline().strip()
except OSError:
continue # unreadable now: leave it for the next drain
p = None
try:
p = json.loads(rec).get("file")
except Exception:
unparseable += 1
kept.append(line) # unparseable line: keep, never destroy
seen_lines.add(line)
pass
if not isinstance(p, str):
unparseable += 1 # keep — never destroy what we can't read
continue
if not isinstance(p, str) or p in retained or p not in processed:
kept.append(line)
seen_lines.add(line)
if p in retained:
continue # stays queued: syncs under a higher mode
try:
os.unlink(full) # staged or dropped: fully processed
except FileNotFoundError:
pass
if unparseable:
import sys as _sys
print(f"BRAIN_SYNC: {unparseable} unparseable queue line(s) held (inspect {queue})", file=_sys.stderr)
tmp = queue + ".tmp." + str(os.getpid())
with open(tmp, "w") as f:
for l in kept:
f.write(l + "\n")
os.replace(tmp, queue)
print(f"BRAIN_SYNC: {unparseable} unparseable spool record(s) held (inspect {spool_dir})", file=sys.stderr)
if dropped:
fd = os.open(drops_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
@@ -374,6 +478,10 @@ subcmd_once() {
# the lock removal.
trap 'rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM
# Convert any legacy single-file queue lines into spool records before the
# drain reads the spool (transition window for pre-spool writers).
migrate_legacy_queue
local mode
mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off)
@@ -449,27 +557,31 @@ subcmd_once() {
fi
# Empty-queue fast path: this is the steady state at every skill boundary.
# Skipping compute/rewrite here is safe — with zero queue lines there is
# nothing to classify, retain, or drop, and a concurrent append after this
# check simply waits for the next boundary. (The detector above already ran:
# its whole point is re-pushing stranded commits when the queue is empty.)
# The lock-release trap installed at acquisition covers this exit.
if [ ! -s "$QUEUE" ]; then
# Skipping compute/finalize here is safe — with zero spool records there is
# nothing to classify, retain, or drop, and a record created after this
# check simply waits for the next boundary. The legacy file is checked too:
# an OLD writer may have recreated it after the migration above (it gets
# migrated next run, but the depth is honest now). (The detector above
# already ran: its whole point is re-pushing stranded commits when the
# queue is empty.) The lock-release trap installed at acquisition covers
# this exit.
if ! spool_has_records && [ ! -s "$QUEUE" ]; then
write_status "idle" "queue empty"
exit 0
fi
local paths_file class_file
local paths_file class_file snapshot_file
paths_file=$(mktemp /tmp/brain-sync-paths.XXXXXX) || { rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; }
class_file=$(mktemp /tmp/brain-sync-class.XXXXXX) || { rm -f "$paths_file"; rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; }
snapshot_file=$(mktemp /tmp/brain-sync-snapshot.XXXXXX) || { rm -f "$paths_file" "$class_file"; rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; }
# Single trap covers all: lock cleanup AND tempfile cleanup.
trap 'rm -f "$paths_file" "$class_file" 2>/dev/null; rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM
trap 'rm -f "$paths_file" "$class_file" "$snapshot_file" 2>/dev/null; rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM
compute_paths_to_stage "$mode" "$class_file" > "$paths_file"
compute_paths_to_stage "$mode" "$class_file" "$snapshot_file" > "$paths_file"
if [ ! -s "$paths_file" ]; then
# Nothing stageable. Rewrite the queue (retained entries + concurrent
# appends survive; classified drops removed) instead of truncating it.
rewrite_queue "$paths_file" "$class_file"
# Nothing stageable. Finalize the snapshot (retained entries survive;
# classified drops removed; records created after the snapshot untouched).
finalize_queue "$snapshot_file" "$class_file"
local summary
summary=$(queue_summary "$class_file")
write_status "idle" "no stageable changes${summary:+ ($summary)}"
@@ -520,8 +632,8 @@ subcmd_once() {
git -C "$GSTACK_HOME" -c user.email="gstack@localhost" -c user.name="gstack-brain-sync" \
commit -q -m "$msg" 2>/dev/null || {
# Nothing to commit (e.g. all files already committed). The drained
# paths leave the queue; retained + concurrent entries survive (#2549).
rewrite_queue "$paths_file" "$class_file"
# records leave the spool; retained + post-snapshot records survive.
finalize_queue "$snapshot_file" "$class_file"
write_status "idle" "queue drained but no new changes to commit"
exit 0
}
@@ -535,10 +647,10 @@ subcmd_once() {
hint=$(remote_auth_hint)
write_status "push_failed" "push failed: auth error; commit retained locally, will retry next run. fix: $hint"
echo "BRAIN_SYNC: push failed: auth. fix: $hint" >&2
# Drained paths leave the queue — they live in the local commit, which
# Drained records leave the spool — they live in the local commit, which
# the run-start detector re-pushes next time (#2549). Retained +
# concurrent entries survive the rewrite.
rewrite_queue "$paths_file" "$class_file"
# post-snapshot records survive the finalize.
finalize_queue "$snapshot_file" "$class_file"
exit 0
fi
@@ -552,7 +664,7 @@ subcmd_once() {
if git -C "$GSTACK_HOME" merge --no-edit "origin/$branch" >/dev/null 2>&1; then
if GSTACK_HOME="$GSTACK_HOME" _receipted_git closed brain-sync "$push_host" curated-memory-git-push "artifacts_sync_mode!=off" \
bash -c 'git -C "$1" push origin HEAD 2>/dev/null' _ "$GSTACK_HOME"; then
rewrite_queue "$paths_file" "$class_file"
finalize_queue "$snapshot_file" "$class_file"
date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
write_status "ok" "pushed $n file(s) after rebase"
exit 0
@@ -561,12 +673,12 @@ subcmd_once() {
fi
# Commit exists locally; the run-start detector re-pushes it next time.
write_status "push_failed" "push failed: $(printf '%s' "$push_err" | head -1); commit retained locally, will retry next run"
rewrite_queue "$paths_file" "$class_file"
finalize_queue "$snapshot_file" "$class_file"
exit 0
}
# Success: drained paths leave the queue (retained + concurrent survive).
rewrite_queue "$paths_file" "$class_file"
# Success: drained records leave the spool (retained + post-snapshot survive).
finalize_queue "$snapshot_file" "$class_file"
date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
write_status "ok" "pushed $n file(s)"
exit 0
@@ -578,9 +690,13 @@ subcmd_status() {
else
echo '{"status":"unknown","message":"no status file yet"}'
fi
# Supplemental info (not in status file).
local queue_depth=0
[ -f "$QUEUE" ] && queue_depth=$(wc -l < "$QUEUE" | tr -d ' ')
# Supplemental info (not in status file). Depth = spool record files plus
# any not-yet-migrated legacy queue lines (transition window).
local queue_depth spool_depth legacy_depth
spool_depth=$(ls "$QUEUE_DIR"/*.json 2>/dev/null | wc -l | tr -d ' ')
legacy_depth=0
[ -f "$QUEUE" ] && legacy_depth=$(wc -l < "$QUEUE" | tr -d ' ')
queue_depth=$(( spool_depth + legacy_depth ))
local last_push="never"
[ -f "$LAST_PUSH_FILE" ] && last_push=$(cat "$LAST_PUSH_FILE" 2>/dev/null || echo never)
local mode
@@ -611,13 +727,22 @@ subcmd_drop_queue() {
echo "Refusing: --drop-queue discards pending syncs. Pass --yes to confirm." >&2
exit 1
fi
if [ ! -f "$QUEUE" ]; then
# Remove spool record files, then truncate any legacy queue remnant.
local n=0 f
for f in "$QUEUE_DIR"/*.json; do
[ -e "$f" ] || continue
rm -f "$f" 2>/dev/null && n=$(( n + 1 ))
done
if [ -f "$QUEUE" ]; then
local legacy_n
legacy_n=$(wc -l < "$QUEUE" | tr -d ' ')
n=$(( n + legacy_n ))
: > "$QUEUE"
fi
if [ "$n" -eq 0 ]; then
echo "queue already empty"
exit 0
fi
local n
n=$(wc -l < "$QUEUE" | tr -d ' ')
: > "$QUEUE"
echo "dropped $n queue entries"
}
@@ -627,11 +752,11 @@ subcmd_discover_new() {
fi
# Walk allowlist globs; enqueue any file where mtime+size differs from cursor.
python3 - "$GSTACK_HOME" "$ALLOWLIST" "$DISCOVER_CURSOR" <<'PYEOF' 2>/dev/null || true
import sys, os, json, fnmatch
import sys, os, json, fnmatch, time
from datetime import datetime, timezone
gstack_home, allowlist_path, cursor_path = sys.argv[1:4]
queue_path = os.path.join(gstack_home, ".brain-queue.jsonl")
spool_dir = os.path.join(gstack_home, ".brain-queue.d")
skip_path = os.path.join(gstack_home, ".brain-skip.txt")
def load_lines(path):
@@ -689,34 +814,36 @@ for root, dirs, files in os.walk(gstack_home):
if cursor.get(rel) != key:
to_enqueue.append((rel, key))
# Append to the queue directly. The previous implementation shelled out to
# Write spool records directly. The previous implementation shelled out to
# gstack-brain-enqueue once per file, but Windows Python cannot exec a
# bash-shebang script (the spawn fails with a fork error), so discovery
# enqueued nothing on Windows even after the path-match fix above.
# Writing the queue line here is platform-agnostic; the drain step
# Writing the record here is platform-agnostic; the drain step
# (compute_paths_to_stage) still re-applies the skip-list + privacy filters.
if to_enqueue:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
written = []
try:
# One atomic append per record (O_APPEND, each line < PIPE_BUF), matching
# gstack-brain-enqueue's concurrency contract so a writer-shim append
# running in parallel can't interleave mid-record. Buffered text writes
# don't guarantee that. Compact separators match the shim's JSON shape.
fd = os.open(queue_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
try:
for rel, key in to_enqueue:
rec = json.dumps({"file": rel, "ts": ts}, separators=(",", ":"))
os.write(fd, (rec + "\n").encode("utf-8"))
finally:
os.close(fd)
# One spool FILE per record (tmp write + atomic os.replace), matching
# gstack-brain-enqueue's maildir contract: writers and the drain never
# share an inode, so a parallel writer or drain can't race this.
# Compact separators match the shim's JSON shape.
os.makedirs(spool_dir, exist_ok=True)
for i, (rel, key) in enumerate(to_enqueue):
rec = json.dumps({"file": rel, "ts": ts}, separators=(",", ":"))
tmp = os.path.join(spool_dir, f".tmp-{os.getpid()}-d{i}")
with open(tmp, "w") as f:
f.write(rec + "\n")
os.replace(tmp, os.path.join(spool_dir, f"{int(time.time())}-{os.getpid()}-d{i}.json"))
written.append((rel, key))
except OSError:
# Queue write failed (disk full, AV file lock). Leave the cursor
# unadvanced so these files are retried on the next discover instead of
# being silently recorded as synced (which loses the change until the
# file next changes).
to_enqueue = []
# Spool write failed (disk full, AV file lock). Leave the cursor
# unadvanced for unwritten records so they are retried on the next
# discover instead of being silently recorded as synced (which loses
# the change until the file next changes).
pass
# Advance the cursor only for records actually written.
for rel, key in to_enqueue:
for rel, key in written:
new_cursor[rel] = key
save_cursor(cursor_path, new_cursor)