fix(gbrain-sync): configurable timeouts + resume from gbrain checkpoint (#1611)

The memory and code stages hardcoded a 35-min spawn timeout. On brains with
~2000+ staged files, /sync-gbrain --full reliably SIGTERM'd the child at
exactly 35 minutes with exit 143. gbrain left ~/.gbrain/import-checkpoint.json
pointing at the staging dir, but gstack-memory-ingest's SIGTERM handler
unconditionally cleaned the dir up — so the next run found a checkpoint
pointing at nothing and restaged from scratch, repeating the SIGTERM forever.

Three changes:

1. Configurable timeouts via env (bounds 60_000ms - 86_400_000ms, default
   2_100_000ms = 35min unchanged):
     GSTACK_SYNC_MEMORY_TIMEOUT_MS
     GSTACK_SYNC_CODE_TIMEOUT_MS
   Out-of-range or non-numeric values warn and fall back to the default.

2. SIGTERM in gstack-memory-ingest no longer always cleans up the staging
   dir. If gbrain has written ~/.gbrain/import-checkpoint.json pointing at
   the active staging dir, the dir is PRESERVED for next-run resume.
   Otherwise (no checkpoint pointing here, crash before gbrain ever
   touched it) it's cleaned up as before.

3. Next /sync-gbrain run detects gbrain's checkpoint via decideResume() in
   gstack-gbrain-sync.ts:
     - no checkpoint               → fresh ingest pass
     - checkpoint + staging ok     → set GSTACK_INGEST_RESUME_DIR; child
                                      reuses staging dir and skips
                                      writeStaged; gbrain import resumes
                                      from processedIndex+1
     - checkpoint + staging gone   → warn "previous checkpoint stale
                                      (staging dir gone), restaging from
                                      scratch" and proceed

Reuses gbrain's own checkpoint as the source of truth (D1 — no double-store
state). Detect-then-fallback semantics per C1.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-05-21 09:44:10 -07:00
parent a05546cddc
commit 700c9a4ff8
2 changed files with 214 additions and 18 deletions
+78 -14
View File
@@ -1272,13 +1272,37 @@ function cleanupStagingDir(dir: string): void {
* 1. forward the signal to the child (otherwise gbrain orphans, holds the
* PGLite write lock, and burns CPU — observed during 2026-05-10 cold-run
* testing)
* 2. synchronously clean up the staging dir BEFORE process.exit (otherwise
* finally blocks in async callers don't run after process.exit from
* inside a signal handler, leaking the staging dir on every interrupt)
* 2. PRESERVE the staging dir when gbrain has written an import-checkpoint
* pointing at it (the next /sync-gbrain run can resume from
* processedIndex+1). Otherwise synchronously clean up before
* process.exit, since `finally` blocks in ingestPass never run after
* process.exit fires from inside a signal handler.
*
* Resume semantics added for #1611: prior behavior unconditionally cleaned
* up the staging dir on SIGTERM, so the gbrain checkpoint always pointed at
* a missing dir and the next run had to restage from scratch.
*/
let _activeImportChild: ChildProcess | null = null;
let _activeStagingDir: string | null = null;
let _signalHandlersInstalled = false;
/**
* Returns true if gbrain has written ~/.gbrain/import-checkpoint.json with
* `dir` matching the current active staging dir. Indicates the next run
* can resume against this staging dir.
*/
function stagingDirIsCheckpointed(stagingDir: string): boolean {
try {
const cpPath = join(homedir(), ".gbrain", "import-checkpoint.json");
if (!existsSync(cpPath)) return false;
const raw = readFileSync(cpPath, "utf-8");
const cp = JSON.parse(raw) as { dir?: string };
return cp.dir === stagingDir;
} catch {
return false;
}
}
function installSignalForwarder(): void {
if (_signalHandlersInstalled) return;
_signalHandlersInstalled = true;
@@ -1290,11 +1314,24 @@ function installSignalForwarder(): void {
// child may have already exited between the alive-check and the kill
}
}
// Synchronously clean up the active staging dir before exiting. The async
// `finally` blocks in ingestPass never run after process.exit fires from
// inside this handler, so cleanup has to happen here.
if (_activeStagingDir) {
cleanupStagingDir(_activeStagingDir);
if (stagingDirIsCheckpointed(_activeStagingDir)) {
// Preserve for next-run resume. The orchestrator's decideResume()
// (in gstack-gbrain-sync.ts) will see the checkpoint + dir and
// re-invoke gbrain import against this same staging dir, picking
// up from processedIndex+1. See #1611.
try {
process.stderr.write(
`[memory-ingest] ${signal} received — preserving staging dir for resume: ${_activeStagingDir}\n`,
);
} catch {
// best-effort: stderr may be closed already
}
} else {
// No checkpoint pointing here — the import never reached gbrain or
// crashed before writing one. Clean up so we don't leak the dir.
cleanupStagingDir(_activeStagingDir);
}
_activeStagingDir = null;
}
// Re-raise to default action so the parent actually exits. Without this,
@@ -1444,19 +1481,46 @@ async function ingestPass(args: CliArgs): Promise<BulkResult> {
// entirely. gstack-brain-sync push will pick the dir up via its allowlist
// and the brain admin's pull job will index transcripts into the remote
// brain. Local PGLite (if any) stays code-only.
//
// Resume branch for #1611: when the orchestrator sets
// GSTACK_INGEST_RESUME_DIR (because gbrain's import-checkpoint.json points
// at an existing dir from a prior SIGTERM'd run), reuse that staging dir
// and skip the prepare/writeStaged phase entirely. gbrain's checkpoint
// tells it where to resume.
const remoteHttpMode = isRemoteHttpMcpMode();
const stagingDir = remoteHttpMode
? makePersistentTranscriptDir()
: makeStagingDir();
const resumeDir = process.env.GSTACK_INGEST_RESUME_DIR;
const resuming = !remoteHttpMode
&& typeof resumeDir === "string"
&& resumeDir.length > 0
&& existsSync(resumeDir);
const stagingDir = resuming
? resumeDir!
: remoteHttpMode
? makePersistentTranscriptDir()
: makeStagingDir();
// Register staging dir with the signal forwarder so SIGTERM/SIGINT can
// synchronously clean it up before process.exit (the async finally block
// below does NOT run after a signal-handler exit). In remote-http mode we
// skip registration — the dir is meant to persist.
// either preserve (when gbrain checkpointed it) or synchronously clean up.
// The async finally block below does NOT run after a signal-handler exit.
// In remote-http mode we skip registration — the dir is meant to persist.
if (!remoteHttpMode) {
_activeStagingDir = stagingDir;
}
try {
const staging = writeStaged(prep.prepared, stagingDir);
let staging: StagingResult;
if (resuming) {
// Pages are already on disk from the previous run. Skip writeStaged.
// The "written" count for the verdict reflects what's on disk now;
// gbrain's import will skip already-completed entries via its own
// checkpoint (processedIndex+1).
if (!args.quiet) {
console.error(
`[memory-ingest] resuming previous staging dir ${stagingDir} (skipping prepare phase)`,
);
}
staging = { staging_dir: stagingDir, written: prep.prepared.length, errors: [], stagedPathToSource: new Map() };
} else {
staging = writeStaged(prep.prepared, stagingDir);
}
failed += staging.errors.length;
if (!args.quiet && staging.errors.length > 0) {
for (const e of staging.errors.slice(0, 5)) {