From ec5fed2921176356bc23c11e32023e101eb0fe93 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Wed, 3 Jun 2026 08:42:25 -0700 Subject: [PATCH] =?UTF-8?q?v0.42.20.0=20fix:=20reliability=20wave=20?= =?UTF-8?q?=E2=80=94=20PGLite=20capture=20lock-pin=20+=20Postgres=20reconn?= =?UTF-8?q?ect=20race=20+=20search=20embed-hang=20(#1762=20#1745=20#1775)?= =?UTF-8?q?=20(#1810)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(core): drain fire-and-forget sinks before disconnect via a background-work registry (#1762) New src/core/background-work.ts registry (Map, ordered drain, awaited abort). facts-queue (order 0, abort=shutdown), last-retrieved (1), and eval-capture (3, now self-tracked) register as sinks. Both CLI exit paths (op-dispatch finally + handleCliOnly finally) drain the registry before engine.disconnect() so a PGLite db.close() can't race in-flight work into the re-pump busy-loop that pinned the single-writer lock. Op-dispatch error path converts process.exit(1) to exitCode+return so the finally still drains. * fix(ai): bound every outbound AI call so a stalled provider can't hang (#1762/#1775) withDefaultTimeout composes a per-touchpoint default deadline (chat 300s, embed/multimodal 60s) with any caller signal via AbortSignal.any. Applied at the SDK call layer (chat generateText, expand generateObject, OCR, per-sub-batch embed) — covers native-anthropic + retries — plus per-request multimodal fetch. embedQuery forwards abortSignal. Env: GBRAIN_AI_{CHAT,EMBED,MULTIMODAL}_TIMEOUT_MS. * fix(postgres): module-mode reconnect preserves the shared singleton (#1745) reconnect() branches on connection style. Module-singleton engines re-establish idempotently via db.connect() (no-op when alive) + refresh the ConnectionManager read pool, never db.disconnect() — so a transient blip no longer nulls the shared sql out from under concurrent ops (which threw 'connect() has not been called'). Fail-loud on real connect failure. Instance pools keep teardown+recreate. * fix(search): bound the query-time embed so a stall falls back to keyword (#1775) search/query default to cheap-hybrid (embeds the query); a stalled provider made the embed never settle, so the keyword fallback never engaged and the command force-exited with no output. One shared QueryEmbedDeadline (6s, floored 2s per embed) covers both the cache-lookup and inner embeds via embedQueryBounded (abortSignal + Promise.race) → existing keyword fallback engages. Also registers the search-cache background-work drainer (now bounded). Env: GBRAIN_QUERY_EMBED_TIMEOUT_MS. * test+chore: reliability wave tests + v0.42.11.0 (#1762 #1745 #1775) New: background-work registry unit, query-embed deadline unit, eval-capture drain unit, postgres reconnect E2E (#1745), gbrain capture exit-clean case in the PGLite serial test. Updated fix-wave-structural assertions to the registry shape. VERSION/package.json/CHANGELOG -> 0.42.11.0; TODOS retrofit marked done. Incorporates + hardens PR #1763 (drain-before-disconnect + embed fetch timeout); the residual hung-Haiku hole is closed by the facts shutdown() abort belt. Co-Authored-By: ElliotDrel Co-Authored-By: Claude Opus 4.8 (1M context) * docs: document background-work registry + v0.42.11.0 reliability wave in CLAUDE.md (regen llms) * chore: bump release version 0.42.11.0 -> 0.42.20.0 Rename the reliability-wave release version per request. Trio (VERSION / package.json / CHANGELOG) reconciled; in-code version-tag comments and test fixtures updated; llms regenerated. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: ElliotDrel Co-authored-by: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 116 +++++++++++++++ TODOS.md | 45 +++--- VERSION | 2 +- docs/architecture/KEY_FILES.md | 1 + package.json | 2 +- src/cli.ts | 92 ++++++------ src/core/ai/gateway.ts | 65 +++++++- src/core/background-work.ts | 111 ++++++++++++++ src/core/embedding.ts | 2 +- src/core/eval-capture.ts | 55 +++++++ src/core/facts/queue.ts | 16 ++ src/core/last-retrieved.ts | 9 ++ src/core/postgres-engine.ts | 49 ++++-- src/core/search/hybrid.ts | 139 +++++++++++++++++- test/core/background-work.test.ts | 136 +++++++++++++++++ test/e2e/pglite-cli-exit.serial.test.ts | 40 +++++ test/e2e/postgres-reconnect-singleton.test.ts | 98 ++++++++++++ test/eval-capture-drain.test.ts | 77 ++++++++++ test/fix-wave-structural.test.ts | 77 +++++----- test/search/query-embed-deadline.test.ts | 85 +++++++++++ 20 files changed, 1095 insertions(+), 122 deletions(-) create mode 100644 src/core/background-work.ts create mode 100644 test/core/background-work.test.ts create mode 100644 test/e2e/postgres-reconnect-singleton.test.ts create mode 100644 test/eval-capture-drain.test.ts create mode 100644 test/search/query-embed-deadline.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 04b5dd652..ad1df3284 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,122 @@ All notable changes to GBrain will be documented in this file. +## [0.42.20.0] - 2026-06-03 + +**Three ways GBrain could freeze or go silent are fixed.** This release closes a +reliability cluster around how GBrain shuts down a command and talks to AI +providers. If you've seen `gbrain capture` hang forever, `gbrain dream` print +"connect() has not been called" on Postgres, or `gbrain search` return nothing +and then say "engine.disconnect() did not return within 10000ms" — those are all +fixed. + +What was happening, in plain terms: + +- **`gbrain capture` froze and locked you out (#1762).** On a longer page, + capture finishes, prints the receipt, then kicks off a small background job to + extract facts. On the embedded PGLite database, GBrain was closing the database + *while that job was still running*, which spun the close into a 100%-CPU loop + that never returned and held the database lock open. Every later `gbrain` + command then died with "Timed out waiting for PGLite lock" until you killed the + stuck process. Now GBrain waits for that background work to finish (or cleanly + cancels it) before closing the database. +- **`gbrain dream` failed mid-cycle on Postgres (#1745).** A brief connection + blip made GBrain rebuild its shared database connection — but the rebuild + yanked the connection out from under other work happening at the same time, so + the `sync`, `synthesize`, and link-extraction phases threw "connect() has not + been called" and produced zero pages every cycle. Now a blip recovers without + tearing down the shared connection (Postgres heals dead sockets on its own). +- **`gbrain search` / `query` went silent (#1775, regression from 0.22.8).** + Search now blends keyword + vector results, which means it embeds your query + first. If your embedding provider stalled (one user's did), the embed never + came back, so search never fell through to keyword results and the command + timed out with no output. Now the query-embed is time-bounded (~6 seconds, well + under the exit watchdog), so a stalled provider falls back to keyword results + instead of hanging. + +**Under the hood, one unifying change makes this whole class of bug harder to +reintroduce:** + +- **Every background write is now drained before exit.** GBrain has four + "fire-and-forget" sinks that write to the database after a command returns its + answer (last-retrieved tracking, fact extraction, the search cache, and eval + capture). Each had independently caused or risked the lock-pin. They now + register with one background-work registry that GBrain drains on every exit + path. A fifth sink added later auto-participates — no one has to remember it. +- **Every outbound AI call has a timeout.** Chat, expansion, embeddings, OCR, + and multimodal calls now carry a default wall-clock deadline (configurable via + `GBRAIN_AI_CHAT_TIMEOUT_MS` / `GBRAIN_AI_EMBED_TIMEOUT_MS` / + `GBRAIN_AI_MULTIMODAL_TIMEOUT_MS`). A stalled provider socket can no longer + hang a command forever. This covers the default Anthropic path too, not just + OpenAI-compatible providers. + +Nothing changes in how you use GBrain. These are all teardown/reliability fixes. + +Credit: @ElliotDrel diagnosed the corrected #1762 root cause (the un-drained +facts queue, not the originally-suspected missing teardown contract) and wrote +the first fix in PR #1763, which this release incorporates and hardens. Thanks to +the #1745 and #1775 reporters for the precise repros. + +### To take advantage of v0.42.20.0 + +`gbrain upgrade` is all you need — these are runtime fixes, no migration or +schema change. After upgrading: + +1. If `gbrain capture` was hanging on PGLite, it now returns to the shell and the + next command runs without "Timed out waiting for PGLite lock". +2. If `gbrain dream` was printing "connect() has not been called" on Postgres, + run `gbrain dream` again and check `synth_pages > 0`. +3. If `gbrain search` returned nothing, it now returns ranked keyword results + within a few seconds even when your embedding provider is down. Tune the + query-embed deadline with `GBRAIN_QUERY_EMBED_TIMEOUT_MS` (default 6000) if + your provider is reliably slower. + +### Itemized changes + +- **`src/core/background-work.ts` (NEW):** process background-work registry — + `registerBackgroundWorkDrainer`, `drainAllBackgroundWorkForCliExit`, a + `Map` (idempotent registration), explicit `(order, name)` drain + order (facts first for the live-engine window), and an awaited `abort()` for + stragglers. `__registerDrainerForTest` test seam. +- **Four sinks register drainers:** `facts/queue.ts` (order 0, `abort` = + `shutdown()` which cancels a hung facts:absorb Haiku), `last-retrieved.ts` + (order 1), `search/hybrid.ts` (order 2, with `awaitPendingSearchCacheWrites` + now bounded — was an unbounded `Promise.allSettled`), `eval-capture.ts` + (order 3, new `awaitPendingEvalCaptures` + tracked `captureEvalCandidate`). +- **`src/cli.ts`:** both the op-dispatch finally AND `handleCliOnly`'s finally + call `drainAllBackgroundWorkForCliExit()` before `engine.disconnect()`; + `handleCliOnly` gains the force-exit defense (the drain is the causal fix, the + timer is secondary). Op-dispatch's error path converts `process.exit(1)` → + `exitCode + return` so the finally still drains + disconnects on error. +- **`src/core/ai/gateway.ts`:** `withDefaultTimeout(caller, ms)` composes a + default deadline with any caller signal (`AbortSignal.any`, shorter wins), + threaded into `chat()` (`generateText`), `expand()` (`generateObject` — was + unbounded), `generateOcrText()` (was unbounded), and the per-sub-batch embed + call; multimodal direct fetches get a per-request timeout. Per-touchpoint + defaults: chat 300s, embed/multimodal 60s. `embedQuery` accepts + forwards + `abortSignal`. +- **`src/core/postgres-engine.ts`:** `reconnect()` branches on connection style. + Module-singleton engines re-establish idempotently via `db.connect()` + + refresh the ConnectionManager read pool, never `db.disconnect()` (no null + window). Instance pools keep teardown+recreate. +- **`src/core/search/hybrid.ts`:** one shared `QueryEmbedDeadline` (default 6s, + `GBRAIN_QUERY_EMBED_TIMEOUT_MS`) threaded into both the cache-lookup embed and + the inner embed via `embedQueryBounded` (abortSignal aborts the socket; + `Promise.race` guarantees the await rejects even if the provider ignores the + abort) → the existing keyword fallback engages. +- **Tests:** `test/core/background-work.test.ts`, `test/search/query-embed-deadline.test.ts`, + `test/eval-capture-drain.test.ts`, a `gbrain capture` exit-cleanly case in + `test/e2e/pglite-cli-exit.serial.test.ts`, the `#1745` reconnect E2E in + `test/e2e/postgres-reconnect-singleton.test.ts`, and updated structural + assertions in `test/fix-wave-structural.test.ts`. + +#### Deferred (follow-ups, not in this release) +- Decouple the op-dispatch force-exit timer so it wraps `disconnect()` only and + fix its misleading message. +- Convert `runSync`'s ~20 internal `process.exit` sites to `exitCode + return` + for graceful drain on sync error exits (today they avoid the hang by skipping + disconnect; worst case is a transient PGLite stale-lock that self-heals). +- A gateway-level idle-timeout (vs absolute) for streaming chat. ## [0.42.19.0] - 2026-06-02 **`gbrain skillopt --write-capture` rollouts now get full tool schemas, closing the last gap in the AI SDK v6 tool-loop fix.** The v6 fix that got agent loops working again on non-Anthropic providers (DeepSeek, Qwen, Groq, local models) shipped in v0.42.11.0 — but it fixed only one of the two places skillopt builds tool definitions. The `--write-capture` path (the virtual put_page/submit_job/file_upload registry the optimizer uses to test write-flavored skills) still handed the model a stripped-down schema with `enum`, `default`, and `items` dropped, so the optimizer couldn't see a tool's allowed values and proposed invalid calls. Both builders now use the same shared mapper. diff --git a/TODOS.md b/TODOS.md index 36889831f..031a07b23 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1522,25 +1522,34 @@ Three items deferred: mutex, or document the constraint and assert single-flight at the call site. -- [ ] **Retrofit `awaitPendingSearchCacheWrites` with the same bounded - timeout v0.41.8.0 added to `awaitPendingLastRetrievedWrites`.** The - v0.36.1.x #1090 fix at `src/core/search/hybrid.ts:36-45` shipped the - drain pattern without a timeout; v0.41.8.0 added the timeout + warn - pattern to the new `awaitPendingLastRetrievedWrites` helper. For - symmetry (and to close the same future-failure mode in the cache - drain), apply the same `Promise.race` + stderr warn pattern. ~15 LOC - + 2 unit cases. Pair this with the drain-helper extraction below. +- [x] **Retrofit `awaitPendingSearchCacheWrites` with a bounded timeout.** + DONE in v0.42.20.0 (#1762 reliability wave): `awaitPendingSearchCacheWrites` + is now bounded (`Promise.race` + leftover count), matching + `awaitPendingLastRetrievedWrites`. -- [ ] **Extract a shared `createDrainHelper()` factory when a third - fire-and-forget surface appears.** Per D4 in the v0.41.8.0 eng - review: two surfaces is the threshold for noticing, three for - extracting. `src/core/search/hybrid.ts:awaitPendingSearchCacheWrites` - + `src/core/last-retrieved.ts:awaitPendingLastRetrievedWrites` are - the two surfaces today. When a third surface is added (or when the - timeout-symmetry retrofit above lands and the duplication becomes - load-bearing), extract a `src/core/drain-helper.ts` factory consumed - by both call sites. Pair with the symmetry retrofit so they fire - together as one focused refactor. +- [x] **Extract a shared drain abstraction once a third fire-and-forget surface + appears.** DONE in v0.42.20.0: rule-of-four was met (last-retrieved, facts, + search-cache, eval-capture), so `src/core/background-work.ts` (a registry, not + a per-surface factory) is the single drain owner; each sink registers a + drainer and CLI exit calls `drainAllBackgroundWorkForCliExit`. + +- [ ] **(v0.42.20.0 follow-up) Convert `runSync`'s ~20 internal `process.exit` + sites to `exitCode + return`.** Today those error/cost-gate paths skip the + background-work drain + graceful disconnect (they avoid the #1762 hang by + skipping disconnect entirely; worst case is a transient PGLite stale-lock that + self-heals via stale-reclaim). The common sync SUCCESS path already drains via + handleCliOnly's finally. Convert for graceful drain on sync error exits. + +- [ ] **(v0.42.20.0 follow-up) Decouple the op-dispatch force-exit timer** so it + wraps `engine.disconnect()` only (it's armed before the handler today, doubling + as a blanket handler watchdog) and fix its misleading "engine.disconnect() did + not return…" message that fires even when the handler (not disconnect) was slow. + +- [ ] **(v0.42.20.0 follow-up) Gateway idle-timeout (vs absolute) for streaming + chat.** `withDefaultTimeout` uses an absolute `AbortSignal.timeout`; a streaming + generation actively producing tokens past the chat default (300s) would abort. + Non-streaming `generateText` makes this low-risk today; revisit if a real + long-stream caller trips it. --- ## v0.41 Eval-loop wave follow-ups (v0.42+) diff --git a/VERSION b/VERSION index 77b005fd1..29ac5e954 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.19.0 +0.42.20.0 \ No newline at end of file diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index a0aeb9753..368afeac6 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -11,6 +11,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/operations.ts` — Contract-first operation definitions (the foundation). Exports upload validators `validateUploadPath`, `validatePageSlug`, `validateFilename`, plus `matchesSlugAllowList(slug, prefixes)` (glob matcher: `/*` matches recursive children; bare `` matches exact only). `OperationContext.remote` is a REQUIRED field flagging untrusted callers; `OperationContext.allowedSlugPrefixes` is the trusted-workspace allow-list set by the dream cycle; `OperationContext.auth?: AuthInfo` is threaded through HTTP dispatch for scope enforcement in `serve-http.ts` before the op runs. `put_page` enforces: when `viaSubagent` and `allowedSlugPrefixes` is set, slug must match the allow-list; else the legacy `wiki/agents//...` namespace check applies. Auto-link skipped only when `remote=true && !trustedWorkspace`. Every `Operation` carries `scope?: 'read' | 'write' | 'admin'` + `localOnly?: boolean`; `sync_brain`, `file_upload`, `file_list`, `file_url` are `admin + localOnly` (rejected over HTTP). Four trust-boundary call sites (`put_page` allowlist, file_upload trust-narrowing, submit_job protected-name guard, auto-link skip) use FAIL-CLOSED semantics: `ctx.remote === false` for trusted-only sites, `ctx.remote !== false` for "untrust unless explicit-false" — anything not strictly `false` is treated as remote (closes the HTTP MCP shell-job RCE where a read+write OAuth token could submit `shell` jobs). `sourceScopeOpts(ctx)` encodes the source-scoped read precedence ladder — federated array (`ctx.auth.allowedSources`) wins over scalar (`ctx.sourceId`/`ctx.auth.sourceId`) over nothing; every read-side op handler routes through it so a source-bound OAuth client can't see neighboring sources via `search`/`query`/`list_pages`/`get_page`/`find_experts`/query's image path. `put_page`'s inline disk write-through is the shared `writePageThrough` helper (`src/core/write-through.ts`), ATOMIC via temp-sibling + rename so a crash or concurrent `gbrain sync` can't read a half-written `.md`; same helper backs `gbrain brainstorm/lsd --save`. - `src/core/engine.ts` — Pluggable engine interface (BrainEngine). `clampSearchLimit(limit, default, cap)` takes an explicit cap so per-operation caps can be tighter than `MAX_SEARCH_LIMIT`. Exports `LinkBatchInput`/`TimelineBatchInput` for the bulk-insert API (`addLinksBatch`/`addTimelineEntriesBatch`). `readonly kind: 'postgres' | 'pglite'` discriminator lets `src/core/migrate.ts` and others branch without `instanceof` + dynamic imports. Methods: `batchLoadEmotionalInputs(slugs?)` (CTE-shaped read with per-table aggregates so page × N tags × M takes never produces N×M rows), `setEmotionalWeightBatch(rows)` (`UPDATE FROM unnest($1::text[],$2::text[],$3::real[])` composite-keyed on `(slug, source_id)`), `getRecentSalience(opts)`, `findAnomalies(opts)`. `PageFilters` has `sort?: 'updated_desc'|'updated_asc'|'created_desc'|'slug'` + `PAGE_SORT_SQL` whitelist consumed by both engines. `listAllPageRefs(): Promise>` ordered by `(source_id, slug)` — cheap cross-source enumeration replacing the `getAllSlugs()→getPage(slug)` N+1 (which silently defaulted to `source_id='default'`); parity across postgres-engine.ts + pglite-engine.ts; Pinned by `test/e2e/multi-source-bug-class.test.ts`. `SearchOpts`+`PageFilters` add `sourceIds?: string[]` (federated read axis; both engines apply `WHERE source_id = ANY($N::text[])` when set, preserve scalar `sourceId` fast path when unset); `traverseGraph(slug, depth, opts?)` and `traversePaths(slug, opts?)` accept `opts.sourceId`/`opts.sourceIds`. `traverseGraph` opts has `frontierCap?: number` (per-iteration recursive-CTE cap, approx per-BFS-layer); return type `Promise` for MCP wire stability; export `TraverseGraphOpts`; Postgres uses parenthesized `LIMIT N ORDER BY (slug, id)` inside the recursive term, PGLite mirrors with positional params; Pinned by `test/regressions/v0_36_frontier_cap.test.ts`. Phantom-redirect methods: `refreshPageBody(slug, sourceId, compiled_truth, timeline, content_hash)` narrow-UPDATEs three columns + updated_at, skipping soft-deleted rows (content_hash refresh required so `gbrain sync` sees the canonical as unchanged after fence merge); `migrateFactsToCanonical(phantomSlug, canonicalSlug, sourceId)` UPDATEs `entity_slug`+`source_markdown_slug` on every active fact row keyed on the phantom, preserving embedding/validUntil/kind/status/source_session/confidence; parity at `test/phantom-redirect-engine-parity.test.ts`. `getAdjacencyBoosts(pageIds): Promise>` powers the per-query graph-signals stage — one SQL query returning inbound-link counts among top-K plus a cross-source count (links from differing `source_id`); `COALESCE(p.source_id,'default')` null safety, `HAVING >= 1`, cross-source CASE-WHEN excludes the target's own source; parity SQL across both engines; `SearchResult` gains optional `base_score`, `backlink_boost`, `salience_boost`, `recency_boost`, `exact_match_boost`, `graph_adjacency_boost`, `graph_cross_source_boost`, `session_demote_factor`, `reranker_delta` + internal staging fields; Pinned by `test/e2e/graph-signals-engine.test.ts`. Two REQUIRED methods: `deletePages(slugs, {sourceId}): Promise` (single-batch primitive returning slugs actually deleted) and `resolveSlugsByPaths(paths, {sourceId}): Promise>` (batch path→slug lookup); `sourceId` REQUIRED on both at the type level (asymmetric with single-row `deletePage` which keeps optional/'default'); both short-circuit on empty input and throw when `> DELETE_BATCH_SIZE`. Embedding-signature stale-detection quartet: `countStaleChunks(opts?)` gains optional `signature?: string` widening the stale predicate from `embedding IS NULL` to ALSO include chunks whose JOINed page `embedding_signature IS NOT NULL AND <> $signature` (NULL signature is GRANDFATHERED, never counted; omit `signature` for the legacy NULL-only count); `sumStaleChunkChars(opts?: {sourceId?, signature?}): Promise` = `SUM(LENGTH(chunk_text))` over stale chunks (same predicate + embed_skip filter + optional sourceId scope), used by `gbrain sync --all` cost preview via `estimateCostFromChars`; `setPageEmbeddingSignature(slug, {sourceId?, signature})` stamps `pages.embedding_signature` after a page's chunks (re)embed, idempotent no-op when page absent; `invalidateStaleSignatureEmbeddings({signature, sourceId?}): Promise` NULLs `embedding`+`embedded_at` on every chunk whose page signature is set AND differs, returning the count, called BEFORE `listStaleChunks` so signature-drift pages flow through the NULL-embedding keyset cursor unchanged (NULL never invalidated). Widens `findOrphanPages(opts?: {sourceId?, sourceIds?})` (candidate-side scoping only; inbound links counted from any source). Pinned by `test/sum-stale-chunk-chars.test.ts`, `test/embedding-signature-stale.test.ts`, `test/e2e/engine-parity.test.ts`. Free-text alias layer: `resolveAliases(aliasNorms, opts?): Promise>>` (READ; maps each normalized alias to declaring `(slug, source_id)` pairs, source-scoped) and `setPageAliases(slug, sourceId, aliasNorms)` (WRITE; replaces the full alias set, delete-then-insert, empty clears, idempotent on the unique triple), called by the `importFromContent` ingest projection and the `reindex --aliases` backfill; parity across both engines, Pinned by `test/search/page-aliases-engine.test.ts`. `searchVector` in both engines injects the shared `buildBestPerPagePoolCte` per-page max-pool so a page surfaces on its strongest chunk. - `src/core/engine-constants.ts` — single source of truth for engine batch-sizing constants. Exports `DELETE_BATCH_SIZE = 500` consumed by both engines' `deletePages` + `resolveSlugsByPaths` and by the sync delete + rename loops. Lives outside `engine.ts` (the interface module) to avoid circular-import worry — bounded per-statement work for predictable lock hold time + write amplification. +- `src/core/background-work.ts` (#1762/#1745/#1775) — process background-work registry: the single owner of "drain every fire-and-forget DB-write sink before the CLI disconnects." `registerBackgroundWorkDrainer({name, order, drain(timeoutMs), abort?})` + `drainAllBackgroundWorkForCliExit({timeoutMs})` over a `Map` (idempotent registration by name; `__registerDrainerForTest` returns an unregister handle). Drains in explicit `(order, name)` order — facts FIRST (order 0) so its abort-path DB `logIngest` runs against the freshest live engine — and AWAITS `abort()` only when `drain()` reports `unfinished>0`. Best-effort per drainer: one sink's failure never blocks the others or the disconnect. FOUR sinks register at module import (rule-of-four): `facts/queue.ts` (order 0; `abort`=`shutdown()` cancels a hung facts:absorb Haiku via internalAbort), `last-retrieved.ts` (order 1), `search/hybrid.ts` (order 2; `awaitPendingSearchCacheWrites` bounded via `Promise.race`), `eval-capture.ts` (order 3; `captureEvalCandidate` self-tracks its promise via `awaitPendingEvalCaptures`). Both CLI-exit paths (`src/cli.ts` op-dispatch finally + `handleCliOnly` finally) call it before `engine.disconnect()` — closing the PGLite busy-loop where `db.close()` raced an in-flight job and pinned the single-writer lock (#1762). CLI-EXIT-ONLY: the facts `shutdown()` abort is permanent process state, never call in a long-lived `gbrain serve`. Companion changes: `src/core/ai/gateway.ts` `withDefaultTimeout(caller, ms)` bounds every outbound AI call (chat 300s, embed+multimodal 60s; env `GBRAIN_AI_{CHAT,EMBED,MULTIMODAL}_TIMEOUT_MS`; composed with caller signals via `AbortSignal.any`) and the op-dispatch + handleCliOnly force-exit timers `process.exit(process.exitCode ?? 0)` so a hung disconnect can't mask an errored op as success; `src/core/postgres-engine.ts` `reconnect()` module-mode branch re-establishes via idempotent `db.connect()` + `connectionManager.setReadPool` refresh instead of `db.disconnect()` (no null window for concurrent ops; fail-loud on real connect failure — #1745); `src/core/search/hybrid.ts` `embedQueryBounded` + a shared `QueryEmbedDeadline` (6s, floored 2s per embed via `MIN_QUERY_EMBED_BUDGET_MS`; env `GBRAIN_QUERY_EMBED_TIMEOUT_MS`) bounds the cache-lookup AND inner query embeds so a stalled provider falls back to keyword instead of hanging past the 10s force-exit (#1775). Incorporates + hardens PR #1763 (@ElliotDrel). Pinned by `test/core/background-work.test.ts`, `test/search/query-embed-deadline.test.ts`, `test/eval-capture-drain.test.ts`, `test/e2e/postgres-reconnect-singleton.test.ts`, `test/e2e/pglite-cli-exit.serial.test.ts`, `test/fix-wave-structural.test.ts`. - `src/core/search/graph-signals.ts` — per-query graph-signals helper. `applyGraphSignals(results, engine, opts)` runs as the 4th post-fusion stage (after backlink/salience/recency). Three boosts: `ADJACENCY_BOOST=1.05` (page linked from 2+ OTHER top-K results — local hub for THIS query), `CROSS_SOURCE_BOOST=1.10` (page linked from 2+ DIFFERENT sources — corroborated across team brains, dormant in single-source brains), `SESSION_DEMOTE=0.95` (3+ results from same chat session — keep the highest-scoring at full score, demote the rest). All three inherit the floor-ratio gate preventing weak pages from being boosted past strong ones via popularity. `computeScoreDistribution(results)` emits min/p25/p50/p75/p95/max + `reorder_band_width`. `sessionPrefix(slug)` extracts the chat-session anchor (`chat/2026-05-15-...`). Pure `pairedBootstrapPValue(deltas, resamples, rng)` exported for eval gates. Test seam via `adjacencyFn` DI. Fail-open: any error logs via `logGraphSignalsFailure` (JSONL audit via `audit-writer`) and returns the input array unchanged. Pinned by `test/search/graph-signals.test.ts` (incl. the IRON-RULE floor-gate regression). - `src/core/search/hybrid.ts` extension — `runPostFusionStages` has a 4th stage (`graphSignalsEnabled`, `onGraphMeta`, `onScoreDistribution`). `base_score` stamped at function entry idempotently (captured ONCE before any boost stage mutates `score`). Each post-fusion stage stamps its multiplier: `applyBacklinkBoost`→`backlink_boost`, `applySalienceBoost`→`salience_boost`, `applyRecencyBoost`→`recency_boost`. `applyReranker` (earlier in the pipeline) stamps `reranker_delta` as a rank delta (positive = improved). `applyExactMatchBoost` in `src/core/search/intent-weights.ts` stamps `exact_match_boost` when fired. Per-stage attribution powers `gbrain search --explain` — every boost surface carries its own field so `formatResultsExplain` reads them all without coupling to internal stage ordering. - `src/core/search/explain-formatter.ts` — renders `SearchResult[]` as a multi-line per-result breakdown for `gbrain search --explain`. Reads every boost-stamping field. Handles the "no boosts applied" empty path. 4-decimal precision with trailing-zero strip. Pinned by `test/search/explain-formatter.test.ts`. diff --git a/package.json b/package.json index 69317023a..69fc66f43 100644 --- a/package.json +++ b/package.json @@ -143,5 +143,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.19.0" + "version": "0.42.20.0" } diff --git a/src/cli.ts b/src/cli.ts index c9bfe10cf..3347566cc 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -25,7 +25,7 @@ import type { AIGatewayConfig } from './core/ai/types.ts'; import type { BrainEngine } from './core/engine.ts'; import { operations, OperationError } from './core/operations.ts'; import type { Operation, OperationContext } from './core/operations.ts'; -import { awaitPendingLastRetrievedWrites, type DrainOutcome } from './core/last-retrieved.ts'; +import { drainAllBackgroundWorkForCliExit } from './core/background-work.ts'; import { shouldForceExitAfterMain } from './core/cli-force-exit.ts'; import { serializeMarkdown } from './core/markdown.ts'; import { parseGlobalFlags, setCliOptions, getCliOptions } from './core/cli-options.ts'; @@ -351,7 +351,10 @@ async function main() { console.warn( `[cli] engine.disconnect() did not return within ${DISCONNECT_HARD_DEADLINE_MS}ms — force-exiting`, ); - process.exit(0); + // v0.42.20.0 (codex): honor an exit code an errored op already set — + // a bare process.exit(0) here would mask a failed op as success if the + // drain/disconnect then hangs. + process.exit(process.exitCode ?? 0); }, DISCONNECT_HARD_DEADLINE_MS); // unref so the timer itself doesn't keep the event loop alive — only // the actual pending work (PGLite WASM handle) does. Without unref, @@ -359,7 +362,6 @@ async function main() { forceExitTimer.unref?.(); } - let drainResult: DrainOutcome = { outcome: 'drained', pending: 0 }; try { const ctx = await makeContext(engine, params); const rawResult = await op.handler(ctx, params); @@ -370,55 +372,32 @@ async function main() { const result = JSON.parse(JSON.stringify(rawResult)); const output = formatResult(op.name, result); if (output) process.stdout.write(output); - if (op.name === 'query') { - const { awaitPendingSearchCacheWrites } = await import('./core/search/hybrid.ts'); - await awaitPendingSearchCacheWrites(); - } - // Drain unconditionally for every op — empty-set fast-path is a - // few microseconds. Not per-op-name gated: that was the original - // PR #1259 mistake that left search and get_page exposed. - drainResult = await awaitPendingLastRetrievedWrites(); } catch (e: unknown) { - // C9 fix: drain BEFORE process.exit so a successful op that throws - // during stdout/format still gets its bumpLastRetrievedAt UPDATE - // a chance to commit. Bounded by the drain's own 5s timeout; the - // outer hard-exit timer above bounds the disconnect path. - try { await awaitPendingLastRetrievedWrites(); } catch { /* best-effort */ } + // v0.42.20.0 (codex D4): on error, set exitCode + return so the `finally` + // STILL runs (drains every background-work sink + disconnects). A bare + // process.exit(1) here would skip the finally → skip the drain + disconnect + // (leaves facts/cache/eval-capture writes racing teardown). The finally's + // drain bounds teardown; the outer hard-deadline timer bounds a hung one. if (e instanceof OperationError) { console.error(`Error [${e.code}]: ${e.message}`); if (e.suggestion) console.error(` Fix: ${e.suggestion}`); - process.exit(1); + } else { + console.error(e instanceof Error ? e.message : String(e)); } - console.error(e instanceof Error ? e.message : String(e)); - process.exit(1); + process.exitCode = 1; } finally { - // v0.41.25.0 (#1570) — drain the facts:absorb queue BEFORE disconnect - // so the fire-and-forget queue worker has a live engine to write its - // log against. Closes the bug class that absorb-log.ts:87-100 names: - // facts subsystem holds an engine reference past CLI exit, fires its - // post-completion log against a dead singleton, surfaces as a 'No - // database connection' stderr line on every `gbrain capture`. - // - // 1s timeout is per codex finding 10 from the v0.41.25 plan review: - // ops that don't enqueue facts (most read paths) pay only the - // 0-pending fast-path cost (~microseconds). Capture / import / sync - // that DO enqueue pay up to 1s while in-flight Haiku calls finish. - // Lazy-import keeps this off the hot path for ops that never touch - // the facts queue at all. - try { - const { getFactsQueue } = await import('./core/facts/queue.ts'); - await getFactsQueue().drainPending({ timeout: 1000 }); - } catch { /* best-effort; never block disconnect on drain failure */ } + // v0.42.20.0 — drain ALL fire-and-forget sinks (facts, last-retrieved, + // search-cache, eval-capture) via the background-work registry BEFORE + // disconnect, so a PGLite db.close() can't race in-flight work into the + // re-pump busy-loop (#1762). facts drains first (order 0) so its abort-path + // DB logIngest gets the freshest live-engine window. 1s per-sink timeout: + // read paths with no pending work pay the ~0ms fast path; capture/import + // that DO enqueue pay up to 1s (+ facts shutdown grace) while in-flight + // Haiku finishes. The unref'd hard-deadline timer above is the backstop if + // disconnect or a lingering socket keeps Bun's loop alive. + await drainAllBackgroundWorkForCliExit({ timeoutMs: 1000 }); await engine.disconnect(); if (forceExitTimer) clearTimeout(forceExitTimer); - // Narrow force-exit: only when the drain timed out AND we are NOT - // running a daemon. The drain helper already stderr-warned with the - // pending count, so the diagnostic signal is preserved. Without - // this guard a hung underlying promise can still keep Bun's loop - // alive past disconnect — Codex outside-voice finding #1. - if (drainResult.outcome === 'timeout' && shouldForceExitAfterMain()) { - process.exit(0); - } } } @@ -1909,7 +1888,30 @@ async function handleCliOnly(command: string, args: string[]) { } } finally { syncWatchdog?.dispose(); // #1633: tear down the hard-deadline watchdog on clean exit - if (command !== 'serve') await engine.disconnect(); + // v0.42.20.0 (#1762) — the CLI_ONLY path (which owns `gbrain capture`) + // lacked the op-dispatch drain-before-disconnect contract. `put_page` fires + // a fire-and-forget facts:absorb job AFTER printing the receipt; on a + // multi-chunk page that job is in flight when this finally tears the engine + // down, and `engine.disconnect()` nulling PGLite's _db mid-job spins + // db.close() into a 100%-CPU busy-loop that pins the single-writer lock. + // Drain every background-work sink first (facts shutdown() abort cancels a + // hung Haiku), THEN disconnect. The drain-before-disconnect is the causal + // fix; the force-exit defense below is secondary (it CANNOT preempt a WASM + // busy-loop on a pinned JS thread — that's exactly why the drain matters). + if (command !== 'serve') { + const forceExit = shouldForceExitAfterMain(); + let hardExitTimer: ReturnType | undefined; + if (forceExit) { + hardExitTimer = setTimeout(() => { + console.warn('[cli] engine.disconnect() did not return within 10000ms — force-exiting'); + process.exit(process.exitCode ?? 0); + }, 10_000); + hardExitTimer.unref?.(); + } + await drainAllBackgroundWorkForCliExit(); + await engine.disconnect(); + if (hardExitTimer) clearTimeout(hardExitTimer); + } } } diff --git a/src/core/ai/gateway.ts b/src/core/ai/gateway.ts index 2dd2f5a94..21c5ce57a 100644 --- a/src/core/ai/gateway.ts +++ b/src/core/ai/gateway.ts @@ -53,6 +53,42 @@ import { hasAnthropicKey } from './anthropic-key.ts'; import { AIConfigError, AITransientError, normalizeAIError } from './errors.ts'; import { runGuardrails, hasGuardrails, type GuardrailHook } from '../guardrails.ts'; +// ---- Gateway-wide AI-HTTP timeout (v0.42.20.0, #1762/#1775) ---- +// +// Plain `fetch` (Bun/Node) has NO default request timeout, so a stalled provider +// socket makes an `await` never settle — which hangs `gbrain capture`/`search` +// and, on PGLite, pins the single-writer lock. The AI SDK's `maxRetries` only +// fires on a SETTLED error; a half-open socket never settles. So we bound at the +// SDK CALL layer: default an `abortSignal` into every generateText / generateObject +// / embed call. This (1) covers EVERY provider — including `native-anthropic` +// (the default chat model + the facts:absorb Haiku), which the AI SDK forwards +// the signal to as `fetch(url, {signal})`; and (2) bounds the WHOLE call incl. +// internal retries, not one attempt. Direct-`fetch` paths (multimodal) get the +// signal explicitly. Rerank is already bounded by its recipe `default_timeout_ms`. +function resolveAiTimeoutMs(envVar: string, fallback: number): number { + const raw = process.env[envVar]; + if (raw === undefined) return fallback; + const n = Number(raw); + return Number.isFinite(n) && n > 0 ? n : fallback; +} +/** chat / expansion / OCR — generous; only catches true hangs (non-streaming generateText). */ +const AI_CHAT_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_CHAT_TIMEOUT_MS', 300_000); +/** embed sub-batch (per SDK call, NOT per whole import). */ +const AI_EMBED_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_EMBED_TIMEOUT_MS', 60_000); +/** multimodal per request. */ +const AI_MULTIMODAL_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_MULTIMODAL_TIMEOUT_MS', 60_000); + +/** + * Compose a caller signal with a default wall-clock timeout. When the caller + * supplies its own (Fix 3's 6s query deadline, the facts queue's shutdown abort, + * a budget signal), `AbortSignal.any` makes whichever fires FIRST win — so a + * shorter caller deadline always takes precedence over the default backstop. + */ +function withDefaultTimeout(caller: AbortSignal | undefined, timeoutMs: number): AbortSignal { + const timeout = AbortSignal.timeout(timeoutMs); + return caller ? AbortSignal.any([caller, timeout]) : timeout; +} + const MAX_CHARS = 8000; // v0.36.0.0 (D3 + D4): ZeroEntropy zembed-1 at 1280d via Matryoshka is the // new default for embedding. Real-corpus benchmark across 20 queries: @@ -1440,10 +1476,11 @@ async function embedSubBatch( model, values: texts, providerOptions: providerOpts, - // v0.33.4: caller-supplied abortSignal + maxRetries passthrough. - // Undefined fields are ignored by the AI SDK so the call shape stays - // identical for production callers that don't opt in. - ...(opts?.abortSignal !== undefined && { abortSignal: opts.abortSignal }), + // v0.42.20.0 — default a per-SUB-BATCH embed timeout (codex #3: bounding + // once at embed() top would cap a whole multi-batch import; this is the + // per-SDK-call scope). Composes with a caller signal (Fix 3's 6s query + // deadline) — shorter wins. + abortSignal: withDefaultTimeout(opts?.abortSignal, AI_EMBED_TIMEOUT_MS), ...(opts?.maxRetries !== undefined && { maxRetries: opts.maxRetries }), }); @@ -1503,12 +1540,16 @@ export async function embedOne(text: string): Promise { */ export async function embedQuery( text: string, - opts?: { embeddingModel?: string; dimensions?: number }, + opts?: { embeddingModel?: string; dimensions?: number; abortSignal?: AbortSignal }, ): Promise { const [v] = await embed([text], { inputType: 'query', embeddingModel: opts?.embeddingModel, dimensions: opts?.dimensions, + // v0.42.20.0 (Fix 3) — forward a caller deadline so the query-time embed + // can be bounded BELOW the CLI force-exit; composes with the gateway embed + // default via withDefaultTimeout (shorter wins). + abortSignal: opts?.abortSignal, }); return v; } @@ -1642,6 +1683,9 @@ export async function embedMultimodal( Authorization: `Bearer ${apiKey}`, }, body: JSON.stringify(body), + // v0.42.20.0 (codex #4) — per-request multimodal timeout (direct fetch + // bypasses the SDK abortSignal). + signal: AbortSignal.timeout(AI_MULTIMODAL_TIMEOUT_MS), }); } catch (err) { throw normalizeAIError(err, `embedMultimodal(${recipe.id}:${parsed.modelId})`); @@ -1784,6 +1828,8 @@ async function embedMultimodalOpenAICompat( [authResult.headerName]: authResult.token, }, body: JSON.stringify(body), + // v0.42.20.0 (codex #4) — per-request multimodal timeout (direct fetch). + signal: AbortSignal.timeout(AI_MULTIMODAL_TIMEOUT_MS), }); } catch (err) { throw normalizeAIError(err, `embedMultimodal(${recipe.id}:${modelId})`); @@ -2034,6 +2080,9 @@ export async function expand(query: string): Promise { const result = await generateObject({ model, schema: ExpansionSchema, + // v0.42.20.0 (codex P0) — expansion had NO abortSignal; same stalled-socket + // class as chat. Default the chat timeout. + abortSignal: withDefaultTimeout(undefined, AI_CHAT_TIMEOUT_MS), prompt: [ 'Rewrite the search query below into 3-4 different, related queries that would help find relevant documents.', 'Return ONLY the JSON object. Do NOT include the original query in the result.', @@ -2084,6 +2133,8 @@ export async function generateOcrText(imageBytes: Buffer, mime: string): Promise const base64 = imageBytes.toString('base64'); const result = await generateText({ model, + // v0.42.20.0 (codex) — OCR is a 5th unbounded generateText entry point. + abortSignal: withDefaultTimeout(undefined, AI_CHAT_TIMEOUT_MS), messages: [ { role: 'system', @@ -2612,7 +2663,9 @@ export async function chat(opts: ChatOpts): Promise { messages: toModelMessages(opts.messages) as any, tools: opts.tools && opts.tools.length > 0 ? tools : undefined, maxOutputTokens: opts.maxTokens ?? 4096, - abortSignal: opts.abortSignal, + // v0.42.20.0 — default a chat timeout (composes with the caller's signal, + // shorter wins). Covers native-anthropic (the default provider + facts Haiku). + abortSignal: withDefaultTimeout(opts.abortSignal, AI_CHAT_TIMEOUT_MS), providerOptions: Object.keys(providerOptions).length > 0 ? providerOptions : undefined, }); diff --git a/src/core/background-work.ts b/src/core/background-work.ts new file mode 100644 index 000000000..9dc004b3b --- /dev/null +++ b/src/core/background-work.ts @@ -0,0 +1,111 @@ +/** + * v0.42.20.0 (#1762 / #1745 / #1775 reliability wave) — process background-work + * registry. Single source of truth for "drain every fire-and-forget sink before + * the CLI exits / disconnects." + * + * WHY THIS EXISTS (rule-of-four): four independent fire-and-forget sinks each + * write to the DB after an op returns its response — + * - `last-retrieved.ts` UPDATE pages.last_retrieved_at (#1247/#1269/#1290) + * - `facts/queue.ts` facts:absorb Haiku job + logIngest (#1762) + * - `search/hybrid.ts` query_cache write + * - `eval-capture.ts` eval_candidates INSERT + * On PGLite, if `engine.disconnect()` nulls `_db` while one of these is in + * flight, the sink's "not connected" error path re-pumps via queueMicrotask and + * spins `db.close()` into a 100%-CPU busy-loop that pins the single-writer lock + * (the #1762 incident). The fix is to DRAIN every sink before disconnect. A + * registry (not a hand-written N-call helper) makes that structural: a future + * 5th sink that registers is auto-drained, and the drain is invoked from THREE + * exit points (op-dispatch success finally, op-dispatch error catch, CLI_ONLY + * finally) without repeating the sink list at each. + * + * register (at module import) ─┐ + * last-retrieved (order 1) │ + * facts (order 0) ├─► Map + * search-cache (order 2) │ + * eval-capture (order 3) ┘ + * │ CLI exit + * ▼ + * drainAllBackgroundWorkForCliExit ──► sort by (order, name) + * for each: await drain(timeoutMs) + * if unfinished>0 && abort: + * await abort() ◄─ facts shutdown() + * ▼ + * engine.disconnect() (caller) + * + * Registration MUST live in the enqueue-owning module (so "module not imported + * ⇒ no work enqueued ⇒ nothing to drain" holds). The Map is keyed by name so a + * re-import / test mock REPLACES rather than duplicating (an array would + * double-register). + */ + +export interface BackgroundWorkDrainer { + /** Stable identity; also the Map key (idempotent registration). */ + name: string; + /** + * Explicit drain order — lower runs first. Facts is 0 so its abort-path DB + * `logIngest` gets the freshest live-engine window before the fast + * last-retrieved / search-cache drains. Ties break by name for determinism. + */ + order: number; + /** Resolve when in-flight work settles OR the bound elapses; report leftovers. */ + drain(timeoutMs: number): Promise<{ unfinished: number }>; + /** + * Optional hard-stop for stragglers (facts-queue: `shutdown()`). AWAITED by + * the registry so the aborted job's DB write settles against a live engine + * BEFORE the caller disconnects. Only invoked when `drain` reports unfinished. + */ + abort?(): Promise; +} + +const drainers = new Map(); + +/** Register (or replace, by name) a fire-and-forget sink drainer. */ +export function registerBackgroundWorkDrainer(d: BackgroundWorkDrainer): void { + drainers.set(d.name, d); +} + +/** + * Test seam — registers a drainer and returns an unregister handle. Preferred + * over a blunt reset: real sink modules register at import time and won't re-run + * that top-level side effect on a second import, so a global clear would + * silently drop the production drainers for the rest of the test process. + */ +export function __registerDrainerForTest(d: BackgroundWorkDrainer): () => void { + drainers.set(d.name, d); + return () => { drainers.delete(d.name); }; +} + +/** Test seam — snapshot of registered drainer names (sorted), for assertions. */ +export function __listDrainerNamesForTest(): string[] { + return [...drainers.keys()].sort(); +} + +/** + * CLI-EXIT-ONLY. `abort()` is a permanent process-level state change on a sink + * (the facts queue's `shutdown()` sets `shuttingDown=true` for the process + * lifetime). NEVER call this in a long-lived process (`gbrain serve`). Drains + * every registered sink before `engine.disconnect()` so a PGLite `db.close()` + * can't race in-flight work into the re-pump busy-loop (#1762). + * + * Best-effort and non-throwing: one sink's failure never blocks the others or + * the subsequent disconnect. + */ +export async function drainAllBackgroundWorkForCliExit(opts?: { timeoutMs?: number }): Promise { + const timeoutMs = opts?.timeoutMs ?? 2000; + const ordered = [...drainers.values()].sort( + (a, b) => a.order - b.order || a.name.localeCompare(b.name), + ); + for (const d of ordered) { + try { + const { unfinished } = await d.drain(timeoutMs); + if (unfinished > 0 && d.abort) { + // codex #9: AWAIT — the facts:absorb job writes its absorb-log to the + // DB on settle; the abort must finish against a live engine before the + // caller disconnects. + await d.abort(); + } + } catch { + /* best-effort; never block disconnect on one sink's failure */ + } + } +} diff --git a/src/core/embedding.ts b/src/core/embedding.ts index 7fe3c9148..20515d313 100644 --- a/src/core/embedding.ts +++ b/src/core/embedding.ts @@ -53,7 +53,7 @@ export async function embed(text: string): Promise { */ export async function embedQuery( text: string, - opts?: { embeddingModel?: string; dimensions?: number }, + opts?: { embeddingModel?: string; dimensions?: number; abortSignal?: AbortSignal }, ): Promise { return gatewayEmbedQuery(text, opts); } diff --git a/src/core/eval-capture.ts b/src/core/eval-capture.ts index 9983e892b..2326119fb 100644 --- a/src/core/eval-capture.ts +++ b/src/core/eval-capture.ts @@ -43,6 +43,7 @@ import type { } from './types.ts'; import type { GBrainConfig } from './config.ts'; import { scrubPii } from './eval-capture-scrub.ts'; +import { registerBackgroundWorkDrainer } from './background-work.ts'; // HybridSearchMeta is canonical in src/core/types.ts and exported via the // public `gbrain/types` subpath. Surfaced from hybridSearch via the @@ -157,6 +158,21 @@ export async function captureEvalCandidate( engine: BrainEngine, ctx: CaptureContext, opts: { scrub_pii?: boolean } = {}, +): Promise { + // v0.42.20.0 — track the fire-and-forget promise so the background-work + // registry can drain it before CLI disconnect. Callers still `void` this; the + // returned promise is back-compat for any awaiter. The async DB write + // (logEvalCandidate) is the same lock-pin / disconnect-race class as the other + // sinks — on PGLite an undrained capture racing db.close() can wedge. + const p = doCaptureEvalCandidate(engine, ctx, opts); + trackEvalCapture(p); + return p; +} + +async function doCaptureEvalCandidate( + engine: BrainEngine, + ctx: CaptureContext, + opts: { scrub_pii?: boolean } = {}, ): Promise { try { const input = buildEvalCandidateInput(ctx, opts); @@ -174,6 +190,45 @@ export async function captureEvalCandidate( } } +// v0.42.20.0 — bounded drain + registration (mirrors last-retrieved). The 4th +// fire-and-forget DB-write sink (codex caught this one missing from the +// registry). order 3, no abort (bare INSERT, nothing to hard-stop). +const pendingEvalCaptures = new Set>(); + +function trackEvalCapture(promise: Promise): void { + pendingEvalCaptures.add(promise); + promise.finally(() => pendingEvalCaptures.delete(promise)).catch(() => { /* swallow */ }); +} + +export async function awaitPendingEvalCaptures(timeoutMs = 5_000): Promise<{ unfinished: number }> { + if (pendingEvalCaptures.size === 0) return { unfinished: 0 }; + const snapshot = [...pendingEvalCaptures]; + let timer: ReturnType | undefined; + const timeout = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => resolve('timeout'), timeoutMs); + }); + const drain = Promise.allSettled(snapshot).then(() => 'drained' as const); + const outcome = await Promise.race([drain, timeout]); + if (timer) clearTimeout(timer); + if (outcome === 'timeout') { + const unfinished = pendingEvalCaptures.size; + for (const pr of snapshot) pendingEvalCaptures.delete(pr); + return { unfinished }; + } + return { unfinished: 0 }; +} + +/** Test seam — clears the pending eval-capture set. */ +export function _resetPendingEvalCapturesForTests(): void { + pendingEvalCaptures.clear(); +} + +registerBackgroundWorkDrainer({ + name: 'eval-capture', + order: 3, + drain: (ms) => awaitPendingEvalCaptures(ms), +}); + /** * Check whether capture is enabled for this process. * diff --git a/src/core/facts/queue.ts b/src/core/facts/queue.ts index 475863c7a..723a0fc27 100644 --- a/src/core/facts/queue.ts +++ b/src/core/facts/queue.ts @@ -18,6 +18,8 @@ * concurrency + dropping under load. */ +import { registerBackgroundWorkDrainer } from '../background-work.ts'; + export interface FactsQueueCounters { enqueued: number; completed: number; @@ -253,3 +255,17 @@ export function getFactsQueue(opts?: FactsQueueOpts): FactsQueue { export function __resetFactsQueueForTests(): void { _singleton = null; } + +// v0.42.20.0 — register as a background-work sink (order 0 — drained FIRST so +// its abort-path DB logIngest gets the freshest live-engine window). `abort` = +// shutdown(): sets shuttingDown=true (pump short-circuits) + fires internalAbort +// (the facts:absorb job forwards it to gateway.chat, cancelling a hung Haiku the +// drain-only fix can't). Registry AWAITS the abort so logIngest settles against +// a live engine before disconnect (#1762). `drainPending` itself stays +// non-aborting — the abort is the registry's separate post-drain step. +registerBackgroundWorkDrainer({ + name: 'facts', + order: 0, + drain: (ms) => getFactsQueue().drainPending({ timeout: ms }).then((r) => ({ unfinished: r.unfinished })), + abort: () => getFactsQueue().shutdown(), +}); diff --git a/src/core/last-retrieved.ts b/src/core/last-retrieved.ts index 34ceaa473..1b2a8688f 100644 --- a/src/core/last-retrieved.ts +++ b/src/core/last-retrieved.ts @@ -35,6 +35,7 @@ import type { BrainEngine } from './engine.ts'; import { isUndefinedColumnError } from './utils.ts'; +import { registerBackgroundWorkDrainer } from './background-work.ts'; let _trackRetrievalCache: { ts: number; enabled: boolean } | null = null; const TRACK_RETRIEVAL_CACHE_TTL_MS = 30_000; @@ -125,6 +126,14 @@ export function _peekPendingLastRetrievedWritesForTests(): number { return pendingLastRetrievedWrites.size; } +// v0.42.20.0 — register as a background-work sink (order 1; no abort — bare +// UPDATEs, nothing to hard-stop). Drained before CLI disconnect. +registerBackgroundWorkDrainer({ + name: 'last-retrieved', + order: 1, + drain: (ms) => awaitPendingLastRetrievedWrites(ms).then((r) => ({ unfinished: r.pending })), +}); + /** * Resolve `search.track_retrieval` config with a 30s in-process cache so * hot-path callers don't pay a SELECT per search. Default-on: missing diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 5b50d51e4..0b5cfa35b 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -4781,18 +4781,48 @@ export class PostgresEngine implements BrainEngine { } /** - * Reconnect the engine by tearing down the current pool and creating a fresh one. - * No-ops if no saved config (module-singleton mode) or if already reconnecting. + * Reconnect the engine after a transient connection blip. Branches on + * connection style; no-ops if no saved config or if already reconnecting. * - * v0.42.x (#1685 GAP B): records a pool-recovery audit event so the - * `pool_reap_health` doctor check can answer "reaped N times AND not - * auto-recovering." `ctx.error` (threaded by retry.ts) is classified: a - * CONNECTION_ENDED match is a true pooler reap; anything else (or no error, - * e.g. the supervisor's health-check reconnect) is `reconnect_other`. All - * audit calls are best-effort and never block the reconnect (CODEX #8). + * - MODULE-singleton engines SHARE `db.ts`'s `sql` (#1745). Calling + * `db.disconnect()` here (via `this.disconnect()`) would null it out from + * under EVERY concurrent op (other dream-cycle phases, minion-queue + * `promoteDelayed`), which then throw "connect() has not been called" in the + * disconnect→connect window. postgres.js already auto-replaces dead sockets + * inside its pool, so a transient blip recovers WITHOUT a teardown. Recover + * idempotently instead: `db.connect()` is a no-op when the singleton is alive + * (the common case) and re-establishes it only if some other path nulled it — + * never introducing a null window — then refreshes the ConnectionManager read + * pool. Scope: fixes the singleton-NULL-window bug specifically; it does NOT + * rebuild a genuinely WEDGED-but-live pool (db.connect() no-ops there) — a + * different failure mode postgres.js owns. + * + * - INSTANCE pools (worker engines, `poolSize` set) own their `_sql` — tearing + * it down and rebuilding is correct and isolated; nobody else shares it. This + * path also records a pool-recovery audit event (#1685 GAP B) so the + * `pool_reap_health` doctor check can answer "reaped N times AND not + * auto-recovering." `ctx.error` (threaded by retry.ts) is classified: a + * CONNECTION_ENDED match is a true pooler reap; anything else (or no error, + * e.g. the supervisor's health-check reconnect) is `reconnect_other`. All + * audit calls are best-effort and never block the reconnect (CODEX #8). */ async reconnect(ctx?: { error?: unknown }): Promise { if (!this._savedConfig || this._reconnecting) return; + if (this._connectionStyle !== 'instance') { + // Module-singleton: never tear down the shared pool. db.connect() is + // idempotent (no-op when the singleton is alive — the common #1745 path). + // FAIL-LOUD (codex): do NOT swallow a real connect failure — a swallowed + // error would make reconnect() resolve "successfully" and let the + // supervisor reset its health-failure counter / emit db_reconnected when + // the DB is actually down. A throw propagates as the real cause (matches + // the withRetry+reconnect contract and the instance path's posture). + await db.connect(this._savedConfig); + // If db.connect() RE-CREATED the singleton (another path nulled it), the + // ConnectionManager set at connect-time still points at the ended old + // pool. Refresh it. Idempotent no-op when the singleton was already alive. + this.connectionManager?.setReadPool(db.getConnection()); + return; + } this._reconnecting = true; let isReap = false; @@ -4808,9 +4838,8 @@ export class PostgresEngine implements BrainEngine { } catch { /* audit is best-effort */ } try { - // Tear down old pool (best-effort — it may already be dead) + // Instance pool: tear down old pool (best-effort — it may already be dead). try { await this.disconnect(); } catch { /* swallow */ } - // Create fresh pool await this.connect(this._savedConfig); try { const { logPoolRecovery } = await import('./audit/pool-recovery-audit.ts'); diff --git a/src/core/search/hybrid.ts b/src/core/search/hybrid.ts index a65a06319..179bdfb95 100644 --- a/src/core/search/hybrid.ts +++ b/src/core/search/hybrid.ts @@ -13,6 +13,7 @@ import type { BrainEngine } from '../engine.ts'; import { MAX_SEARCH_LIMIT, clampSearchLimit } from '../engine.ts'; import type { SearchResult, SearchOpts, HybridSearchMeta } from '../types.ts'; import { embed, embedQuery } from '../embedding.ts'; +import { registerBackgroundWorkDrainer } from '../background-work.ts'; import { resolveEmbeddingColumn, isCacheSafe } from './embedding-column.ts'; import { resolveAdaptiveReturn, @@ -72,15 +73,52 @@ export async function stampContentFlags(engine: BrainEngine, results: SearchResu } } -export async function awaitPendingSearchCacheWrites(): Promise { - if (pendingCacheWrites.size === 0) return; - await Promise.allSettled([...pendingCacheWrites]); +/** + * v0.42.20.0 — bounded drain (was an unbounded `Promise.allSettled`, codex + * confirmed; TODOS retrofit). Mirrors `awaitPendingLastRetrievedWrites`: races + * the in-flight cache writes against a timeout and reports leftovers so the + * background-work registry can move on to disconnect instead of hanging on a + * wedged cache write. Drops the timed-out snapshot's references so a long-lived + * process doesn't accumulate forever-pending ghosts. + */ +export async function awaitPendingSearchCacheWrites( + timeoutMs = 5_000, +): Promise<{ unfinished: number }> { + if (pendingCacheWrites.size === 0) return { unfinished: 0 }; + const snapshot = [...pendingCacheWrites]; + let timer: ReturnType | undefined; + const timeout = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => resolve('timeout'), timeoutMs); + }); + const drain = Promise.allSettled(snapshot).then(() => 'drained' as const); + const outcome = await Promise.race([drain, timeout]); + if (timer) clearTimeout(timer); + if (outcome === 'timeout') { + const unfinished = pendingCacheWrites.size; + for (const p of snapshot) pendingCacheWrites.delete(p); + return { unfinished }; + } + return { unfinished: 0 }; +} + +/** Test seam — clears the pending cache-write set so each test starts clean. */ +export function _resetPendingSearchCacheWritesForTests(): void { + pendingCacheWrites.clear(); } function trackCacheWrite(promise: Promise): void { pendingCacheWrites.add(promise); promise.finally(() => pendingCacheWrites.delete(promise)).catch(() => { /* swallow */ }); } + +// v0.42.20.0 — register as a background-work sink (order 2; no abort — bare +// cache INSERTs). Drained before CLI disconnect, for BOTH search and query +// (previously only `query` drained it, and unbounded). +registerBackgroundWorkDrainer({ + name: 'search-cache', + order: 2, + drain: (ms) => awaitPendingSearchCacheWrites(ms), +}); /** * Backlink boost coefficient. Score is multiplied by (1 + BACKLINK_BOOST_COEF * log(1 + count)). * - 0 backlinks: factor = 1.0 (no boost). @@ -652,6 +690,79 @@ export interface HybridSearchOpts extends SearchOpts { * row; everyone else leaves it undefined and pays no cost. */ onMeta?: (meta: HybridSearchMeta) => void; + /** + * v0.42.20.0 (Fix 3, #1775) INTERNAL — shared query-embed deadline threaded + * from `hybridSearchCached` into the inner `hybridSearch` so the cache-lookup + * embed and the inner embed share ONE wall-clock budget (worst case ~one + * timeout, not two). Direct `hybridSearch` callers leave it undefined and get + * a fresh per-call deadline. Not part of the public contract. + */ + _queryEmbedDeadline?: QueryEmbedDeadline; +} + +/** + * v0.42.20.0 (Fix 3, #1775) — bound the query-time embed so a stalled provider + * (the user's zeroentropy case) fails over to keyword instead of hanging past + * the CLI's 10s force-exit. Default 6s leaves headroom under that deadline. + */ +const QUERY_EMBED_TIMEOUT_MS = (() => { + const n = Number(process.env.GBRAIN_QUERY_EMBED_TIMEOUT_MS); + return Number.isFinite(n) && n > 0 ? n : 6_000; +})(); + +/** + * Floor for the remaining shared-deadline budget at each embed call (codex). + * The shared deadline is absolute from `hybridSearchCached` entry, so slow + * expansion/keyword (or a 6s cache-lookup stall) before the inner embed could + * leave ~0 budget and starve a HEALTHY embed into a false keyword-only result. + * Flooring guarantees every embed gets at least this long, so a fast healthy + * embed (~0.5s) always succeeds. Worst case under a stalled provider on the + * cache-miss path: cache-lookup (6s) + inner floor (2s) = 8s, still under the + * 10s CLI force-exit. + */ +const MIN_QUERY_EMBED_BUDGET_MS = 2_000; + +export interface QueryEmbedDeadline { + /** Aborts the underlying fetch (clean socket close) when the budget elapses. */ + signal: AbortSignal; + /** Absolute wall-clock deadline (ms epoch) — shared so a second embed sees the elapsed budget. */ + deadlineAt: number; +} + +export function makeQueryEmbedDeadline(ms = QUERY_EMBED_TIMEOUT_MS): QueryEmbedDeadline { + return { signal: AbortSignal.timeout(ms), deadlineAt: Date.now() + ms }; +} + +/** + * Embed a query bounded by the shared deadline. Two layers: (1) `abortSignal` + * aborts the fetch so the socket closes and the process can exit clean; (2) a + * `Promise.race` against the REMAINING budget GUARANTEES the await rejects even + * if a wedged provider ignores the abort. On rejection the caller's existing + * try/catch falls back to keyword. The losing embed promise's late rejection is + * swallowed so it never surfaces as an unhandledRejection. + */ +export async function embedQueryBounded( + text: string, + embedOpts: { embeddingModel?: string; dimensions?: number } | undefined, + dl: QueryEmbedDeadline, +): Promise { + const p = embedQuery(text, { ...(embedOpts ?? {}), abortSignal: dl.signal }); + p.catch(() => { /* swallow the loser's late rejection */ }); + // Floor the budget so a healthy embed isn't starved when the shared absolute + // deadline was mostly consumed by prior work (codex). Still bounded overall. + const remaining = Math.max(MIN_QUERY_EMBED_BUDGET_MS, dl.deadlineAt - Date.now()); + let timer: ReturnType | undefined; + const deadline = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`query embed deadline ${QUERY_EMBED_TIMEOUT_MS}ms exceeded`)), + remaining, + ); + }); + try { + return await Promise.race([p, deadline]); + } finally { + if (timer) clearTimeout(timer); + } } export async function hybridSearch( @@ -1063,7 +1174,12 @@ export async function hybridSearch( const embedOpts = resolvedCol.embeddingModel ? { embeddingModel: resolvedCol.embeddingModel, dimensions: resolvedCol.dimensions } : undefined; - const embeddings = await Promise.all(queries.map(q => embedQuery(q, embedOpts))); + // v0.42.20.0 (Fix 3) — bound the query embed. Reuse the shared deadline + // threaded from hybridSearchCached (so the cache-lookup embed + this one + // share one ~6s budget); direct callers get a fresh deadline. On timeout + // the embed throws → the catch below falls back to keyword-only. + const embedDl = opts?._queryEmbedDeadline ?? makeQueryEmbedDeadline(); + const embeddings = await Promise.all(queries.map(q => embedQueryBounded(q, embedOpts, embedDl))); queryEmbedding = embeddings[0]; const textLists = await Promise.all( embeddings.map(emb => engine.searchVector(emb, searchOpts)), @@ -1459,6 +1575,13 @@ export async function hybridSearchCached( // attempt it when the cache is enabled AND the gateway has an embedding // provider configured. let queryEmbedding: Float32Array | null = null; + // v0.42.20.0 (Fix 3, #1775) — ONE shared query-embed deadline for the + // cache-lookup embed below AND the inner hybridSearch embed (threaded via + // opts._queryEmbedDeadline). On a stalled provider the cache-lookup embed + // times out (→ cacheStatus 'disabled', fall through), then the inner embed + // sees the already-elapsed budget and fails fast → keyword fallback. Worst + // case ~one timeout (~6s), comfortably under the CLI 10s force-exit. + const queryEmbedDl = makeQueryEmbedDeadline(); if (!skipCache) { try { const { isAvailable } = await import('../ai/gateway.ts'); @@ -1470,7 +1593,10 @@ export async function hybridSearchCached( const providerProbeCached = resolvedColCached.embeddingModel || undefined; if (isAvailable('embedding', providerProbeCached)) { // v0.35.0.0+: query-side embedding (cache lookup path). - queryEmbedding = await embedQuery(query); + // v0.42.20.0 (Fix 3) — bounded by the shared deadline; on timeout this + // throws → caught below → cacheStatus 'disabled' → falls through to the + // inner hybridSearch (which reuses the same elapsed deadline). + queryEmbedding = await embedQueryBounded(query, undefined, queryEmbedDl); } else { cacheStatus = 'disabled'; } @@ -1533,6 +1659,9 @@ export async function hybridSearchCached( const userOnMeta = opts?.onMeta; const results = await hybridSearch(engine, query, { ...opts, + // v0.42.20.0 (Fix 3) — share the query-embed deadline so the inner embed + // doesn't start a fresh 6s budget after the cache-lookup already spent it. + _queryEmbedDeadline: queryEmbedDl, onMeta: (m) => { innerMetaBox.current = m; // Do NOT call userOnMeta here — we'll emit a merged meta below diff --git a/test/core/background-work.test.ts b/test/core/background-work.test.ts new file mode 100644 index 000000000..6b664e73d --- /dev/null +++ b/test/core/background-work.test.ts @@ -0,0 +1,136 @@ +/** + * v0.42.20.0 (#1762) — background-work registry unit tests. + * + * Pure, no DB / no engine: drives the registry via the __registerDrainerForTest + * seam. Pins the contracts the reliability wave depends on: + * - drains in explicit (order, name) order — facts (0) first + * - a drainer reporting unfinished>0 has its abort() AWAITED + * - a throwing drainer doesn't block the others + * - empty registry is a fast no-op + * - Map idempotency: re-registering the same name REPLACES (no duplicate) + * - the unregister handle removes it + */ +import { describe, test, expect } from 'bun:test'; +import { + drainAllBackgroundWorkForCliExit, + __registerDrainerForTest, + __listDrainerNamesForTest, + type BackgroundWorkDrainer, +} from '../../src/core/background-work.ts'; + +function makeRecorder() { + const calls: string[] = []; + return { calls }; +} + +describe('background-work registry', () => { + test('drains in explicit (order, name) order; facts-order-0 first', async () => { + const { calls } = makeRecorder(); + const mk = (name: string, order: number): [BackgroundWorkDrainer, () => void] => { + const d: BackgroundWorkDrainer = { + name, + order, + drain: async () => { calls.push(`drain:${name}`); return { unfinished: 0 }; }, + }; + return [d, __registerDrainerForTest(d)]; + }; + const [, u1] = mk('zzz', 2); + const [, u0] = mk('facts', 0); + const [, u15] = mk('mid', 1); + try { + await drainAllBackgroundWorkForCliExit({ timeoutMs: 50 }); + // Only assert ordering among the ones we registered (production sinks may + // also be registered when their modules were imported). + const ours = calls.filter((c) => ['drain:facts', 'drain:mid', 'drain:zzz'].includes(c)); + expect(ours).toEqual(['drain:facts', 'drain:mid', 'drain:zzz']); + } finally { + u0(); u15(); u1(); + } + }); + + test('abort() is AWAITED only when unfinished>0', async () => { + const seq: string[] = []; + const withUnfinished: BackgroundWorkDrainer = { + name: 'test-straggler', + order: 0, + drain: async () => { seq.push('drain'); return { unfinished: 3 }; }, + abort: async () => { + await new Promise((r) => setTimeout(r, 5)); + seq.push('abort-done'); + }, + }; + const noUnfinished: BackgroundWorkDrainer = { + name: 'test-clean', + order: 1, + drain: async () => { seq.push('drain-clean'); return { unfinished: 0 }; }, + abort: async () => { seq.push('abort-clean-SHOULD-NOT-RUN'); }, + }; + const u1 = __registerDrainerForTest(withUnfinished); + const u2 = __registerDrainerForTest(noUnfinished); + try { + await drainAllBackgroundWorkForCliExit({ timeoutMs: 50 }); + // straggler: drain then abort (awaited → abort-done present); clean: no abort. + expect(seq).toContain('drain'); + expect(seq).toContain('abort-done'); + expect(seq).toContain('drain-clean'); + expect(seq).not.toContain('abort-clean-SHOULD-NOT-RUN'); + // abort-done comes after its own drain (awaited). + expect(seq.indexOf('abort-done')).toBeGreaterThan(seq.indexOf('drain')); + } finally { + u1(); u2(); + } + }); + + test('a throwing drainer does not block the others', async () => { + const seen: string[] = []; + const boom: BackgroundWorkDrainer = { + name: 'test-boom', + order: 0, + drain: async () => { throw new Error('drain blew up'); }, + }; + const ok: BackgroundWorkDrainer = { + name: 'test-ok-after-boom', + order: 1, + drain: async () => { seen.push('ok'); return { unfinished: 0 }; }, + }; + const u1 = __registerDrainerForTest(boom); + const u2 = __registerDrainerForTest(ok); + try { + // Must not reject despite the throwing drainer. + await drainAllBackgroundWorkForCliExit({ timeoutMs: 50 }); + expect(seen).toContain('ok'); + } finally { + u1(); u2(); + } + }); + + test('Map idempotency: re-registering the same name replaces, no duplicate', () => { + const before = __listDrainerNamesForTest().filter((n) => n === 'test-dup').length; + expect(before).toBe(0); + const d1: BackgroundWorkDrainer = { name: 'test-dup', order: 0, drain: async () => ({ unfinished: 0 }) }; + const d2: BackgroundWorkDrainer = { name: 'test-dup', order: 9, drain: async () => ({ unfinished: 0 }) }; + const u1 = __registerDrainerForTest(d1); + const u2 = __registerDrainerForTest(d2); + try { + const count = __listDrainerNamesForTest().filter((n) => n === 'test-dup').length; + expect(count).toBe(1); // replaced, not duplicated + } finally { + u1(); u2(); + } + }); + + test('unregister handle removes the drainer', () => { + const d: BackgroundWorkDrainer = { name: 'test-unreg', order: 0, drain: async () => ({ unfinished: 0 }) }; + const unreg = __registerDrainerForTest(d); + expect(__listDrainerNamesForTest()).toContain('test-unreg'); + unreg(); + expect(__listDrainerNamesForTest()).not.toContain('test-unreg'); + }); + + test('empty registry (no test drainers) resolves fast', async () => { + // Production sinks may be registered, but they fast-path on empty pending + // sets. This just asserts the call resolves without throwing. + await drainAllBackgroundWorkForCliExit({ timeoutMs: 10 }); + expect(true).toBe(true); + }); +}); diff --git a/test/e2e/pglite-cli-exit.serial.test.ts b/test/e2e/pglite-cli-exit.serial.test.ts index 13851c7be..d7b293aa9 100644 --- a/test/e2e/pglite-cli-exit.serial.test.ts +++ b/test/e2e/pglite-cli-exit.serial.test.ts @@ -227,6 +227,46 @@ describe('v0.41.8.0 — PGLite CLI read commands exit cleanly (#1247/#1269/#1290 }, 30_000); }); +describe('v0.42.20.0 — gbrain capture (CLI_ONLY) exits cleanly + frees the lock (#1762)', () => { + test('multi-chunk capture exits 0 within 25s AND a later command runs lock-free', async () => { + // The #1762 repro: capture on a multi-chunk page enqueues a fire-and-forget + // facts:absorb job, then handleCliOnly's finally disconnects mid-job → + // PGLite db.close() busy-loop pins the single-writer lock. The registry + // drain-before-disconnect (Fix 0.A + Fix 1) closes it. Without an API key + // the facts job is a fast no-op, so this is a wiring regression guard: the + // drain+disconnect path runs and capture exits cleanly + the lock is free. + const body = Array.from({ length: 12 }, (_, i) => + `## Section ${i}\n\nThis is paragraph ${i} of a deliberately long meeting note ` + + `with enough prose to split into multiple chunks. Foxtrot tango whiskey ${i}. ` + + `The recursive chunker targets a few hundred tokens per chunk, so a dozen of ` + + `these sections guarantees a multi-chunk page for the capture path.\n`, + ).join('\n'); + const capFile = join(tmpHome, 'capture-input.md'); + writeFileSync(capFile, `---\ntitle: Capture Meeting\n---\n${body}\n`, 'utf-8'); + + const cap = await runWithTimeout( + ['capture', '--file', capFile, '--slug', 'meetings/capture-test', '--type', 'meeting', '--quiet'], + 25_000, + ); + if (cap.code !== 0) { + // The IRON RULE is "does not hang." A non-zero exit that still returned + // within the window is tolerable in keyless CI; a hang (kill at 25s) is not. + expect(cap.durationMs).toBeLessThan(25_000); + } else { + expect(cap.code).toBe(0); + } + + // The real lock-pin symptom: the NEXT command times out waiting for the + // PGLite lock. Assert a subsequent read runs cleanly and quickly. + const next = await runWithTimeout(['get', 'meetings/capture-test'], 15_000); + expect(next.durationMs).toBeLessThan(15_000); + expect(next.stderr).not.toContain('Timed out waiting for PGLite lock'); + if (next.code === 0) { + expect(next.stdout.toLowerCase()).toContain('foxtrot'); + } + }, 60_000); +}); + describe('v0.41.8.0 — daemon survival (regression guard for narrow force-exit)', () => { test('gbrain serve --http stays alive past the timeout window', async () => { // Pick a likely-free ephemeral port. We're testing "still alive diff --git a/test/e2e/postgres-reconnect-singleton.test.ts b/test/e2e/postgres-reconnect-singleton.test.ts new file mode 100644 index 000000000..128fd91a2 --- /dev/null +++ b/test/e2e/postgres-reconnect-singleton.test.ts @@ -0,0 +1,98 @@ +/** + * E2E for #1745 (v0.42.20.0): module-mode reconnect() must NOT tear down the + * shared module singleton. + * + * The bug: a transient blip triggers withRetry's reconnect callback → + * PostgresEngine.reconnect() → this.disconnect() → (module mode) db.disconnect() + * → `sql.end(); sql = null`. Concurrent ops (other dream-cycle phases, the + * minion-queue promoteDelayed loop) read db.getConnection() during that null + * window and throw "No database connection: connect() has not been called". + * + * The fix: module-mode reconnect() never calls db.disconnect(); it idempotently + * re-establishes via db.connect() (a no-op when the singleton is alive, which is + * the common case) and refreshes the ConnectionManager read pool. postgres.js + * auto-heals dead sockets, so a transient blip recovers without a teardown. + * + * Instance-pool engines (worker / `jobs work`) keep the teardown+recreate path. + */ +import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; +import { PostgresEngine } from '../../src/core/postgres-engine.ts'; +import * as db from '../../src/core/db.ts'; + +const DATABASE_URL = process.env.DATABASE_URL; +const skip = !DATABASE_URL; + +if (skip) { + // eslint-disable-next-line no-console + console.log('Skipping postgres-reconnect-singleton E2E (DATABASE_URL not set)'); +} + +describe.skipIf(skip)('#1745 — module-mode reconnect preserves the shared singleton', () => { + beforeAll(async () => { + await db.disconnect(); + await db.connect({ database_url: DATABASE_URL! }); + }, 30_000); + + afterAll(async () => { + await db.disconnect(); + }); + + test('module reconnect() keeps db.getConnection() live (no null window)', async () => { + const engine = new PostgresEngine(); + await engine.connect({ database_url: DATABASE_URL! }); // module singleton (no poolSize) + + // Capture the singleton reference BEFORE reconnect — pre-fix, reconnect's + // disconnect would sql.end() + null this out. + const sqlBefore = db.getConnection(); + + await engine.reconnect(); + + // Post-fix: the singleton is untouched (db.connect() is a no-op when alive), + // so getConnection() still returns a usable client and queries still work. + const sqlAfter = db.getConnection(); + const rows = await sqlAfter.unsafe('SELECT 1 as ok'); + expect((rows[0] as unknown as { ok: number }).ok).toBe(1); + // Same live pool object (no teardown/recreate in module mode). + expect(sqlAfter).toBe(sqlBefore); + }, 30_000); + + test('concurrent reader during reconnect never sees "connect() has not been called"', async () => { + const engine = new PostgresEngine(); + await engine.connect({ database_url: DATABASE_URL! }); + + // Fire a burst of reads through the module singleton WHILE reconnecting. + // Pre-fix, the disconnect→connect window nulled the singleton and the + // concurrent readers threw. Post-fix there is no null window. + const readers = Array.from({ length: 20 }, async () => { + const r = await db.getConnection().unsafe('SELECT 1 as ok'); + return (r[0] as unknown as { ok: number }).ok; + }); + const [reconnectErr, ...results] = await Promise.all([ + engine.reconnect().then(() => null).catch((e) => e), + ...readers, + ]); + expect(reconnectErr).toBeNull(); + for (const ok of results) expect(ok).toBe(1); + }, 30_000); + + test('instance-pool reconnect() still teardown+recreates its own pool', async () => { + const engine = new PostgresEngine(); + await engine.connect({ database_url: DATABASE_URL!, poolSize: 2 }); + + // Works before. + const before = await engine.executeRaw<{ ok: number }>('SELECT 1 as ok'); + expect(before[0].ok).toBe(1); + + await engine.reconnect(); // instance path: tears down + rebuilds _sql + + // Works after (fresh pool). + const after = await engine.executeRaw<{ ok: number }>('SELECT 1 as ok'); + expect(after[0].ok).toBe(1); + + await engine.disconnect(); + + // Module singleton untouched by the instance engine's reconnect/disconnect. + const mod = await db.getConnection().unsafe('SELECT 1 as ok'); + expect((mod[0] as unknown as { ok: number }).ok).toBe(1); + }, 30_000); +}); diff --git a/test/eval-capture-drain.test.ts b/test/eval-capture-drain.test.ts new file mode 100644 index 000000000..2823027a3 --- /dev/null +++ b/test/eval-capture-drain.test.ts @@ -0,0 +1,77 @@ +/** + * v0.42.20.0 (#1762) — eval-capture is the 4th fire-and-forget DB-write sink the + * background-work registry drains before CLI disconnect. `captureEvalCandidate` + * is `void`-ed by the search/query op handlers; its async `logEvalCandidate` + * write is the same lock-pin / disconnect-race class as the other sinks. These + * tests pin the bounded drain (`awaitPendingEvalCaptures`). + */ +import { describe, test, expect, afterEach } from 'bun:test'; +import { + captureEvalCandidate, + awaitPendingEvalCaptures, + _resetPendingEvalCapturesForTests, + type CaptureContext, +} from '../src/core/eval-capture.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; +import type { SearchResult } from '../src/core/types.ts'; + +function makeCtx(): CaptureContext { + const result: SearchResult = { + slug: 'people/alice-example', + page_id: 1, + title: 'Alice Example', + type: 'person', + chunk_text: '…', + chunk_source: 'compiled_truth', + chunk_id: 42, + chunk_index: 0, + score: 0.9, + stale: false, + source_id: 'default', + }; + return { + tool_name: 'query', + query: 'who is alice', + results: [result], + meta: { vector_enabled: true, detail_resolved: 'medium', expansion_applied: false }, + latency_ms: 1, + remote: false, + expand_enabled: false, + detail: null, + job_id: null, + subagent_id: null, + }; +} + +afterEach(() => _resetPendingEvalCapturesForTests()); + +describe('awaitPendingEvalCaptures', () => { + test('empty set drains instantly', async () => { + const r = await awaitPendingEvalCaptures(50); + expect(r.unfinished).toBe(0); + }); + + test('drains a settled capture to unfinished:0', async () => { + const engine = { + logEvalCandidate: async () => 1, + logEvalCaptureFailure: async () => {}, + } as unknown as BrainEngine; + void captureEvalCandidate(engine, makeCtx(), { scrub_pii: false }); + const r = await awaitPendingEvalCaptures(1000); + expect(r.unfinished).toBe(0); + }); + + test('a hanging capture is bounded by the timeout (not a hang)', async () => { + const engine = { + // Never resolves — simulates a wedged DB write. + logEvalCandidate: () => new Promise(() => {}), + logEvalCaptureFailure: async () => {}, + } as unknown as BrainEngine; + void captureEvalCandidate(engine, makeCtx(), { scrub_pii: false }); + const start = Date.now(); + const r = await awaitPendingEvalCaptures(150); + const elapsed = Date.now() - start; + expect(r.unfinished).toBe(1); + expect(elapsed).toBeLessThan(1000); + }); +}); diff --git a/test/fix-wave-structural.test.ts b/test/fix-wave-structural.test.ts index 985c9dad8..16c9ccb75 100644 --- a/test/fix-wave-structural.test.ts +++ b/test/fix-wave-structural.test.ts @@ -20,19 +20,19 @@ import { describe, test, expect } from 'bun:test'; import { readFileSync } from 'fs'; -describe('v0.36.1.x #1125 — query drain cache writes before CLI exit', () => { - test("cli.ts awaits awaitPendingSearchCacheWrites for the 'query' op", () => { - const src = readFileSync('src/cli.ts', 'utf8'); - // Sequence: 'query' op match → import the drain → await - expect(src).toMatch(/op\.name\s*===\s*'query'[\s\S]{0,200}awaitPendingSearchCacheWrites/); - expect(src).toMatch(/await\s+awaitPendingSearchCacheWrites\(\)/); - }); - - test('hybrid.ts exports the drain helper + trackCacheWrite', () => { +describe('v0.42.20.0 — search-cache drained via the background-work registry', () => { + // Supersedes the v0.36.1.x #1125 query-only drain: search-cache now registers + // a registry drainer (drained for BOTH search and query, bounded), and cli.ts + // drains the whole registry rather than calling awaitPendingSearchCacheWrites + // directly for the 'query' op only. + test('hybrid.ts registers a bounded search-cache drainer', () => { const src = readFileSync('src/core/search/hybrid.ts', 'utf8'); expect(src).toMatch(/export async function awaitPendingSearchCacheWrites/); expect(src).toMatch(/pendingCacheWrites\.add\(promise\)/); expect(src).toMatch(/trackCacheWrite\(/); + // Now bounded (was an unbounded Promise.allSettled) + registered. + expect(src).toMatch(/registerBackgroundWorkDrainer\(\{[\s\S]*?name:\s*'search-cache'/); + expect(src).toMatch(/Promise\.race/); }); }); @@ -123,51 +123,58 @@ describe('v0.36.1.x #1124 — query --no-expand actually negates expand', () => }); }); -describe('v0.41.8.0 #1247/#1269/#1290 — drain last-retrieved before CLI disconnect', () => { - test('cli.ts imports awaitPendingLastRetrievedWrites', () => { +describe('v0.42.20.0 — background-work registry drains every sink before disconnect', () => { + // Supersedes the v0.41.8.0 #1247/#1269/#1290 per-call last-retrieved drain: + // last-retrieved is now one of four registry sinks; cli.ts drains the whole + // registry (drainAllBackgroundWorkForCliExit) before disconnect on BOTH the + // op-dispatch path AND the CLI_ONLY path (the latter closes #1762 for capture). + test('cli.ts imports + uses drainAllBackgroundWorkForCliExit', () => { const src = readFileSync('src/cli.ts', 'utf8'); - // Allow additional type-imports from the same module (e.g. `type DrainOutcome`) - expect(src).toMatch(/import\s+\{[^}]*\bawaitPendingLastRetrievedWrites\b[^}]*\}\s*from\s+['"]\.\/core\/last-retrieved\.ts['"]/); + expect(src).toMatch(/import\s+\{\s*drainAllBackgroundWorkForCliExit\s*\}\s*from\s+['"]\.\/core\/background-work\.ts['"]/); + // Two call sites: op-dispatch finally + handleCliOnly finally. + const calls = src.match(/await\s+drainAllBackgroundWorkForCliExit\s*\(/g) ?? []; + expect(calls.length).toBeGreaterThanOrEqual(2); }); - test('last-retrieved.ts exports the drain + tracks promises in a module-scoped Set', () => { + test('last-retrieved.ts still exports the bounded drain + registers a drainer', () => { const src = readFileSync('src/core/last-retrieved.ts', 'utf8'); expect(src).toMatch(/export async function awaitPendingLastRetrievedWrites/); expect(src).toMatch(/pendingLastRetrievedWrites\s*=\s*new\s+Set/); - expect(src).toMatch(/pendingLastRetrievedWrites\.add\(promise\)/); - // Per D5+D8: snapshot pattern (Codex finding #3) + bounded timeout expect(src).toMatch(/Promise\.race/); - expect(src).toMatch(/drain timed out/); + expect(src).toMatch(/registerBackgroundWorkDrainer\(\{[\s\S]*?name:\s*'last-retrieved'/); }); - test('cli.ts behavioral positioning: drain appears BEFORE engine.disconnect in op-dispatch', () => { + test('all four sinks register a drainer', () => { + expect(readFileSync('src/core/facts/queue.ts', 'utf8')) + .toMatch(/registerBackgroundWorkDrainer\(\{[\s\S]*?name:\s*'facts'[\s\S]*?abort:/); + expect(readFileSync('src/core/search/hybrid.ts', 'utf8')) + .toMatch(/name:\s*'search-cache'/); + expect(readFileSync('src/core/last-retrieved.ts', 'utf8')) + .toMatch(/name:\s*'last-retrieved'/); + expect(readFileSync('src/core/eval-capture.ts', 'utf8')) + .toMatch(/name:\s*'eval-capture'/); + }); + + test('cli.ts behavioral positioning: registry drain appears BEFORE engine.disconnect (op-dispatch)', () => { const src = readFileSync('src/cli.ts', 'utf8'); - // Per D5+D8: replaces the brittle literal-output regex from PR #1259 - // with a behavioral-positioning assertion. The drain CALL must appear - // textually before the disconnect CALL in the local-engine path. - // Match `await fn(` not bare names — bare names also appear in - // comments and would false-match the comment ordering. const localPath = src.match(/\/\/ Local engine path \(unchanged behavior[\s\S]+?^\}/m); expect(localPath).not.toBeNull(); const block = localPath![0]; - const drainCallRe = /await\s+awaitPendingLastRetrievedWrites\s*\(/; + const drainCallRe = /await\s+drainAllBackgroundWorkForCliExit\s*\(/; const disconnectCallRe = /await\s+engine\.disconnect\s*\(/; expect(block).toMatch(drainCallRe); expect(block).toMatch(disconnectCallRe); - const drainMatch = block.match(drainCallRe); - const disconnectMatch = block.match(disconnectCallRe); - const drainIdx = block.indexOf(drainMatch![0]); - const disconnectIdx = block.indexOf(disconnectMatch![0]); - expect(drainIdx).toBeGreaterThan(-1); - expect(disconnectIdx).toBeGreaterThan(-1); + const drainIdx = block.indexOf(block.match(drainCallRe)![0]); + const disconnectIdx = block.indexOf(block.match(disconnectCallRe)![0]); expect(drainIdx).toBeLessThan(disconnectIdx); }); - test('cli.ts uses shouldForceExitAfterMain only on the timeout path', () => { - const src = readFileSync('src/cli.ts', 'utf8'); - expect(src).toMatch(/import\s+\{\s*shouldForceExitAfterMain\s*\}\s*from\s+['"]\.\/core\/cli-force-exit\.ts['"]/); - // The force-exit gate MUST be conditioned on drainResult.outcome ==='timeout' - expect(src).toMatch(/drainResult\.outcome\s*===\s*['"]timeout['"]/); + test('background-work.ts: Map registry, ordered drain, awaited abort, test seam', () => { + const src = readFileSync('src/core/background-work.ts', 'utf8'); + expect(src).toMatch(/new\s+Map/); + expect(src).toMatch(/sort\(\s*\(a,\s*b\)\s*=>\s*a\.order\s*-\s*b\.order/); + expect(src).toMatch(/if\s*\(unfinished\s*>\s*0\s*&&\s*d\.abort\)\s*\{[\s\S]*?await\s+d\.abort\(\)/); + expect(src).toMatch(/export function __registerDrainerForTest/); }); test('cli-force-exit.ts daemon guard excludes "serve"', () => { diff --git a/test/search/query-embed-deadline.test.ts b/test/search/query-embed-deadline.test.ts new file mode 100644 index 000000000..c4d1181b7 --- /dev/null +++ b/test/search/query-embed-deadline.test.ts @@ -0,0 +1,85 @@ +/** + * v0.42.20.0 (Fix 3, #1775) — query-embed deadline unit tests. + * + * The regression: `search`/`query` default to cheap-hybrid, which embeds the + * query. A stalled embedding provider made the embed `await` never settle, so + * the handler never reached the keyword fallback and the CLI force-exited at 10s + * with no output. `embedQueryBounded` bounds the embed so it THROWS on timeout + * → the caller's existing try/catch falls back to keyword. + * + * These tests prove the bound fires even when the transport IGNORES the + * abortSignal (the Promise.race guarantee — codex #3: abortSignal alone is + * insufficient against a wedged provider), and that a shared/elapsed deadline + * makes a second embed fail FAST (worst case ~one timeout, not two). + */ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import { + configureGateway, + resetGateway, + __setEmbedTransportForTests, +} from '../../src/core/ai/gateway.ts'; +import { embedQueryBounded, makeQueryEmbedDeadline } from '../../src/core/search/hybrid.ts'; + +describe('embedQueryBounded — query-embed deadline', () => { + beforeEach(() => { + resetGateway(); + configureGateway({ + embedding_model: 'voyage:voyage-4-large', + embedding_dimensions: 1024, + env: { VOYAGE_API_KEY: 'voyage-fake' }, + }); + }); + afterEach(() => { + __setEmbedTransportForTests(null); + resetGateway(); + }); + + test('rejects within the budget when the transport hangs (ignores abort)', async () => { + // Transport never resolves AND ignores the abort signal — only the + // Promise.race deadline can save us. + __setEmbedTransportForTests(() => new Promise(() => { /* hang forever */ })); + const dl = makeQueryEmbedDeadline(200); + const start = Date.now(); + let threw = false; + try { + await embedQueryBounded('locker code', undefined, dl); + } catch (e) { + threw = true; + expect(String((e as Error).message)).toContain('deadline'); + } + const elapsed = Date.now() - start; + expect(threw).toBe(true); + // The 200ms deadline is floored to MIN_QUERY_EMBED_BUDGET_MS (2s) — the bound + // still fires (not infinite hang), comfortably under the 10s CLI force-exit. + expect(elapsed).toBeLessThan(3000); + }); + + test('an already-elapsed shared deadline is floored, not fresh-6s (codex floor)', async () => { + __setEmbedTransportForTests(() => new Promise(() => { /* hang forever */ })); + // Simulate the inner embed reusing a deadline the cache-lookup already spent. + const dl = { signal: AbortSignal.timeout(1), deadlineAt: Date.now() - 5 }; + const start = Date.now(); + let threw = false; + try { + await embedQueryBounded('q', undefined, dl); + } catch { + threw = true; + } + const elapsed = Date.now() - start; + expect(threw).toBe(true); + // Floored to MIN_QUERY_EMBED_BUDGET_MS (2s) — NOT a fresh 6s (would blow the + // cached-path total past the 10s force-exit) and NOT ~0 (would starve a + // healthy inner embed). So: rejects after ~2s, comfortably under 6s. + expect(elapsed).toBeGreaterThanOrEqual(1800); + expect(elapsed).toBeLessThan(3500); + }); + + test('resolves with the embedding when the transport returns in time', async () => { + const vec = Array.from({ length: 1024 }, () => 0.1); + __setEmbedTransportForTests(async () => ({ embeddings: [vec], usage: { tokens: 1 } }) as any); + const dl = makeQueryEmbedDeadline(2000); + const out = await embedQueryBounded('q', undefined, dl); + expect(out).toBeInstanceOf(Float32Array); + expect(out.length).toBe(1024); + }); +});