diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a1f1ba82..29c12a8d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,25 @@ All notable changes to GBrain will be documented in this file. +## [0.42.38.0] - 2026-06-09 + +**Three independent job-layer bugs that left autopilot wedged or swallowed a command's output are fixed, each traced to source.** A triage of the job/lock/teardown layer (gbrain#1972) pulled them into one wave. + +A crashed sync (OOM, a recycle, a kill) used to strand its lock row: the source looked "syncing" forever because reclaim only happened when something else came along and contended for the same lock. There was no background sweep, so a low-traffic source could sit falsely locked for a long time. Now every cycle reaps locks whose holder process is provably dead on this host — scoped to the sync/cycle lock namespaces, never to elections or the worker supervisor, and guarded against PID reuse so a recycled PID can never clear a live lock. `gbrain doctor --fix` runs the same reaper for brains that don't run autopilot. + +Short one-shot CLI calls also got their full latency and output back. Database teardown could block for the full force-exit deadline against a transaction-mode pooler and then exit hard mid-write, which truncated the command's real output — the reason a relational query could come back empty even though the query itself worked. Teardown is now bounded by gbrain's own deadline instead of the connection driver's, so a short command returns in milliseconds with its output intact. + +And the cooperative-abort work started in v0.42.29 (which only covered the embed phase) now covers every long phase a cycle runs — extract, fact extraction, and consolidation all check for cancellation between batches, so a cancelled cycle relinquishes its worker promptly instead of being force-evicted. A cancelled cycle also no longer records itself as a completed full run. + +### Fixed +- **Stale dead-holder locks are reaped automatically (gbrain#1972, adjacent to #1470).** A background, host-scoped sweep at cycle start deletes `gbrain-sync:*` / `gbrain-cycle*` locks whose holder PID is dead, with a snapshot-matched delete that's safe against PID reuse and a 60s grace window. Other lock namespaces (elections, supervisor, reindex) keep their existing TTL behavior, untouched. `gbrain doctor --fix` reaps too, for no-autopilot brains. +- **One-shot CLI calls no longer hang on teardown or lose their output (gbrain#1959).** Pool disconnect is bounded by a gbrain-owned deadline (both pools closed concurrently) instead of blocking until the hard force-exit fired and truncated stdout. A short command returns promptly with intact output. +- **Cooperative abort now covers every long cycle phase (gbrain#1737 follow-up).** `extract` (incremental + full-walk), `extract_facts` (including its per-page embed and the phantom-redirect lock-retry), and `consolidate` check the abort signal between batches; `lint` yields periodically so it can be cancelled too. A cycle aborted mid-phase no longer stamps `last_full_cycle_at` as a completed run, and a new per-phase duration warning names any phase that overruns the worker's force-evict deadline. + +### To take advantage of v0.42.38.0 + +`gbrain upgrade`. No configuration needed — the lock reaper, bounded teardown, and abort coverage are all on by default. If a source has looked stuck "syncing" with no live process, the next cycle (or `gbrain doctor --fix`) clears it automatically. + ## [0.42.37.0] - 2026-06-08 **Cross-source reads now honor the caller's grant everywhere, a single bad frontmatter value no longer wedges a whole `lint`/`sync` run, and a handful of long-standing papercuts are gone.** A triage of the open issue backlog pulled the highest-impact bugs into one wave. diff --git a/TODOS.md b/TODOS.md index 69ebba340..1bd0164d1 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1,5 +1,24 @@ # TODOS +## gbrain#1972 job-layer follow-up (v0.43+) + +Filed from the #1972 fix (stale-lock reaper + bounded disconnect + complete +cooperative-abort). One item was deliberately gated, not deferred blindly. See plan + +GSTACK REVIEW REPORT at `~/.claude/plans/system-instruction-you-are-working-curious-pike.md`. + +- [ ] **P2 — `findBacklinkGaps` sync→async refactor (gated on telemetry).** The backlinks + phase does its heavy work in a single synchronous call (`findBacklinkGaps`, + `src/commands/backlinks.ts:71` — nested `readdirSync` double-walk, no `await` seam), so it + cannot be cooperatively aborted: a >30s run on a huge brain blocks the event loop and gets + force-evicted. lint was made yield-able this wave (it was already async); backlinks needs + `findBacklinkGaps` converted to async-with-periodic-yields, threaded through + `runBacklinksCore` + `runPhaseBacklinks`. **Why gated:** the trigger is UNCONFIRMED — we + don't know backlinks ever exceeds 30s. This wave added the phase-duration force-evict + attribution log (`FORCE_EVICT_DEADLINE_MS` in `src/core/cycle.ts`), which names any phase + that crosses the deadline. Do this refactor only if a production 24h pull shows backlinks + crossing it; otherwise it's a hot-loop rewrite for a non-occurring case. **Where:** + `src/commands/backlinks.ts`, `src/core/cycle.ts` (runPhaseBacklinks signal threading). + ## gbrain#1881 sync reclone ownership follow-ups (v0.43+) Filed from the #1881 fix (`gbrain sync --strategy code` deleted a user's working diff --git a/VERSION b/VERSION index 6f19cb7c6..79e10702e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.37.0 +0.42.38.0 diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index ef7ccc392..dd25c0c2e 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -32,7 +32,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/sync.ts:resolveSlugByPathOrSourcePath` — Resolves a slug by `pages.source_path` first (returns the stored slug for frontmatter-fallback pages whose path doesn't derive a slug), then falls back to `resolveSlugForPath(path)`. Threaded into all 4 delete/rename call sites (`performSync`'s un-syncable cleanup at ~:531, deletes at ~:603, rename oldSlug at ~:622). Without this, emoji-only / Thai / Arabic filenames whose slug came from frontmatter would orphan on delete/rename (the delete path would compute the wrong path-derived slug). Best-effort query — pre-migration brains fall through to the legacy path. - `src/core/sources-ops.ts` — Multi-source registration + clone-lifecycle ops (`addSource`, `recloneIfMissing`, `defaultCloneDir`, `isOwnedClone`, `unownedHint`). **Reclone-ownership invariant (must-never-violate): gbrain may only delete/re-clone a clone it created, NEVER a user working tree.** `recloneIfMissing` deletes `local_path`, so it gates on `isOwnedClone(src)` and throws a `SourceOpError('unmanaged_path', ...)` BEFORE any filesystem op when ownership is unprovable — fail-closed. Ownership is proven by `config.managed_clone === true` (written by `addSource`'s `--url` path, covering default-location and `--clone-dir` clones) OR `local_path === defaultCloneDir(id)` (back-compat for pre-marker clones, via exact normalized-path equality, symlink-free). A row with `remote_url` + an unowned `local_path` (a user-registered working tree, e.g. `sources add --path`) is refused untouched; re-add with `--url` to regain auto-reclone. The reclone is EXDEV-safe: clone into a SIBLING temp of `local_path` (not the shared `clones/.tmp`, which may sit on a different mount than a `--clone-dir` target), then swap (move old aside → move new in → drop old) so `local_path` is never left missing-and-unrecoverable; on swap failure the original is restored, and if restore fails the error names the `aside` path so it's never reflexively deleted. A TOCTOU re-check re-confirms ownership immediately before the destructive move and rejects a symlink leaf swapped in after the entry check (`symlink_escape`). `unownedHint(src, state)` is the shared recovery message used by both the core error and the `gbrain sync --source` CLI error; `gbrain sources restore` special-cases `unmanaged_path` to print "DB row restored; gbrain syncs this path read-only" instead of the misleading "try sync to recover" guidance. `SourceOpErrorCode` includes `unmanaged_path`. Pinned by `test/sources-ops.test.ts`, `test/sources-resync-recovery.test.ts`. - `src/core/utils.ts` — Shared SQL utilities extracted from postgres-engine.ts. Exports `parseEmbedding(value)` (throws on unknown input, used by migration + ingest paths where data integrity matters) and `tryParseEmbedding(value)` (returns `null` + warns once per process, used by search/rescore paths where availability matters more than strictness). `isUndefinedColumnError(err)` predicate — pattern-matches Postgres SQLSTATE 42703 / "column ... does not exist" with engine-driver shape variation tolerated; replaces bare `catch {}` blocks in `oauth-provider.ts` so genuine errors (lock timeout, network blip, permission denied) propagate while column-missing falls through to the legacy fallback. `validateSourceId(id)` throws on anything outside `^[a-z0-9_-]+$`, used by the per-source disk-layout fix in patterns.ts/synthesize.ts before any `join(brainDir, '.sources', source_id, slug+'.md')` so source_id can't traverse out of brainDir. `rowToPage` populates the required `Page.source_id` from the SELECT projection (`scripts/check-source-id-projection.sh` enforces every projection feeding `rowToPage` includes the column). -- `src/core/db.ts` — Connection management, schema initialization. `resolveSessionTimeouts()` returns `statement_timeout` + `idle_in_transaction_session_timeout` (defaults 5min each, env-overridable via `GBRAIN_STATEMENT_TIMEOUT`/`GBRAIN_IDLE_TX_TIMEOUT`/`GBRAIN_CLIENT_CHECK_INTERVAL`). Both `connect()` (module singleton) and `PostgresEngine.connect()` (worker pool) consume the result via postgres.js's `connection` option, sending GUCs as startup parameters that survive PgBouncer transaction mode (`setSessionDefaults` kept as a back-compat no-op shim). `connect()` returns `Promise` — `true` iff THIS call created the module singleton, `false` if it joined an existing one; the decision is atomic (no `await` between the `if (sql)` null-check and the synchronous `sql = postgres(...)` assignment), so two concurrent module connects can't both claim creation. `PostgresEngine` stores the return as its `_ownsModuleSingleton` token and only the creating engine may `db.disconnect()` the singleton — a borrower probe engine (lint/doctor config-lift) no-ops its disconnect, so its teardown can't null the connection the long-lived cycle owner is still using (the dream-cycle "connect() has not been called" failure). The module `sql` is only ever nulled by `db.disconnect()` (postgres.js auto-reconnects its own internal pool and never touches our reference). `disconnect()` snapshots + nulls `sql` before awaiting `s.end()` so a concurrent connect can't join a pool that's already closing. +- `src/core/db.ts` — Connection management, schema initialization. `resolveSessionTimeouts()` returns `statement_timeout` + `idle_in_transaction_session_timeout` (defaults 5min each, env-overridable via `GBRAIN_STATEMENT_TIMEOUT`/`GBRAIN_IDLE_TX_TIMEOUT`/`GBRAIN_CLIENT_CHECK_INTERVAL`). Both `connect()` (module singleton) and `PostgresEngine.connect()` (worker pool) consume the result via postgres.js's `connection` option, sending GUCs as startup parameters that survive PgBouncer transaction mode (`setSessionDefaults` kept as a back-compat no-op shim). `connect()` returns `Promise` — `true` iff THIS call created the module singleton, `false` if it joined an existing one; the decision is atomic (no `await` between the `if (sql)` null-check and the synchronous `sql = postgres(...)` assignment), so two concurrent module connects can't both claim creation. `PostgresEngine` stores the return as its `_ownsModuleSingleton` token and only the creating engine may `db.disconnect()` the singleton — a borrower probe engine (lint/doctor config-lift) no-ops its disconnect, so its teardown can't null the connection the long-lived cycle owner is still using (the dream-cycle "connect() has not been called" failure). The module `sql` is only ever nulled by `db.disconnect()` (postgres.js auto-reconnects its own internal pool and never touches our reference). `disconnect()` snapshots + nulls `sql` before awaiting the pool end so a concurrent connect can't join a pool that's already closing. The end routes through `endPoolBounded(pool)` (#1972) — a gbrain-owned `Promise.race` of `pool.end({ timeout: POOL_END_TIMEOUT_SECONDS })` against a hard timer — so a PgBouncer transaction-mode drain that never settles can't block teardown until the CLI's 10s force-exit fires and truncates stdout (#1959); it never throws. `connection-manager.ts` ends its direct + read pools concurrently through the same helper so the per-pool bounds don't stack. - `src/commands/migrate-engine.ts` — Bidirectional engine migration (`gbrain migrate --to supabase/pglite`). - `src/core/import-file.ts` — importFromFile + importFromContent (chunk + embed + tags). `importFromContent` and `importCodeFile` stamp `pages.embedding_signature` via `setPageEmbeddingSignature(slug, {sourceId, signature: currentEmbeddingSignature()})` when the import actually embedded (not `--no-embed`) so a model/dims swap is detectable as stale; `importCodeFile` only stamps when every chunk was freshly embedded this call (`needsEmbedIndexes.length === chunks.length`), mixed reuse-by-hash pages stay unstamped (`reindex --code --force` / `embed --stale` handle those). `importFromContent`'s tag reconciliation is ADD-ONLY: it only `addTag` (idempotent, ON CONFLICT DO NOTHING). The `tags` table has no provenance column and frontmatter tags are stripped from stored `pages.frontmatter` (markdown.ts:118), so a frontmatter-origin tag can't be distinguished from a DB-enrichment tag (auto-tag / dream synthesize / signal-detector) at re-import — deletion is unsafe (would wipe enrichment under `gbrain reindex --markdown`). Accepted trade-off: removing a tag from frontmatter no longer removes it from the DB on next sync (needs a `tag_source` provenance column). Pinned by `test/reindex-preserve-tags.test.ts` + `test/import-file.test.ts`. - `src/core/sync.ts` — Pure sync functions (manifest parsing, filtering, slug conversion). Exported `pruneDir(name: string): boolean` is the single source of truth for descent-time directory exclusion across walkers — blocks `node_modules` (no leading dot, so naive walkers slipped through and inflated MISSING_OPEN counts via vendor packages), dot-prefix dirs, `ops/`, and `*.raw` sidecars; `isSyncable` applies it per path segment, and `walkMarkdownFiles` in `src/commands/extract.ts` + `listTextFiles` in `src/core/cycle/transcript-discovery.ts` consult it BEFORE recursing to save the IO of walking thousands of vendor files (closes #923 + #202). `manageGitignore` worktree discriminator matches the gitdir path segment (`/modules/` = submodule, `/worktrees/` = worktree, per Git's documented layout) so Conductor worktrees (first-class repos) get `.gitignore` management for storage-tiering (closes #889). The sync-failure ledger (failure store, error classifier, the shared bookmark gate, and the doctor severity rule) lives in `src/core/sync-failure-ledger.ts`; `sync.ts` re-exports `classifyErrorCode`, `summarizeFailuresByCode`, `loadSyncFailures`, `unacknowledgedSyncFailures`, `acknowledgeSyncFailures`, `recordSyncFailures`, `decideSyncFailureSeverity`, `applySyncFailureGate`, and the `SyncFailure` type for backward-compatible imports — see its entry below. @@ -280,7 +280,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/progress.ts` — Shared bulk-action progress reporter. Writes to stderr. Modes: `auto` (TTY `\r`-rewriting; non-TTY plain lines), `human`, `json` (JSONL), `quiet`. Rate-gated by `minIntervalMs` and `minItems`. `startHeartbeat(reporter, note)` for single long queries. `child()` composes phase paths. Singleton SIGINT/SIGTERM coordinator emits `abort` events for every live phase. EPIPE defense on both sync throws and stream `'error'` events. Zero dependencies. `emitHumanLine` is prefix-aware — inside a `withSourcePrefix(id, ...)` scope from `src/core/console-prefix.ts` it prepends `[id] ` (and TTY-rewrite mode `\r\x1b[2K` carries the prefix inside the clear-to-EOL escape); `emitJson` is intentionally NOT prefixed so NDJSON consumers don't choke on a `[id] {...}` shape. - `src/core/console-prefix.ts` — `AsyncLocalStorage`-backed per-source line-prefix helper. Exports `withSourcePrefix(id, fn)` (runs `fn` with `id` as active prefix; nested wraps replace then restore), `getSourcePrefix()` (read-only accessor; test seam), and `slog(...)` / `serr(...)` (prefix-aware `console.log`/`console.error`). Embedded-newline-safe: a multi-line string under prefix `[foo]` emits `[foo] line1\n[foo] line2`. Outside a wrap, `slog`/`serr` fall through to bare `console.log`/`console.error` so single-source callers see identical output (back-compat invariant). Use `src.id` (slug-validated by `sources add`) NOT `src.name` (free-form) to defeat log-injection through newline/control-character names. Coverage: `src/commands/sync.ts` performSync + callees, `src/commands/embed.ts` runEmbedCore + helpers, `src/core/progress.ts` emitHumanLine. - `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers. -- `src/core/db-lock.ts` — generic `tryAcquireDbLock(engine, lockId, ttlMinutes)` over the `gbrain_cycle_locks` table. Parameterized lock id so scopes nest cleanly: `gbrain-cycle` for the broad cycle (held by `cycle.ts`) and `gbrain-sync` (`SYNC_LOCK_ID`) for `performSync`'s narrower writer window. UPSERT-with-TTL semantics survive PgBouncer transaction pooling (unlike session-scoped `pg_try_advisory_lock`); crashed holders auto-release once their TTL expires. It also does automatic same-host dead-pid takeover: when the upsert finds a held, NOT-TTL-expired lock whose holder is on this host and provably dead, it reclaims via a guarded `DELETE WHERE id=$1 AND holder_pid=$2` + one normal-upsert retry returning the standard handle (refresh/release intact). The liveness check is the exported `classifyHolderLiveness(pid, host, ageMs, opts?)` / `isHolderDeadLocally(...)` (injectable `process.kill` seam; `HOLDER_TAKEOVER_GRACE_MS = 60_000` PID-reuse guard; EPERM classified as `alive` so a live process you don't own is never stolen). TTL-expired locks stay the upsert's job; cross-host stays TTL-only. `runBreakLock` (`src/commands/sync.ts`) consumes the same predicate. Pinned by `test/db-lock-auto-takeover.test.ts`. +- `src/core/db-lock.ts` — generic `tryAcquireDbLock(engine, lockId, ttlMinutes)` over the `gbrain_cycle_locks` table. Parameterized lock id so scopes nest cleanly: `gbrain-cycle` for the broad cycle (held by `cycle.ts`) and `gbrain-sync` (`SYNC_LOCK_ID`) for `performSync`'s narrower writer window. UPSERT-with-TTL semantics survive PgBouncer transaction pooling (unlike session-scoped `pg_try_advisory_lock`); crashed holders auto-release once their TTL expires. It also does automatic same-host dead-pid takeover: when the upsert finds a held, NOT-TTL-expired lock whose holder is on this host and provably dead, it reclaims via a guarded `DELETE WHERE id=$1 AND holder_pid=$2` + one normal-upsert retry returning the standard handle (refresh/release intact). The liveness check is the exported `classifyHolderLiveness(pid, host, ageMs, opts?)` / `isHolderDeadLocally(...)` (injectable `process.kill` seam; `HOLDER_TAKEOVER_GRACE_MS = 60_000` PID-reuse guard; EPERM classified as `alive` so a live process you don't own is never stolen). TTL-expired locks stay the upsert's job; cross-host stays TTL-only. `runBreakLock` (`src/commands/sync.ts`) consumes the same predicate. Background reaper (#1972): `reapDeadHolderLocks(engine)` is the periodic sweep the contention path lacked — it deletes locks whose holder is `isHolderDeadLocally`, scoped to the `gbrain-sync:*` / `gbrain-cycle`/`gbrain-cycle:*` namespaces ONLY (election/supervisor/reindex locks keep TTL-only behavior, untouched), via `deleteLockRowExact(engine, id, pid, acquiredAt)` — a snapshot-matched delete (`date_trunc('milliseconds', acquired_at) = $3`, so the ms a JS Date keeps survives) that's TOCTOU-safe against a reused PID taking the lock between SELECT and DELETE. `cycle.ts` runs it at cycle start (before the sync phase); `gbrain doctor --fix` runs it for no-autopilot brains. `selectLockRows(engine, opts?)` + a shared row→`LockSnapshot` mapper are the single canonical reader now backing `inspectLock` + `listStaleLocks` + the reaper (was triplicated). Pinned by `test/db-lock-auto-takeover.test.ts` + `test/db-lock-reap.test.ts`. - `src/core/sync-concurrency.ts` — single source of truth for the parallel-sync policy. Exports `autoConcurrency(engine, fileCount, override?)` (PGLite always serial; explicit override clamped to >=1; auto path returns `DEFAULT_PARALLEL_WORKERS=4` when `fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD=100`), `shouldRunParallel(workers, fileCount, explicit)` (explicit `--workers` bypasses the >50-file floor), and `parseWorkers(s)` (rejects `'0'`, `'-3'`, `'foo'`, `'1.5'`, trailing chars). Used by `performSync`, `performFullSync`, `runImport`, and the Minion `sync` handler so the sites can't drift. `DEFAULT_PARALLEL_SOURCES = 4` is a SEPARATE constant for the per-source fan-out under `gbrain sync --all` — kept distinct from `DEFAULT_PARALLEL_WORKERS` because total live Postgres connections per wave ≈ `DEFAULT_PARALLEL_SOURCES × DEFAULT_PARALLEL_WORKERS × 2 (per-file pool)` = 32 at both defaults (each per-file worker opens its own `PostgresEngine` with `poolSize = min(2, resolvePoolSize(2))`); `sync.ts` warns when `parallel × workers × 2 > 16`. `resolveWorkersWithClamp(engine, override, commandName, fileCount)` wraps `autoConcurrency` with a per-command stderr clamp warning on PGLite (per-(command, requested) dedup via module-scoped warned-once set with `_resetWorkersClampWarningsForTest()` seam) and is the canonical surface for every bulk-command `--workers N` flag (extract-conversation-facts, extract, edges-backfill, reindex-multimodal, reindex, reindex-code); embed.ts deliberately bypasses it and keeps `GBRAIN_EMBED_CONCURRENCY || 20`. `resolveMaxConnections()` (reads `GBRAIN_MAX_CONNECTIONS`, undefined when unset) + `clampWorkersForConnectionBudget(workers, perWorkerPool, maxConnections, parentPool)` back the opt-in single-sync connection-footprint clamp so a big sync stays under a low pooler cap (`parent_pool + workers×perWorkerPool ≤ budget`); `gbrain doctor`'s `pool_budget` check (`computePoolBudgetCheck` / `checkPoolBudget` in `src/commands/doctor.ts`) warns when the budget leaves no room for a worker, pointing at `GBRAIN_POOL_SIZE=2`. Pinned by `test/pglite-workers-clamp.test.ts`. - `src/core/worker-pool.ts` — Canonical sliding-pool + bounded-semaphore primitive (extracted from `src/commands/embed.ts` sliding-pool sites and `src/commands/eval-cross-modal.ts` `runWithLimit` semaphore). Two exports: `runSlidingPool({items, workers, onItem, signal?, onError?, failureLabel?, onProgress?})` + `runWithLimit({items, limit, fn, signal?})`. Atomicity invariant: `const idx = nextIdx++` is one synchronous JS statement (no `await` between read and write — guaranteed by the single-threaded event loop), documented in the module header AND enforced by `scripts/check-worker-pool-atomicity.sh` (wired into `bun run verify`), which rejects importing `worker_threads` in any consuming file and inserting `await` between the `nextIdx` read and write. `MUST_ABORT_ERROR_TAGS` set is seeded with `BUDGET_EXHAUSTED` from `src/core/budget/budget-tracker.ts`; tagged errors (matched via `err.tag === 'BUDGET_EXHAUSTED'` to avoid cross-module import) bypass `onError` and hard-abort the pool via `AbortController.abort()` to in-flight `onItem` — the budget cap is a structural ceiling under concurrency. `failures[]` shape is `{idx, label, error}` records (NOT full items; callers supply `failureLabel(item) => string`) for bounded memory under huge brains. Pinned by `test/worker-pool.test.ts` + `test/scripts/check-worker-pool-atomicity.test.ts`. Drives every `--workers N` bulk command. - `src/commands/embed.ts` extension — both inline sliding-pool sites (`embedAll` simple at `:458-467` and `embedAllStale` paginated + AbortSignal at `:586-632`) call `runSlidingPool` from the shared worker-pool helper. Invariant-level contract preserved (counts + cost + AbortSignal propagation + per-batch rate-limit retry via `embedBatchWithBackoff`); byte-equality on progress-event ORDERING is NOT promised. The `GBRAIN_EMBED_CONCURRENCY || 20` default is preserved and embed bypasses `resolveWorkersWithClamp` because the 20-worker default would otherwise silently change every brain's embed hot path. Pinned by `test/embed-helper-migration.test.ts` (asserts the helper is wired in AND the pre-migration `let nextIdx = 0` + `Promise.all(Array.from({length: numWorkers}, ...))` shapes are gone). @@ -376,7 +376,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/jobs.ts` extension — registers 11 Minion handlers: `reindex`, `repair-jsonb`, `orphans`, `integrity`, `purge`, `synthesize` (PROTECTED), `patterns` (PROTECTED), `consolidate` (PROTECTED), `extract_facts`, `resolve_symbol_edges`, `recompute_emotional_weight`. Phase wrappers delegate to `runCycle({phases:[name]})` so `src/core/cycle.ts` stays the single source of truth for phase semantics. The standalone `sync` handler passes `noExtract: true` to match `runPhaseSync`'s contract (doctor's remediation plan emitting `[sync, extract]` would otherwise double-extract). - `src/core/minions/protected-names.ts` extension — `PROTECTED_JOB_NAMES` includes `synthesize`, `patterns`, `consolidate`. These phases internally submit `subagent` children with `allowProtectedSubmit=true` and can spend Anthropic credits. Only trusted local callers (CLI, autopilot, `doctor --remediate`) can submit them; MCP requests are rejected by `submit_job`'s protected-name guard. - `src/commands/autopilot.ts` extension — targeted-submit loop instead of blanket `autopilot-cycle` dispatch. Each tick: cheap `engine.getHealth()` (single SQL count) + `computeRecommendations()`, then route by shape — `score >= 95 AND no plan AND <60min since last full` → sleep; `score >= 95 AND >=60min` → submit `autopilot-cycle` (60-min floor exercises phase-coupling invariants on healthy brains); `plan <= 3 steps AND est <5min` → submit individual handlers; `plan large OR score < 70` → submit full `autopilot-cycle`. The `gbrain-cycle` lock ensures targeted submissions and the full cycle can't run concurrently. `maxWaiting: 1` per submit closes the queue-fan-out vector. -- `src/core/abort-check.ts` (#1737) — one canonical place for cooperative-abort checks across gbrain's long loops. `isAborted(signal?)` → boolean (for loops that `break` and return partial progress). `throwIfAborted(signal?, label?)` throws an `AbortError` (`name === 'AbortError'`) at phase boundaries, preferring the signal's `reason` ('wall-clock'/'lock-lost'/'shutdown') so the unwind self-describes. `anySignal(internal, external?)` composes two signals into one that fires when EITHER does (platform `AbortSignal.any` with a manual-relay fallback), returning the internal unchanged when there's no external so non-aborting callers pay nothing. The fix for the #1737 cycle-wedge: the embed phase ignored its abort signal and ran to completion, so `gbrain_cycle_locks` stayed held and later autopilot cycles skipped with `cycle_already_running`; threading these checks through `runPhaseEmbed → runEmbedCore → embedAll(Stale)/embedPage` lets the phase bail and release the lock immediately. Pinned by `test/abort-check.test.ts`. +- `src/core/abort-check.ts` (#1737) — one canonical place for cooperative-abort checks across gbrain's long loops. `isAborted(signal?)` → boolean (for loops that `break` and return partial progress). `throwIfAborted(signal?, label?)` throws an `AbortError` (`name === 'AbortError'`) at phase boundaries, preferring the signal's `reason` ('wall-clock'/'lock-lost'/'shutdown') so the unwind self-describes. `anySignal(internal, external?)` composes two signals into one that fires when EITHER does (platform `AbortSignal.any` with a manual-relay fallback), returning the internal unchanged when there's no external so non-aborting callers pay nothing. The fix for the #1737 cycle-wedge: the embed phase ignored its abort signal and ran to completion, so `gbrain_cycle_locks` stayed held and later autopilot cycles skipped with `cycle_already_running`; threading these checks through `runPhaseEmbed → runEmbedCore → embedAll(Stale)/embedPage` lets the phase bail and release the lock immediately. Coverage now spans every long cycle-reachable phase (#1972), not just embed: `extract` (incremental `extractForSlugs` + the full-walk `extractLinksFromDir`/`extractTimelineFromDir`, all via `runSlidingPool`'s signal), `extract_facts` (per-page loop + the per-page `embed` signal + `runPhantomRedirectPass`'s 30s lock-retry), `consolidate`'s bucket loop, and `lint` (which is synchronous, so it `await`s a periodic yield to let the signal land). `runCycle` adds a terminal abort check before stamping `last_full_cycle_at` so a cancelled cycle never reports a completed full run, plus a per-phase `duration_ms` warning that names any phase overrunning the worker's 30s force-evict deadline. Pinned by `test/abort-check.test.ts` + `test/cycle-abort.test.ts`. - `src/core/cycle.ts` extension — `purge` phase (the cycle's 9th phase, for soft-delete TTLs) also GCs stale `op_checkpoints` rows older than 7 days. Non-fatal on pre-v67 brains (DROP-target-table check before DELETE). #1737: the cycle threads its abort signal into the embed phase (`runPhaseEmbed(engine, dryRun, signal)`) so a timed-out cycle's long embed phase honors cancellation and releases `gbrain_cycle_locks` right away instead of after a full backlog run. - `src/commands/embed.ts` extension — wires `--background` as the reference integration for the `maybeBackground()` helper. `gbrain embed --stale --background` submits as a Minion job, prints `job_id=N` to stdout, exits 0. Composable: `JOB=$(gbrain embed --stale --background | grep -oE 'job_id=[0-9]+' | cut -d= -f2); gbrain jobs follow $JOB`. The other six commands (`extract`, `lint`, `backlinks`, `reindex`, `integrity`, `pages`) adopt the same pattern in a follow-up wave. #1737: `runEmbedCore` accepts an optional `signal` threaded down both the `--stale` and `--all` paths (`embedAllStale`/`embedAll`/`embedPage`); each composes it with the internal wall-clock budget via `anySignal` and checks `isAborted`/`effectiveSignal.aborted` in every per-slug loop, page-claim pool, and `embedBatch` call, so a worker abort (wall-clock timeout / lock loss / SIGTERM) stops embedding within a batch. Pinned by `test/embed.serial.test.ts`. - `src/core/cli-options.ts` extension — `maybeBackground(opName, fingerprintArgs, runDirect)` helper. Same semantics in TTY and cron (no `--no-tty-detect` flag, no surprise behavior change between contexts): when `--background` is passed, submits the op as a Minion job via `op_checkpoints` for resumability and returns the `job_id`. `--background --follow` execs `gbrain jobs follow ` so the user sees the same stderr stream they'd get from a direct call. PGLite degrades to inline execution with a clear stderr note ("PGLite worker pool not yet supported; running inline"). Returns a tagged union the caller dispatches on. diff --git a/package.json b/package.json index fbd6e4e2d..9d30c7d7d 100644 --- a/package.json +++ b/package.json @@ -143,5 +143,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.37.0" + "version": "0.42.38.0" } diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index b59e6fc8b..08651157e 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -2346,12 +2346,39 @@ export function checkAutopilotLockScope(): Check { * but the main work is blocked. Requires explicit heartbeat probe; * speculation until production data shows the case. */ -export async function checkStaleLocks(engine: BrainEngine): Promise { +export async function checkStaleLocks( + engine: BrainEngine, + opts: { fix?: boolean; dryRun?: boolean } = {}, +): Promise { try { - const { listStaleLocks } = await import('../core/db-lock.ts'); + const { listStaleLocks, reapDeadHolderLocks } = await import('../core/db-lock.ts'); + + // #1972: under `gbrain doctor --fix`, reap dead-holder sync/cycle locks + // using the SAME namespace-scoped, host-scoped, snapshot-matched reaper the + // cycle runs at start. This is the self-heal path for no-autopilot brains: a + // brain that never runs `gbrain dream` never hits the cycle-start sweep, so + // doctor --fix is how its crashed-sync locks get cleared. DB-only, so it's + // orthogonal to (and unaffected by) the skills-dir --fix safety gate above. + // Best-effort: a reap failure falls through to the warn path below. + let reapedIds: string[] = []; + if (opts.fix && !opts.dryRun) { + try { + reapedIds = (await reapDeadHolderLocks(engine)).reapedIds; + } catch { /* fall through; listStaleLocks still surfaces remaining locks */ } + } + const reapedNote = reapedIds.length > 0 + ? `Reaped ${reapedIds.length} dead-holder lock(s): ${reapedIds.join(', ')}.` + : null; + const stale = await listStaleLocks(engine); if (stale.length === 0) { - return { name: 'stale_locks', status: 'ok', message: 'No stale locks (no rows with ttl_expires_at < NOW())' }; + return { + name: 'stale_locks', + status: 'ok', + message: reapedNote + ? `${reapedNote} No stale locks remain.` + : 'No stale locks (no rows with ttl_expires_at < NOW())', + }; } const lines = stale.slice(0, 10).map(s => { const ageH = Math.floor(s.age_ms / 3600_000); @@ -2360,11 +2387,15 @@ export async function checkStaleLocks(engine: BrainEngine): Promise { return ` ${s.id} (pid ${s.holder_pid} on ${s.holder_host}, age ${ageH}h) → ${breakHint}`; }); const tail = stale.length > 10 ? ` ... and ${stale.length - 10} more.` : null; + const header = opts.fix + ? `${stale.length} stale lock(s) remain that could not be auto-reaped (live holder, cross-host, or within the PID-reuse grace):` + : `${stale.length} stale lock(s) detected (ttl_expires_at < NOW()):`; return { name: 'stale_locks', status: 'warn', message: [ - `${stale.length} stale lock(s) detected (ttl_expires_at < NOW()):`, + reapedNote, + header, ...lines, tail, ].filter(Boolean).join('\n'), @@ -6951,7 +6982,7 @@ export async function buildChecks( checks.push(checkAutopilotLockScope()); // v0.41.6.0 D3 — stale_locks (gbrain_cycle_locks rows with ttl_expires_at < NOW()) progress.heartbeat('stale_locks'); - checks.push(await checkStaleLocks(engine)); + checks.push(await checkStaleLocks(engine, { fix: doFix, dryRun })); // v0.38 — cycle_phase_scope (informational; no DB cost) progress.heartbeat('cycle_phase_scope'); checks.push(checkCyclePhaseScope()); diff --git a/src/commands/extract.ts b/src/commands/extract.ts index a4a8bb788..7ef9e337c 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -60,6 +60,7 @@ import { createHash } from 'crypto'; // v0.41.15.0 (T7, D9): --workers N for the fs-walk inner loops via the // shared sliding-pool helper + PGLite-clamp wrapper. import { runSlidingPool } from '../core/worker-pool.ts'; +import { isAborted } from '../core/abort-check.ts'; import { parseWorkers, resolveWorkersWithClamp } from '../core/sync-concurrency.ts'; // Batch size for addLinksBatch / addTimelineEntriesBatch. @@ -526,6 +527,14 @@ export interface ExtractOpts { * own pagination and stay serial in v0.41.15.0. */ workers?: number; + /** + * #1972: cooperative-abort signal. Forwarded into the sliding pool (which + * propagates it to every worker) and checked at the top of each onItem, so a + * cancelled cycle's extract (incremental OR full-walk) relinquishes its + * worker slot well under the 30s force-evict. Honored by the cycle-reachable + * paths: extractForSlugs, extractLinksFromDir, extractTimelineFromDir. + */ + signal?: AbortSignal; } /** @@ -564,7 +573,7 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr // Nothing changed — skip entirely. return result; } - const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode, workers); + const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode, workers, opts.signal); result.links_created = r.links_created; result.timeline_entries_created = r.timeline_created; result.pages_processed = r.pages; @@ -573,12 +582,12 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr // Full walk path: CLI `gbrain extract` or first-run. if (opts.mode === 'links' || opts.mode === 'all') { - const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode, workers); + const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal); result.links_created = r.created; result.pages_processed = r.pages; } if (opts.mode === 'timeline' || opts.mode === 'all') { - const r = await extractTimelineFromDir(engine, opts.dir, dryRun, jsonMode, workers); + const r = await extractTimelineFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal); result.timeline_entries_created = r.created; result.pages_processed = Math.max(result.pages_processed, r.pages); } @@ -931,6 +940,7 @@ async function extractForSlugs( // shared flush primitive; JS single-threaded event loop makes the // shared counter increments atomic. workers: number = 1, + signal?: AbortSignal, ): Promise<{ links_created: number; timeline_created: number; pages: number }> { // Build the full slug set for link resolution (fast: just readdir, no file reads) const allFiles = walkMarkdownFiles(brainDir); @@ -992,8 +1002,12 @@ async function extractForSlugs( await runSlidingPool({ items: slugs, workers, + signal, failureLabel: (slug) => slug, onItem: async (slug) => { + // #1972: bail before doing any work for this slug on abort. Trailing + // flushLinks/flushTimeline still commit accumulated rows — no torn write. + if (isAborted(signal)) return; const relPath = slug + '.md'; const fullPath = join(brainDir, relPath); try { @@ -1048,6 +1062,7 @@ async function extractLinksFromDir( engine: BrainEngine, brainDir: string, dryRun: boolean, jsonMode: boolean, // v0.41.15.0 (T7): in-process worker count. Default 1. workers: number = 1, + signal?: AbortSignal, ): Promise<{ created: number; pages: number }> { const files = walkMarkdownFiles(brainDir); const allSlugs = new Set(files.map(f => pathToSlug(f.relPath))); @@ -1088,8 +1103,11 @@ async function extractLinksFromDir( await runSlidingPool({ items: files, workers, + signal, failureLabel: (f) => f.relPath, onItem: async (file) => { + // #1972: bail before this file on abort; trailing flush() commits the batch. + if (isAborted(signal)) return; try { const content = readFileSync(file.path, 'utf-8'); const links = await extractLinksFromFile(content, file.relPath, allSlugs, { globalBasename }); @@ -1123,6 +1141,7 @@ async function extractTimelineFromDir( engine: BrainEngine, brainDir: string, dryRun: boolean, jsonMode: boolean, // v0.41.15.0 (T7): in-process worker count. Default 1. workers: number = 1, + signal?: AbortSignal, ): Promise<{ created: number; pages: number }> { const files = walkMarkdownFiles(brainDir); @@ -1153,8 +1172,11 @@ async function extractTimelineFromDir( await runSlidingPool({ items: files, workers, + signal, failureLabel: (f) => f.relPath, onItem: async (file) => { + // #1972: bail before this file on abort; trailing flush() commits the batch. + if (isAborted(signal)) return; try { const content = readFileSync(file.path, 'utf-8'); const slug = pathToSlug(file.relPath); diff --git a/src/commands/lint.ts b/src/commands/lint.ts index c9bfdba8d..e14e05456 100644 --- a/src/commands/lint.ts +++ b/src/commands/lint.ts @@ -18,6 +18,7 @@ import { readFileSync, writeFileSync, readdirSync, statSync, lstatSync, existsSync } from 'fs'; import { join, relative } from 'path'; +import { isAborted } from '../core/abort-check.ts'; import { parseMarkdown, type ParseValidationCode } from '../core/markdown.ts'; import { assessContentSanity, @@ -406,6 +407,13 @@ export interface LintOpts { * create + disconnect a competing module-style engine that nulls the * shared db singleton mid-cycle. */ engine?: BrainEngine; + /** + * #1972: cooperative-abort signal. lint's per-page work is synchronous, so + * without a periodic yield the event loop can't deliver an abort and a + * very large lint would block past the worker's 30s force-evict. The loop + * yields + checks this every 200 pages. + */ + signal?: AbortSignal; } export interface LintResult { @@ -444,7 +452,16 @@ export async function runLintCore(opts: LintOpts): Promise { let totalFixed = 0; let pagesWithIssues = 0; - for (const page of pages) { + for (let idx = 0; idx < pages.length; idx++) { + const page = pages[idx]; + // #1972: every 200 pages, yield to the event loop and honor abort. The + // yield is what lets the abort signal actually fire (the rest of the loop + // is synchronous); the break returns a valid partial LintResult since each + // page is independently read + written. + if (idx > 0 && idx % 200 === 0) { + if (isAborted(opts.signal)) break; + await new Promise((resolve) => setImmediate(resolve)); + } const content = readFileSync(page, 'utf-8'); const issues = lintContent(content, isSingleFile ? page : relative(opts.target, page), lintOpts); if (issues.length === 0) continue; diff --git a/src/core/connection-manager.ts b/src/core/connection-manager.ts index 73cd913b8..7073d101f 100644 --- a/src/core/connection-manager.ts +++ b/src/core/connection-manager.ts @@ -37,7 +37,7 @@ */ import postgres from 'postgres'; -import { resolvePrepare, resolveSessionTimeouts, resolvePoolSize } from './db.ts'; +import { resolvePrepare, resolveSessionTimeouts, resolvePoolSize, endPoolBounded } from './db.ts'; import { redactPgUrl } from './url-redact.ts'; import { logConnectionEvent } from './connection-audit.ts'; @@ -400,15 +400,21 @@ export class ConnectionManager { * (db.ts singleton path). Direct pool is always ours. */ async disconnect(): Promise { + // #1972: end both pools concurrently with a gbrain-owned hard bound, so the + // per-pool drains don't STACK (sequential 2s waits → ~4-6s once the + // instance/module pool is added downstream) and neither can hang teardown + // past the CLI's 10s force-exit. endPoolBounded never throws. + const ends: Promise[] = []; if (this._directPool) { - try { await this._directPool.end(); } catch { /* idempotent */ } + ends.push(endPoolBounded(this._directPool)); this._directPool = null; this._directInit = null; } if (this._readPool && !this._readPoolOwnedExternally) { - try { await this._readPool.end(); } catch { /* idempotent */ } + ends.push(endPoolBounded(this._readPool)); this._readPool = null; } + await Promise.all(ends); } /** diff --git a/src/core/cycle.ts b/src/core/cycle.ts index d254ffbed..a4dce8ffe 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -49,7 +49,7 @@ import { gbrainPath } from './config.ts'; import type { BrainEngine } from './engine.ts'; import { createProgress, type ProgressReporter } from './progress.ts'; import { getCliOptions, cliOptsToProgressOptions } from './cli-options.ts'; -import { tryAcquireDbLock, type DbLockHandle } from './db-lock.ts'; +import { tryAcquireDbLock, reapDeadHolderLocks, type DbLockHandle } from './db-lock.ts'; import { assertValidSourceId } from './source-id.ts'; // ─── Types ───────────────────────────────────────────────────────── @@ -328,8 +328,13 @@ export interface CycleReport { * - 'failed' : lock acquired but all attempted phases failed */ status: CycleStatus; - /** Present when status = 'skipped'. E.g., 'cycle_already_running' or 'no_database'. */ + /** Present when status = 'skipped'. E.g., 'cycle_already_running' or 'no_database'. Also 'aborted' when the cycle was cancelled mid-flight (#1972). */ reason?: string; + /** + * #1972: dead-holder sync/cycle locks the cycle-start reaper cleared this + * run (count + lock ids). Omitted when nothing was reaped or no engine. + */ + reaped_dead_holder_locks?: { reaped: number; reapedIds: string[] }; brain_dir: string | null; phases: PhaseResult[]; totals: { @@ -714,7 +719,7 @@ function checkAborted(signal?: AbortSignal): void { // keyword is the minimal seam that lets behavioral tests drive the // wrapper's result-mapping (counter → status enum + summary) without // going through runCycle's full setup cost. -export async function runPhaseLint(brainDir: string, dryRun: boolean, engine?: BrainEngine | null): Promise { +export async function runPhaseLint(brainDir: string, dryRun: boolean, engine?: BrainEngine | null, signal?: AbortSignal): Promise { try { const { runLintCore } = await import('../commands/lint.ts'); // issue #1678: pass the cycle's live engine so lint's content-sanity @@ -722,7 +727,7 @@ export async function runPhaseLint(brainDir: string, dryRun: boolean, engine?: B // competing module-style engine that nulls the shared db singleton // mid-cycle (which broke every phase after lint with a misleading // "connect() has not been called"). - const result = await runLintCore({ target: brainDir, fix: true, dryRun, engine: engine ?? undefined }); + const result = await runLintCore({ target: brainDir, fix: true, dryRun, engine: engine ?? undefined, signal }); const issues = result.total_issues ?? 0; const fixed = result.total_fixed ?? 0; const remaining = Math.max(0, issues - fixed); @@ -931,6 +936,7 @@ async function runPhaseExtract( brainDir: string, dryRun: boolean, changedSlugs?: string[], + signal?: AbortSignal, ): Promise { try { const { runExtractCore } = await import('../commands/extract.ts'); @@ -952,6 +958,7 @@ async function runPhaseExtract( mode: 'all', dir: brainDir, slugs: changedSlugs, // undefined = full walk (first run / manual) + signal, }); const linksCreated = result?.links_created ?? 0; const timelineCreated = result?.timeline_entries_created ?? 0; @@ -988,6 +995,7 @@ async function runPhaseExtractFacts( sourceId: string, dryRun: boolean, changedSlugs?: string[], + signal?: AbortSignal, ): Promise { try { const { runExtractFacts } = await import('./cycle/extract-facts.ts'); @@ -996,6 +1004,7 @@ async function runPhaseExtractFacts( dryRun, sourceId, brainDir: brainDir ?? undefined, + signal, }); // Empty-fence guard: pre-v51 legacy rows pending the v0_32_2 backfill. @@ -1511,6 +1520,24 @@ export async function runCycle( } } + // #1972: reap dead-holder sync/cycle locks at cycle start — before the sync + // phase needs them — so a crashed sync's stranded lock self-heals THIS tick + // instead of waiting out its TTL. Best-effort, namespace-scoped + host-scoped; + // never touches this cycle's own (live) lock. Skipped on dry-run (no writes). + let reapedLocks: { reaped: number; reapedIds: string[] } | undefined; + if (engine && !dryRun) { + try { + const r = await reapDeadHolderLocks(engine); + if (r.reaped > 0) { + reapedLocks = r; + console.warn(`[cycle] reaped ${r.reaped} dead-holder lock(s): ${r.reapedIds.join(', ')}`); + } + } catch (e) { + // Non-fatal: reaping is a backstop, never blocks the cycle. + console.warn(`[cycle] dead-holder lock reap failed (non-fatal): ${e instanceof Error ? e.message : String(e)}`); + } + } + try { // ── Phase 1: lint ──────────────────────────────────────────── if (phases.includes('lint')) { @@ -1519,7 +1546,7 @@ export async function runCycle( phaseResults.push(skipNoBrainDir('lint')); } else { progress.start('cycle.lint'); - const { result, duration_ms } = await timePhase(() => runPhaseLint(brainDir, dryRun, engine)); + const { result, duration_ms } = await timePhase(() => runPhaseLint(brainDir, dryRun, engine, opts.signal)); result.duration_ms = duration_ms; phaseResults.push(result); progress.finish(); @@ -1627,7 +1654,7 @@ export async function runCycle( // If sync didn't run (phases exclude it) or failed, syncPagesAffected // is undefined → extract falls back to full walk (safe default). progress.start('cycle.extract'); - const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected)); + const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal)); result.duration_ms = duration_ms; phaseResults.push(result); progress.finish(); @@ -1662,7 +1689,7 @@ export async function runCycle( // source installs). const xfSourceId = cycleSourceId ?? 'default'; const { result, duration_ms } = await timePhase(() => - runPhaseExtractFacts(engine, brainDir, xfSourceId, dryRun, syncPagesAffected)); + runPhaseExtractFacts(engine, brainDir, xfSourceId, dryRun, syncPagesAffected, opts.signal)); result.duration_ms = duration_ms; phaseResults.push(result); progress.finish(); @@ -1897,6 +1924,7 @@ export async function runCycle( const { result, duration_ms } = await timePhase(() => runPhaseConsolidate(engine, { dryRun, yieldDuringPhase: opts.yieldDuringPhase, + signal: opts.signal, })); result.duration_ms = duration_ms; phaseResults.push(result); @@ -2189,6 +2217,34 @@ export async function runCycle( const totals = extractTotals(phaseResults); const status = deriveStatus(phaseResults, totals); + // #1972 (Codex #9): a phase that breaks on abort returns status 'ok' with + // partial counts. If the abort fired during the LAST selected phase, no + // between-phase checkAborted ran afterward, so without this guard runCycle + // would compute an ok/partial status AND stamp last_full_cycle_at — marking + // a cancelled run as a completed full cycle, which makes the next tick skip + // work it never actually did. Treat an aborted signal as a non-success run: + // skip the freshness stamp and report status 'partial' with reason 'aborted'. + const aborted = opts.signal?.aborted === true; + + // #1972 (Decision 7A gating): attribute force-evicts. The minion worker + // force-evicts a job 30s after abort and logs "handler ignored abort signal"; + // that log doesn't say WHICH phase blocked. Any phase whose wall-clock exceeds + // that deadline is a force-evict suspect (e.g. a synchronous lint/backlinks on + // a huge brain). Name it here so the next production pull tells us whether the + // remaining gap is backlinks (still uninstrumented per Decision 7A) or already + // covered. Mirrors the 30s grace timer in src/core/minions/worker.ts (the + // setTimeout that logs "handler ignored abort signal (force-evicted)"). + const FORCE_EVICT_DEADLINE_MS = 30_000; + for (const pr of phaseResults) { + if (pr.duration_ms > FORCE_EVICT_DEADLINE_MS) { + console.warn( + `[cycle] phase '${pr.phase}' ran ${Math.round(pr.duration_ms / 1000)}s, exceeding the ` + + `${FORCE_EVICT_DEADLINE_MS / 1000}s worker force-evict deadline — if this cycle is ` + + `force-evicted on abort, '${pr.phase}' is the likely cause (#1972).`, + ); + } + } + // v0.38 (codex r1 P0-5): persist per-source cycle completion timestamp // when the cycle ran successfully against an explicit source. Read by // autopilot's per-source freshness gate next tick. Skipped when: @@ -2200,7 +2256,7 @@ export async function runCycle( // Best-effort: a write failure does NOT change the CycleReport status. // The cost of writing the wrong timestamp post-failure is higher than // the cost of missing a successful write (next cycle will redo work). - if (opts.sourceId && engine && !dryRun && (status === 'ok' || status === 'clean' || status === 'partial')) { + if (opts.sourceId && engine && !dryRun && !aborted && (status === 'ok' || status === 'clean' || status === 'partial')) { try { await engine.updateSourceConfig(opts.sourceId, { last_full_cycle_at: new Date().toISOString(), @@ -2215,7 +2271,9 @@ export async function runCycle( schema_version: '1', timestamp, duration_ms, - status, + status: aborted ? 'partial' : status, + ...(aborted ? { reason: 'aborted' } : {}), + ...(reapedLocks ? { reaped_dead_holder_locks: reapedLocks } : {}), brain_dir: opts.brainDir, phases: phaseResults, totals, diff --git a/src/core/cycle/extract-facts.ts b/src/core/cycle/extract-facts.ts index dda7fbd2f..306e58a5d 100644 --- a/src/core/cycle/extract-facts.ts +++ b/src/core/cycle/extract-facts.ts @@ -42,6 +42,7 @@ import { type PhantomPassResult, } from './phantom-redirect.ts'; import { embed, isAvailable } from '../ai/gateway.ts'; +import { isAborted } from '../abort-check.ts'; export interface ExtractFactsOpts { /** Subset of slugs to reconcile. undefined = walk every page in the brain. */ @@ -59,6 +60,13 @@ export interface ExtractFactsOpts { * standard fence-reconcile loop). */ brainDir?: string; + /** + * #1972: cooperative-abort signal. Checked at the top of the per-page loop, + * threaded into the phantom-redirect pass's lock-retry + phantom loop, and + * forwarded to the per-page batch embed — so a long extract_facts bails well + * under the worker's 30s force-evict instead of running to completion. + */ + signal?: AbortSignal; } export interface ExtractFactsResult { @@ -141,6 +149,7 @@ export async function runExtractFacts( opts.brainDir, sourceId, opts.dryRun ?? false, + opts.signal, ); } catch (e) { // The pass owns its own per-phantom try/catch; reaching this catch @@ -188,6 +197,10 @@ export async function runExtractFacts( // ── Reconcile each page ─────────────────────────────────────── for (const slug of slugs) { + // #1972: bail at the top of the per-page loop on abort. Each page is an + // independent delete-then-insert commit, so breaking leaves a consistent + // partial state; the receipt/rollup below still runs with partial counts. + if (isAborted(opts.signal)) break; result.pagesScanned += 1; const page = await engine.getPage(slug, { sourceId }); @@ -235,7 +248,9 @@ export async function runExtractFacts( if (isAvailable('embedding') && extracted.length > 0) { try { const texts = extracted.map(e => e.fact); - const embeddings = await embed(texts); + // #1972: forward the abort signal so a cancelled cycle's in-flight + // batch embed (a network call) is itself abortable, not just the loop. + const embeddings = await embed(texts, { abortSignal: opts.signal }); // Defensive: embed should return one vector per input; if the // gateway returns a partial array (provider partial-batch retry // returning fewer than requested), only fill what we have. diff --git a/src/core/cycle/phantom-redirect.ts b/src/core/cycle/phantom-redirect.ts index 48259ead3..613fefe1a 100644 --- a/src/core/cycle/phantom-redirect.ts +++ b/src/core/cycle/phantom-redirect.ts @@ -54,6 +54,7 @@ import { } from '../facts-fence.ts'; import { parseMarkdown, splitBody, serializeMarkdown } from '../markdown.ts'; import { tryAcquireDbLock, syncLockId, type DbLockHandle } from '../db-lock.ts'; +import { isAborted } from '../abort-check.ts'; import { logPhantomEvent, type PhantomOutcome } from '../facts/phantom-audit.ts'; /** Tagged-union outcome of a single phantom-redirect attempt. */ @@ -206,10 +207,14 @@ function computePageContentHash(parsed: { async function acquireLockWithRetry( engine: BrainEngine, lockId: string, + signal?: AbortSignal, ): Promise { const deadline = Date.now() + LOCK_TOTAL_TIMEOUT_MS; let handle = await tryAcquireDbLock(engine, lockId, LOCK_TTL_MINUTES); while (!handle) { + // #1972: bail immediately on abort instead of retrying for the full 30s — + // otherwise this loop alone can blow past the worker's 30s force-evict. + if (isAborted(signal)) return null; if (Date.now() >= deadline) return null; await new Promise((resolve) => setTimeout(resolve, LOCK_RETRY_INTERVAL_MS)); handle = await tryAcquireDbLock(engine, lockId, LOCK_TTL_MINUTES); @@ -518,6 +523,7 @@ export async function runPhantomRedirectPass( brainDir: string, sourceId: string, dryRun: boolean, + signal?: AbortSignal, ): Promise { const result = emptyPhantomPassResult(); const limitRaw = process.env.GBRAIN_PHANTOM_REDIRECT_LIMIT; @@ -531,7 +537,7 @@ export async function runPhantomRedirectPass( // contention; we loop with 1s backoff up to 30s total. // v0.40 D16: per-source lock matching performSync's posture. Phantom + same- // source sync still serialize; cross-source parallel sync proceeds unblocked. - const lock = await acquireLockWithRetry(engine, syncLockId(sourceId)); + const lock = await acquireLockWithRetry(engine, syncLockId(sourceId), signal); if (!lock) { logPhantomEvent({ outcome: 'pass_skipped_lock_busy', source_id: sourceId }); result.lock_busy = true; @@ -554,6 +560,8 @@ export async function runPhantomRedirectPass( const touchedSet = new Set(); for (let i = 0; i < Math.min(rows.length, limit); i++) { + // #1972: bail between phantoms on abort (each is independently committed). + if (isAborted(signal)) break; const slug = rows[i].slug; const page = await engine.getPage(slug, { sourceId }); if (!page) continue; diff --git a/src/core/cycle/phases/consolidate.ts b/src/core/cycle/phases/consolidate.ts index 37875c811..8ecd9619c 100644 --- a/src/core/cycle/phases/consolidate.ts +++ b/src/core/cycle/phases/consolidate.ts @@ -25,11 +25,18 @@ import type { BrainEngine, FactRow } from '../../engine.ts'; import type { PhaseResult } from '../../cycle.ts'; import { cosineSimilarity } from '../../facts/classify.ts'; +import { isAborted } from '../../abort-check.ts'; export interface ConsolidatePhaseOpts { dryRun?: boolean; /** In-phase keepalive callback. Awaited between buckets. */ yieldDuringPhase?: () => Promise; + /** + * #1972: cooperative-abort signal. Checked at the top of the bucket loop so a + * long consolidate relinquishes its worker slot well under the 30s + * force-evict instead of running to completion after cancellation. + */ + signal?: AbortSignal; /** Cosine cluster threshold. Default 0.85. */ clusterThreshold?: number; /** Minimum facts per (source, entity) bucket before consolidation. Default 3. */ @@ -83,6 +90,11 @@ export async function runPhaseConsolidate( } for (const b of buckets) { + // #1972: bail at the top of the bucket loop on abort. Each prior bucket's + // per-row INSERT/consolidate is already committed, so breaking returns a + // valid partial envelope (the inner cluster loop is bounded at limit 100, + // so no inner guard is needed). + if (isAborted(opts.signal)) break; if (opts.yieldDuringPhase) { try { await opts.yieldDuringPhase(); } catch { /* keepalive errors non-fatal */ } } diff --git a/src/core/db-lock.ts b/src/core/db-lock.ts index bbdf1b2ac..436b59e16 100644 --- a/src/core/db-lock.ts +++ b/src/core/db-lock.ts @@ -340,58 +340,37 @@ export interface LockSnapshot { ms_since_last_refresh: number | null; } -export async function inspectLock(engine: BrainEngine, lockId: string): Promise { - const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; - const maybePGLite = engine as unknown as { - db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; - }; +/** + * Raw row shape returned by every `gbrain_cycle_locks` SELECT. Lives here so + * `selectLockRows` (the single canonical reader) and its mapper share one type. + */ +interface RawLockRow { + id?: string; + holder_pid?: number; + holder_host?: string; + acquired_at?: Date | string; + ttl_expires_at?: Date | string; + last_refreshed_at?: Date | string | null; +} - let row: { - id?: string; - holder_pid?: number; - holder_host?: string; - acquired_at?: Date | string; - ttl_expires_at?: Date | string; - last_refreshed_at?: Date | string | null; - } | undefined; - - if (engine.kind === 'postgres' && maybePG.sql) { - const sql = maybePG.sql as any; - const rows = await sql` - SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at - FROM gbrain_cycle_locks - WHERE id = ${lockId} - `; - row = rows[0]; - } else if (engine.kind === 'pglite' && maybePGLite.db) { - const { rows } = await maybePGLite.db.query( - `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at - FROM gbrain_cycle_locks - WHERE id = $1`, - [lockId], - ); - row = rows[0] as typeof row; - } else { - throw new Error(`Unknown engine kind for inspectLock: ${engine.kind}`); - } - - if (!row || row.holder_pid === undefined || !row.acquired_at || !row.ttl_expires_at) return null; +/** Canonical column list for every lock SELECT — keep in lockstep with `RawLockRow`. */ +const LOCK_SELECT_COLS = 'id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at'; +/** + * Row → `LockSnapshot` mapper. Coerces postgres.js Date|string columns to Date + * and computes the derived fields (age_ms, ttl_expired, ms_since_last_refresh). + * Returns null for a structurally-incomplete row (missing pid / acquired_at / + * ttl). v0.41.13.0: last_refreshed_at may be NULL on pre-v98 brains. + */ +function rowToLockSnapshot(row: RawLockRow, now: number): LockSnapshot | null { + if (row.holder_pid === undefined || !row.acquired_at || !row.ttl_expires_at) return null; const acquired = row.acquired_at instanceof Date ? row.acquired_at : new Date(row.acquired_at); const ttlExpires = row.ttl_expires_at instanceof Date ? row.ttl_expires_at : new Date(row.ttl_expires_at); - const now = Date.now(); - // v0.41.13.0: last_refreshed_at may be NULL on pre-v98 brains that have - // the column but no acquire has happened since the migration ran. Render - // both `last_refreshed_at` and the computed delta as null so callers can - // distinguish "never observed a refresh" from "refresh fired N ms ago". const lastRefreshed = row.last_refreshed_at == null ? null - : (row.last_refreshed_at instanceof Date - ? row.last_refreshed_at - : new Date(row.last_refreshed_at)); - + : (row.last_refreshed_at instanceof Date ? row.last_refreshed_at : new Date(row.last_refreshed_at)); return { - id: lockId, + id: String(row.id ?? ''), holder_pid: Number(row.holder_pid), holder_host: String(row.holder_host ?? ''), acquired_at: acquired, @@ -403,6 +382,59 @@ export async function inspectLock(engine: BrainEngine, lockId: string): Promise< }; } +interface SelectLockRowsOpts { + /** Return only the row for this exact lock id (the `inspectLock` case). */ + lockId?: string; + /** Return only rows whose ttl_expires_at < NOW() (the `listStaleLocks` case). */ + staleOnly?: boolean; +} + +/** + * The single canonical reader for `gbrain_cycle_locks`. Engine-branches once + * (postgres / pglite) so `inspectLock`, `listStaleLocks`, and the reaper don't + * each re-roll the SELECT + Date-coercion (previously triplicated). Returns + * fully-mapped `LockSnapshot[]`; callers filter in JS (the table holds 0-2 rows + * in practice, so a no-WHERE full read is cheap and keeps the SQL trivial). + */ +async function selectLockRows(engine: BrainEngine, opts: SelectLockRowsOpts = {}): Promise { + const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; + const maybePGLite = engine as unknown as { + db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; + }; + + let rows: RawLockRow[]; + if (engine.kind === 'postgres' && maybePG.sql) { + const sql = maybePG.sql as any; + if (opts.lockId !== undefined) { + rows = await sql`SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE id = ${opts.lockId}`; + } else if (opts.staleOnly) { + rows = await sql`SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE ttl_expires_at < NOW() ORDER BY acquired_at`; + } else { + rows = await sql`SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks ORDER BY acquired_at`; + } + } else if (engine.kind === 'pglite' && maybePGLite.db) { + if (opts.lockId !== undefined) { + rows = (await maybePGLite.db.query(`SELECT ${LOCK_SELECT_COLS} FROM gbrain_cycle_locks WHERE id = $1`, [opts.lockId])).rows as RawLockRow[]; + } else if (opts.staleOnly) { + rows = (await maybePGLite.db.query(`SELECT ${LOCK_SELECT_COLS} FROM gbrain_cycle_locks WHERE ttl_expires_at < NOW() ORDER BY acquired_at`)).rows as RawLockRow[]; + } else { + rows = (await maybePGLite.db.query(`SELECT ${LOCK_SELECT_COLS} FROM gbrain_cycle_locks ORDER BY acquired_at`)).rows as RawLockRow[]; + } + } else { + throw new Error(`Unknown engine kind for selectLockRows: ${engine.kind}`); + } + + const now = Date.now(); + return rows + .map((r) => rowToLockSnapshot(r, now)) + .filter((s): s is LockSnapshot => s !== null); +} + +export async function inspectLock(engine: BrainEngine, lockId: string): Promise { + const rows = await selectLockRows(engine, { lockId }); + return rows[0] ?? null; +} + /** * v0.41.6.0 D3: list every lock whose TTL has expired. Used by gbrain * doctor's `stale_locks` check. The query reuses the same canonical @@ -410,55 +442,7 @@ export async function inspectLock(engine: BrainEngine, lockId: string): Promise< * UPDATE-on-conflict already trusts — no parallel heuristic. */ export async function listStaleLocks(engine: BrainEngine): Promise { - const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; - const maybePGLite = engine as unknown as { - db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; - }; - - let rows: Array<{ id?: string; holder_pid?: number; holder_host?: string; acquired_at?: Date | string; ttl_expires_at?: Date | string; last_refreshed_at?: Date | string | null }>; - - if (engine.kind === 'postgres' && maybePG.sql) { - const sql = maybePG.sql as any; - rows = await sql` - SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at - FROM gbrain_cycle_locks - WHERE ttl_expires_at < NOW() - ORDER BY acquired_at - `; - } else if (engine.kind === 'pglite' && maybePGLite.db) { - const result = await maybePGLite.db.query( - `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at - FROM gbrain_cycle_locks - WHERE ttl_expires_at < NOW() - ORDER BY acquired_at`, - ); - rows = result.rows as typeof rows; - } else { - throw new Error(`Unknown engine kind for listStaleLocks: ${engine.kind}`); - } - - const now = Date.now(); - return rows - .filter(r => r.holder_pid !== undefined && r.acquired_at && r.ttl_expires_at) - .map(r => { - const acquired = r.acquired_at instanceof Date ? r.acquired_at : new Date(r.acquired_at!); - const ttl = r.ttl_expires_at instanceof Date ? r.ttl_expires_at : new Date(r.ttl_expires_at!); - // v0.41.13.0: last_refreshed_at may be NULL on pre-v98 brains. - const lastRefreshed = r.last_refreshed_at == null - ? null - : (r.last_refreshed_at instanceof Date ? r.last_refreshed_at : new Date(r.last_refreshed_at)); - return { - id: String(r.id ?? ''), - holder_pid: Number(r.holder_pid), - holder_host: String(r.holder_host ?? ''), - acquired_at: acquired, - ttl_expires_at: ttl, - age_ms: now - acquired.getTime(), - ttl_expired: true, - last_refreshed_at: lastRefreshed, - ms_since_last_refresh: lastRefreshed ? now - lastRefreshed.getTime() : null, - }; - }); + return selectLockRows(engine, { staleOnly: true }); } /** @@ -588,6 +572,110 @@ export async function deleteLockRowIfStale( throw new Error(`Unknown engine kind for deleteLockRowIfStale: ${engine.kind}`); } +/** + * #1972 — snapshot-matched verify-and-delete for the background reaper. + * + * Unlike `deleteLockRow` (id + holder_pid only), this ALSO pins the observed + * `acquired_at`, closing the TOCTOU window the reaper opens by reading rows + * before deleting them: between the SELECT and the DELETE, the dead holder's + * row could be replaced by a NEW holder that reused the same numeric PID + * (PID-space wraps). That new row carries a newer `acquired_at`, so the match + * fails and the DELETE is a safe no-op. + * + * Why `date_trunc('milliseconds', acquired_at) = $3` and not bare equality: + * postgres.js parses timestamptz into a JS Date (millisecond precision), so the + * snapshot we hold has already lost the microseconds that `acquired_at DEFAULT + * NOW()` writes. A bare `acquired_at = $3` would therefore NEVER match in + * production (microsecond stored value ≠ ms-truncated param) — the reaper would + * silently delete nothing. Truncating both sides to ms makes the comparison + * round-trip-safe on Postgres + PGLite, while a real takeover (seconds later) + * still differs by far more than a millisecond. + */ +export async function deleteLockRowExact( + engine: BrainEngine, + lockId: string, + holderPid: number, + acquiredAt: Date, +): Promise<{ deleted: boolean }> { + const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; + const maybePGLite = engine as unknown as { + db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; + }; + + if (engine.kind === 'postgres' && maybePG.sql) { + const sql = maybePG.sql as any; + const rows: Array<{ id: string }> = await sql` + DELETE FROM gbrain_cycle_locks + WHERE id = ${lockId} + AND holder_pid = ${holderPid} + AND date_trunc('milliseconds', acquired_at) = ${acquiredAt} + RETURNING id + `; + return { deleted: rows.length > 0 }; + } + if (engine.kind === 'pglite' && maybePGLite.db) { + const { rows } = await maybePGLite.db.query( + `DELETE FROM gbrain_cycle_locks + WHERE id = $1 + AND holder_pid = $2 + AND date_trunc('milliseconds', acquired_at) = $3 + RETURNING id`, + [lockId, holderPid, acquiredAt], + ); + return { deleted: rows.length > 0 }; + } + throw new Error(`Unknown engine kind for deleteLockRowExact: ${engine.kind}`); +} + +/** + * #1972 — host-scoped reaper for dead-holder locks. Closes the gap where a sync + * (or cycle) that crashed (OOM, recycle, SIGKILL) strands its lock row until + * something contends for it — a low-traffic source could look "syncing" for + * hours. `tryAcquireDbLock` only reclaims on contention; this is the background + * sweep. Intended to run at cycle start (and under `gbrain doctor --fix` for + * no-autopilot brains). + * + * Scoped to the `gbrain-sync:*` and `gbrain-cycle`/`gbrain-cycle:*` namespaces + * ONLY. `gbrain_cycle_locks` is shared by enrich, the minion supervisor, + * reindex, schema-pack, and elections (`tryWithDbElection`) — a blanket sweep + * would change their TTL-failover timing. Those keep their existing + * on-contention + TTL behavior. + * + * reapDeadHolderLocks (host-scoped, sync/cycle namespaces only) + * selectLockRows → for each row: + * ├─ id not in sync/cycle namespace ───────→ KEEP (blast-radius scope) + * ├─ holder_host != thisHost ──────────────→ KEEP (cross_host; can't probe) + * ├─ kill(pid,0) == alive / EPERM ─────────→ KEEP (live or not-ours) + * ├─ ESRCH && age < 60s grace ─────────────→ KEEP (PID-reuse defense) + * └─ ESRCH && age ≥ 60s grace ─────────────→ deleteLockRowExact(id, pid, acquired_at) + * (snapshot-matched: reused-PID + * fresh row has newer acquired_at → no-op) + */ +function isReapableNamespace(lockId: string): boolean { + return ( + lockId === 'gbrain-cycle' || + lockId.startsWith('gbrain-cycle:') || + lockId.startsWith('gbrain-sync:') + ); +} + +export async function reapDeadHolderLocks( + engine: BrainEngine, + opts: HolderLivenessOpts = {}, +): Promise<{ reaped: number; reapedIds: string[] }> { + const rows = await selectLockRows(engine); + const reapedIds: string[] = []; + for (const s of rows) { + if (!isReapableNamespace(s.id)) continue; + // isHolderDeadLocally combines same-host + ESRCH + the 60s reuse grace, + // so pure-PID-liveness (which PID-space wrap defeats) is never used alone. + if (!isHolderDeadLocally(s.holder_pid, s.holder_host, s.age_ms, opts)) continue; + const { deleted } = await deleteLockRowExact(engine, s.id, s.holder_pid, s.acquired_at); + if (deleted) reapedIds.push(s.id); + } + return { reaped: reapedIds.length, reapedIds }; +} + /** * v0.40 (Federated Sync v2): per-source sync lock helper. * diff --git a/src/core/db.ts b/src/core/db.ts index a38290509..99dae2136 100644 --- a/src/core/db.ts +++ b/src/core/db.ts @@ -7,6 +7,47 @@ import { verifySchema } from './schema-verify.ts'; let sql: ReturnType | null = null; let connectedUrl: string | null = null; +/** + * #1972: hard upper bound (seconds) on a single pool `.end()` drain. postgres.js + * accepts `{ timeout }` but applies it internally — against PgBouncer + * transaction-mode the drain can still hang, and a stubbed `.end()` ignores it + * entirely. So `endPoolBounded` ALSO wraps each end in a gbrain-owned + * Promise.race and passes this value as the postgres.js hint so a healthy drain + * still finishes fast. + */ +export const POOL_END_TIMEOUT_SECONDS = 2; + +/** + * #1972: end a postgres.js pool with a gbrain-owned hard bound. Resolves as soon + * as `.end()` settles OR after POOL_END_TIMEOUT_SECONDS + a small slack — so + * teardown never hangs (the prior bare `.end()` blocked until the CLI's 10s + * force-exit fired, which `process.exit()`s and truncated pending stdout, e.g. + * #1959's relational query came back empty). Never throws: a teardown that + * rejects is worse than one that races past a stuck socket. The race timer is + * the real guarantee; `{ timeout }` just lets a healthy drain return in ms. + * + * Note callers that close MULTIPLE pools should `Promise.all` them rather than + * awaiting sequentially, so the per-pool bounds run concurrently instead of + * stacking. + */ +export async function endPoolBounded( + pool: { end: (opts?: { timeout?: number }) => Promise }, +): Promise { + let timer: ReturnType | undefined; + const guard = new Promise((resolve) => { + timer = setTimeout(resolve, POOL_END_TIMEOUT_SECONDS * 1000 + 500); + timer.unref?.(); + }); + try { + await Promise.race([ + pool.end({ timeout: POOL_END_TIMEOUT_SECONDS }).catch(() => { /* idempotent / already-closed */ }), + guard, + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + /** * Default pool size for Postgres connections. Users on the Supabase transaction * pooler (port 6543) or any multi-tenant pooler can lower this to avoid @@ -258,7 +299,7 @@ export async function disconnect(): Promise { const s = sql; sql = null; connectedUrl = null; - if (s) await s.end(); + if (s) await endPoolBounded(s); } export async function initSchema(): Promise { diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 0c130cd32..cb7bcb14a 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -248,7 +248,9 @@ export class PostgresEngine implements BrainEngine { this.connectionManager = null; } if (this._sql) { - await this._sql.end(); + // #1972: gbrain-owned hard bound so a PgBouncer drain that never settles + // can't block teardown until the CLI's 10s force-exit truncates stdout. + await db.endPoolBounded(this._sql); this._sql = null; // After this point, _connectionStyle stays 'instance' so a second // disconnect() is a no-op rather than falling through and clearing diff --git a/test/cycle-abort.test.ts b/test/cycle-abort.test.ts index b160f6d93..3175421bf 100644 --- a/test/cycle-abort.test.ts +++ b/test/cycle-abort.test.ts @@ -152,3 +152,50 @@ describe('autopilot-cycle handler contract (v0.20.5)', () => { expect(checkCalls).toBeGreaterThanOrEqual(6); }); }); + +describe('#1972 — complete cooperative-abort coverage', () => { + test('aborted signal makes runCycle report partial + reason "aborted" (terminal guard)', async () => { + // phases:[] means no between-phase checkAborted fires, so execution reaches + // the TERMINAL guard (Codex #9). With a pre-aborted signal it must NOT + // report success — status 'partial', reason 'aborted'. + const { runCycle } = await import('../src/core/cycle.ts'); + const abort = new AbortController(); + abort.abort(new Error('timeout')); + const report = await runCycle(null, { + brainDir: '/nonexistent-for-test', + phases: [], + signal: abort.signal, + }); + expect(report.status).toBe('partial'); + expect(report.reason).toBe('aborted'); + }); + + test('cycle.ts threads opts.signal into every long phase + guards the success stamp', async () => { + const fs = await import('fs'); + const src = fs.readFileSync(new URL('../src/core/cycle.ts', import.meta.url), 'utf8'); + const body = src.slice(src.indexOf('export async function runCycle')); + // Each long phase receives the signal. + expect(body).toContain('runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal)'); + expect(body).toMatch(/runPhaseExtractFacts\([^)]*opts\.signal\)/); + expect(body).toContain('signal: opts.signal'); // consolidate opts + expect(body).toContain('runPhaseLint(brainDir, dryRun, engine, opts.signal)'); + // Reaper runs at cycle start. + expect(body).toContain('reapDeadHolderLocks(engine)'); + // Terminal guard: the success stamp is gated on !aborted, and the report + // carries reason 'aborted'. + expect(body).toContain('!aborted'); + expect(body).toContain("reason: 'aborted'"); + // Phase-duration force-evict attribution log (T11). + expect(body).toContain('FORCE_EVICT_DEADLINE_MS'); + }); + + test('every long phase core checks isAborted in its batch loop', async () => { + const fs = await import('fs'); + const read = (p: string) => fs.readFileSync(new URL(p, import.meta.url), 'utf8'); + expect(read('../src/commands/extract.ts')).toContain('if (isAborted(signal)) return;'); + expect(read('../src/core/cycle/extract-facts.ts')).toContain('if (isAborted(opts.signal)) break;'); + expect(read('../src/core/cycle/phases/consolidate.ts')).toContain('if (isAborted(opts.signal)) break;'); + expect(read('../src/core/cycle/phantom-redirect.ts')).toContain('if (isAborted(signal)) break;'); + expect(read('../src/commands/lint.ts')).toContain('if (isAborted(opts.signal)) break;'); + }); +}); diff --git a/test/cycle-consolidate.test.ts b/test/cycle-consolidate.test.ts index 812bffcad..fe3462bc5 100644 --- a/test/cycle-consolidate.test.ts +++ b/test/cycle-consolidate.test.ts @@ -10,6 +10,7 @@ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { configureGateway } from '../src/core/ai/gateway.ts'; import { runPhaseConsolidate } from '../src/core/cycle/phases/consolidate.ts'; let engine: PGLiteEngine; @@ -17,6 +18,20 @@ let engine: PGLiteEngine; beforeAll(async () => { engine = new PGLiteEngine(); await engine.connect({}); + // initSchema() bakes the facts.embedding dim from the gateway's configured + // embedding model; the default is now 1280-d (ZE). This file's fixtures are + // 1536-d, so pin the legacy 1536-d OpenAI config (matching + // test/helpers/legacy-embedding-preload.ts) right before initSchema. The + // global preload sets this, but a co-sharded test that calls resetGateway() + // in its teardown nulls it, leaving initSchema to fall back to the 1280-d + // default and build a halfvec(1280) column the 1536-d inserts can't fill. + // Re-pinning here makes the schema deterministic regardless of shard + // neighbors (surfaced when #1972's new test files reshuffled the shards). + configureGateway({ + embedding_model: 'openai:text-embedding-3-large', + embedding_dimensions: 1536, + env: { ...process.env }, + }); await engine.initSchema(); }); diff --git a/test/db-lock-heartbeat-takeover.test.ts b/test/db-lock-heartbeat-takeover.test.ts index fff0d4381..a4f479547 100644 --- a/test/db-lock-heartbeat-takeover.test.ts +++ b/test/db-lock-heartbeat-takeover.test.ts @@ -16,6 +16,7 @@ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; import { hostname } from 'os'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { withEnv } from './helpers/with-env.ts'; import { tryAcquireDbLock, resolveStealGraceSeconds, @@ -81,22 +82,16 @@ describe('resolveStealGraceSeconds', () => { expect(resolveStealGraceSeconds(1)).toBe(60); }); - test('env override wins', () => { - process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = '123'; - try { + test('env override wins', async () => { + await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: '123' }, async () => { expect(resolveStealGraceSeconds(30)).toBe(123); - } finally { - delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS; - } + }); }); - test('bad env override falls back to derived', () => { - process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = 'nope'; - try { + test('bad env override falls back to derived', async () => { + await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: 'nope' }, async () => { expect(resolveStealGraceSeconds(30)).toBe(600); - } finally { - delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS; - } + }); }); }); diff --git a/test/db-lock-reap.test.ts b/test/db-lock-reap.test.ts new file mode 100644 index 000000000..2b32da5a1 --- /dev/null +++ b/test/db-lock-reap.test.ts @@ -0,0 +1,162 @@ +/** + * #1972 — host-scoped reaper for dead-holder sync/cycle locks. + * + * Covers: + * - reapDeadHolderLocks: namespace scope (sync/cycle only, NOT election/etc), + * same-host dead-PID reaped regardless of TTL, live/cross-host/within-grace + * kept. Uses the injectable process.kill seam so it's deterministic. + * - deleteLockRowExact: snapshot-matched delete (the TOCTOU defense) — a + * non-matching acquired_at is a no-op, the matching one deletes. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { hostname } from 'os'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { + reapDeadHolderLocks, + deleteLockRowExact, + inspectLock, + HOLDER_TAKEOVER_GRACE_MS, + type HolderLivenessOpts, +} from '../src/core/db-lock.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await engine.executeRaw(`DELETE FROM gbrain_cycle_locks`); +}); + +const LOCAL = hostname(); +const OLD_S = Math.ceil(HOLDER_TAKEOVER_GRACE_MS / 1000) + 60; // comfortably past grace +const YOUNG_S = 5; + +/** Insert a lock row with a given age + namespace. ttlFuture controls whether + * the row is TTL-expired (to prove the reaper reaps dead holders regardless). */ +async function seedLock( + id: string, + holderPid: number, + holderHost: string, + ageSeconds: number, + ttlFuture = true, +) { + const ttl = ttlFuture ? `NOW() + INTERVAL '10 minutes'` : `NOW() - INTERVAL '1 minute'`; + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ($1, $2, $3, NOW() - ($4 || ' seconds')::interval, ${ttl}, NOW() - ($4 || ' seconds')::interval)`, + [id, holderPid, holderHost, String(ageSeconds)], + ); +} + +/** process.kill seam: every pid is dead (ESRCH) except those in `live`. */ +function killSeam(live: Set): HolderLivenessOpts { + return { + localHost: LOCAL, + processKill: (pid: number) => { + if (live.has(pid)) return; // alive + const e = new Error('no such process') as NodeJS.ErrnoException; + e.code = 'ESRCH'; + throw e; + }, + }; +} + +async function lockIds(): Promise { + const rows = await engine.executeRaw<{ id: string }>(`SELECT id FROM gbrain_cycle_locks ORDER BY id`); + return rows.map(r => r.id); +} + +describe('reapDeadHolderLocks', () => { + test('reaps same-host dead-PID sync + cycle locks (regardless of TTL)', async () => { + await seedLock('gbrain-sync:src-a', 900001, LOCAL, OLD_S, /*ttlFuture*/ true); // dead, TTL NOT expired + await seedLock('gbrain-cycle', 900002, LOCAL, OLD_S, false); // dead, TTL expired + await seedLock('gbrain-cycle:src-b', 900003, LOCAL, OLD_S, true); // dead, TTL NOT expired + + const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set())); + expect(reaped).toBe(3); + expect(reapedIds.sort()).toEqual(['gbrain-cycle', 'gbrain-cycle:src-b', 'gbrain-sync:src-a']); + expect(await lockIds()).toEqual([]); + }); + + test('keeps a live same-host holder', async () => { + await seedLock('gbrain-sync:src-live', process.pid, LOCAL, OLD_S); + const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid]))); + expect(reaped).toBe(0); + expect(await lockIds()).toEqual(['gbrain-sync:src-live']); + }); + + test('keeps a dead holder still within the PID-reuse grace window', async () => { + await seedLock('gbrain-sync:src-young', 900010, LOCAL, YOUNG_S); + const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set())); + expect(reaped).toBe(0); + expect(await lockIds()).toEqual(['gbrain-sync:src-young']); + }); + + test('keeps a cross-host holder (cannot probe a remote PID)', async () => { + await seedLock('gbrain-sync:src-xhost', 900011, 'a-different-host', OLD_S); + const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set())); + expect(reaped).toBe(0); + expect(await lockIds()).toEqual(['gbrain-sync:src-xhost']); + }); + + test('NEVER reaps a non-sync/cycle namespace, even with a dead PID (blast radius)', async () => { + await seedLock('gbrain-election:leader', 900020, LOCAL, OLD_S); + await seedLock('some-other-lock', 900021, LOCAL, OLD_S); + const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set())); + expect(reaped).toBe(0); + expect(await lockIds()).toEqual(['gbrain-election:leader', 'some-other-lock']); + }); + + test('empty table → {reaped:0}', async () => { + const r = await reapDeadHolderLocks(engine, killSeam(new Set())); + expect(r).toEqual({ reaped: 0, reapedIds: [] }); + }); + + test('mixed set: reaps only the eligible rows', async () => { + await seedLock('gbrain-sync:dead', 900030, LOCAL, OLD_S); // reap + await seedLock('gbrain-sync:live', process.pid, LOCAL, OLD_S); // keep (live) + await seedLock('gbrain-cycle:xhost', 900031, 'other', OLD_S); // keep (cross-host) + await seedLock('gbrain-election:x', 900032, LOCAL, OLD_S); // keep (namespace) + const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid]))); + expect(reaped).toBe(1); + expect(reapedIds).toEqual(['gbrain-sync:dead']); + expect(await lockIds()).toEqual(['gbrain-cycle:xhost', 'gbrain-election:x', 'gbrain-sync:live']); + }); +}); + +describe('deleteLockRowExact (snapshot-matched, TOCTOU defense)', () => { + test('no-op when acquired_at does not match; deletes when it does', async () => { + await seedLock('gbrain-sync:exact', 900040, LOCAL, OLD_S); + const snap = await inspectLock(engine, 'gbrain-sync:exact'); + expect(snap).not.toBeNull(); + + // Simulate a reused-PID takeover: same id + pid, but a different (newer) + // acquired_at than the one the reaper snapshotted → must NOT delete. + const wrong = new Date(snap!.acquired_at.getTime() + 3_600_000); + const miss = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, wrong); + expect(miss.deleted).toBe(false); + expect(await lockIds()).toEqual(['gbrain-sync:exact']); + + // The matching snapshot deletes. + const hit = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, snap!.acquired_at); + expect(hit.deleted).toBe(true); + expect(await lockIds()).toEqual([]); + }); + + test('no-op when holder_pid does not match', async () => { + await seedLock('gbrain-sync:pidguard', 900050, LOCAL, OLD_S); + const snap = await inspectLock(engine, 'gbrain-sync:pidguard'); + const res = await deleteLockRowExact(engine, 'gbrain-sync:pidguard', 111111, snap!.acquired_at); + expect(res.deleted).toBe(false); + expect(await lockIds()).toEqual(['gbrain-sync:pidguard']); + }); +}); diff --git a/test/extract-abort.test.ts b/test/extract-abort.test.ts new file mode 100644 index 000000000..5580c40ba --- /dev/null +++ b/test/extract-abort.test.ts @@ -0,0 +1,67 @@ +/** + * #1972 (Codex #7) — cooperative abort on the cycle-reachable extract paths. + * + * The cycle calls runExtractCore with `slugs` defined (incremental → + * extractForSlugs) or undefined (full walk → extractLinksFromDir / + * extractTimelineFromDir). All three must bail promptly on a pre-aborted + * signal. Behavioral, dryRun-only (no DB writes, so no FK setup needed). + */ + +import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; +import { mkdtempSync, writeFileSync, rmSync } from 'fs'; +import { join } from 'path'; +import { tmpdir } from 'os'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { runExtractCore } from '../src/commands/extract.ts'; + +let engine: PGLiteEngine; +let dir: string; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); + dir = mkdtempSync(join(tmpdir(), 'extract-abort-')); + // a.md links to b; both exist so the wikilink resolves. + writeFileSync(join(dir, 'a.md'), '# A\n\nsee [[b]] for more.\n'); + writeFileSync(join(dir, 'b.md'), '# B\n\nplain.\n'); +}); + +afterAll(async () => { + await engine.disconnect(); + rmSync(dir, { recursive: true, force: true }); +}); + +function aborted(): AbortSignal { + const a = new AbortController(); + a.abort(new Error('timeout')); + return a.signal; +} + +describe('extract cooperative abort', () => { + test('incremental path (extractForSlugs) processes 0 pages when pre-aborted', async () => { + const r = await runExtractCore(engine, { + mode: 'links', dir, dryRun: true, slugs: ['a', 'b'], signal: aborted(), + }); + expect(r.pages_processed).toBe(0); + }); + + test('incremental path processes all pages without a signal', async () => { + const r = await runExtractCore(engine, { + mode: 'links', dir, dryRun: true, slugs: ['a', 'b'], + }); + expect(r.pages_processed).toBe(2); + }); + + test('full-walk path (extractLinksFromDir) creates 0 links when pre-aborted', async () => { + const r = await runExtractCore(engine, { + mode: 'links', dir, dryRun: true, signal: aborted(), + }); + expect(r.links_created).toBe(0); + }); + + test('full-walk path resolves the wikilink without a signal', async () => { + const r = await runExtractCore(engine, { mode: 'links', dir, dryRun: true }); + expect(r.links_created).toBeGreaterThanOrEqual(1); + }); +}); diff --git a/test/phantom-redirect-per-source-lock.test.ts b/test/phantom-redirect-per-source-lock.test.ts index 3d5351f32..65e8f4e66 100644 --- a/test/phantom-redirect-per-source-lock.test.ts +++ b/test/phantom-redirect-per-source-lock.test.ts @@ -31,7 +31,11 @@ describe('phantom-redirect lock contract', () => { test('IRON-RULE: acquireLockWithRetry call passes syncLockId(sourceId)', () => { // Banned: the bare `SYNC_LOCK_ID` constant slipped back in. // Required: the per-source helper threaded with the active sourceId. - expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*\)/); + // #1972: the call now threads an optional abort signal as a third arg + // (acquireLockWithRetry(engine, syncLockId(sourceId), signal)). The IRON-RULE + // is unchanged: it must pass the per-source syncLockId(sourceId), never bare + // SYNC_LOCK_ID. Allow the optional trailing signal arg. + expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*(?:,\s*signal\s*)?\)/); expect(SRC).not.toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*SYNC_LOCK_ID\s*\)/); }); diff --git a/test/postgres-disconnect-bounded.test.ts b/test/postgres-disconnect-bounded.test.ts new file mode 100644 index 000000000..68dea784f --- /dev/null +++ b/test/postgres-disconnect-bounded.test.ts @@ -0,0 +1,40 @@ +/** + * #1972 — gbrain-owned hard bound on pool teardown. + * + * The bug: `pool.end()` against PgBouncer transaction-mode never drains, so + * disconnect blocked until the CLI's 10s force-exit fired and truncated stdout. + * postgres.js's own `{ timeout }` is internal (a stub ignores it; it's not a + * guarantee we own), so `endPoolBounded` wraps every end in a Promise.race we + * control. These tests assert the bound is real (resolves even when `.end()` + * never settles) and that we still pass `{ timeout }` so a healthy drain is fast. + */ + +import { describe, test, expect } from 'bun:test'; +import { endPoolBounded, POOL_END_TIMEOUT_SECONDS } from '../src/core/db.ts'; + +describe('endPoolBounded', () => { + test('resolves fast when .end() settles quickly, forwarding { timeout }', async () => { + let calledWith: unknown; + const pool = { end: async (opts?: { timeout?: number }) => { calledWith = opts; } }; + const t0 = Date.now(); + await endPoolBounded(pool); + expect(Date.now() - t0).toBeLessThan(500); + expect(calledWith).toEqual({ timeout: POOL_END_TIMEOUT_SECONDS }); + }); + + test('resolves within the gbrain bound even when .end() NEVER settles', async () => { + // This is the PgBouncer hang: .end() returns a promise that never resolves. + // The bare `await pool.end()` would hang until the CLI's 10s force-exit. + const pool = { end: () => new Promise(() => { /* never resolves */ }) }; + const t0 = Date.now(); + await endPoolBounded(pool); + const elapsed = Date.now() - t0; + expect(elapsed).toBeGreaterThanOrEqual(POOL_END_TIMEOUT_SECONDS * 1000); + expect(elapsed).toBeLessThan(5000); // well under the CLI's 10s force-exit deadline + }); + + test('never throws when .end() rejects (teardown must not propagate)', async () => { + const pool = { end: async () => { throw new Error('pool boom'); } }; + await expect(endPoolBounded(pool)).resolves.toBeUndefined(); + }); +}); diff --git a/test/postgres-engine-singleton-ownership.test.ts b/test/postgres-engine-singleton-ownership.test.ts index 68e9ec78c..093be2507 100644 --- a/test/postgres-engine-singleton-ownership.test.ts +++ b/test/postgres-engine-singleton-ownership.test.ts @@ -81,7 +81,10 @@ describe('postgres-engine / module-singleton ownership (#1471)', () => { const disconnect = stripComments(extractFn(DB_SRC, 'disconnect')); const snapshotIdx = disconnect.search(/const\s+s\s*=\s*sql/); const nullIdx = disconnect.search(/\bsql\s*=\s*null/); - const endIdx = disconnect.search(/\bawait\s+s\.end\s*\(/); + // #1972: the pool end is now wrapped in the gbrain-owned hard bound + // `endPoolBounded(s)` instead of a bare `s.end()`. The ordering contract is + // unchanged: snapshot + null the singleton BEFORE awaiting the end. + const endIdx = disconnect.search(/\bawait\s+(?:s\.end\s*\(|endPoolBounded\s*\(\s*s\b)/); expect(snapshotIdx).toBeGreaterThanOrEqual(0); expect(nullIdx).toBeGreaterThanOrEqual(0); expect(endIdx).toBeGreaterThanOrEqual(0);