mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
v0.42.20.0 fix: reliability wave — PGLite capture lock-pin + Postgres reconnect race + search embed-hang (#1762 #1745 #1775) (#1810)
* fix(core): drain fire-and-forget sinks before disconnect via a background-work registry (#1762) New src/core/background-work.ts registry (Map<name,drainer>, 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 <noreply@github.com> Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * 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) <noreply@anthropic.com> --------- Co-authored-by: ElliotDrel <noreply@github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
ElliotDrel
Claude Opus 4.8
ElliotDrel
Claude Opus 4.8
parent
3d2add15d9
commit
ec5fed2921
+116
@@ -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<name, drainer>` (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.
|
||||
|
||||
@@ -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<T>()` 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+)
|
||||
|
||||
File diff suppressed because one or more lines are too long
+1
-1
@@ -143,5 +143,5 @@
|
||||
"bun": ">=1.3.10"
|
||||
},
|
||||
"license": "MIT",
|
||||
"version": "0.42.19.0"
|
||||
"version": "0.42.20.0"
|
||||
}
|
||||
|
||||
+47
-45
@@ -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<typeof setTimeout> | 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+59
-6
@@ -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<Float32Array> {
|
||||
*/
|
||||
export async function embedQuery(
|
||||
text: string,
|
||||
opts?: { embeddingModel?: string; dimensions?: number },
|
||||
opts?: { embeddingModel?: string; dimensions?: number; abortSignal?: AbortSignal },
|
||||
): Promise<Float32Array> {
|
||||
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<string[]> {
|
||||
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<ChatResult> {
|
||||
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,
|
||||
});
|
||||
|
||||
|
||||
@@ -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<name, drainer>
|
||||
* 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<void>;
|
||||
}
|
||||
|
||||
const drainers = new Map<string, BackgroundWorkDrainer>();
|
||||
|
||||
/** 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<void> {
|
||||
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 */
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -53,7 +53,7 @@ export async function embed(text: string): Promise<Float32Array> {
|
||||
*/
|
||||
export async function embedQuery(
|
||||
text: string,
|
||||
opts?: { embeddingModel?: string; dimensions?: number },
|
||||
opts?: { embeddingModel?: string; dimensions?: number; abortSignal?: AbortSignal },
|
||||
): Promise<Float32Array> {
|
||||
return gatewayEmbedQuery(text, opts);
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
// 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<void> {
|
||||
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<Promise<unknown>>();
|
||||
|
||||
function trackEvalCapture(promise: Promise<unknown>): 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<typeof setTimeout> | 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.
|
||||
*
|
||||
|
||||
@@ -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(),
|
||||
});
|
||||
|
||||
@@ -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
|
||||
|
||||
+39
-10
@@ -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<void> {
|
||||
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');
|
||||
|
||||
+134
-5
@@ -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<void> {
|
||||
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<typeof setTimeout> | 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<unknown>): 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<Float32Array> {
|
||||
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<typeof setTimeout> | undefined;
|
||||
const deadline = new Promise<never>((_, 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
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
@@ -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<number>(() => {}),
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<string,\s*BackgroundWorkDrainer>/);
|
||||
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"', () => {
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user