mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
v0.42.24.0 fix(minions): route lock claim/renewLock through direct session pool (#1822)
* fix(minions): route lock claim/renewLock through direct session pool The Minion lock heartbeat (claim + renewLock) ran every UPDATE through engine.executeRaw(), which is hardcoded to the read pool. On Supabase that is the transaction-mode pooler (6543), which recycles connections per transaction. A lock is held for minutes, so the pooler periodically reaps the socket mid-heartbeat -> CONNECTION_ENDED -> the lock looks expired -> the worker force-evicts its own job and the claim loop wedges silently. Add BrainEngine.executeRawDirect(): same contract as executeRaw, but routes to the direct session-mode pool (5432, GBRAIN_DIRECT_DATABASE_URL) when dual-pool is active. No-op delegation on PGLite / non-Supabase / kill-switch. claim/renewLock now use it. Single-statement UPDATEs only, so the double-claim guard and the renewLock no-inline-retry contract are preserved. Statements inside an open transaction keep their tx connection (in-transaction guard keys on peekReadPool() !== _sql); the lock hot-path never runs inside transaction(). The Postgres impl shares its cancellation plumbing with executeRaw via a private runUnsafe helper. New test/postgres-execute-raw-direct.test.ts covers the routing decision (dual-pool on/off x in-tx/not + abort short-circuit) without a live Postgres; queue-lock-retry.test.ts gains a guard that claim can never fall back to executeRaw. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * v0.42.24.0 chore: bump version and changelog Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs: update project documentation for v0.42.24.0 Document executeRawDirect on the BrainEngine contract and the claim/renewLock direct-session-pool routing in KEY_FILES.md. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test: make D3 executeRaw-no-retry guard refactor-aware The DRY refactor in this PR extracted executeRaw/executeRawDirect's shared cancellation plumbing into a private runUnsafe(conn, ...) helper, so the single conn.unsafe() call moved out of executeRaw's body. The D3 guard read executeRaw's source and asserted conn.unsafe( appeared exactly once there, which now fails (it's zero — executeRaw delegates). The D3 invariant (no per-call retry wrapper) is unchanged; it just spans the delegate now. Update the guard to check both public methods delegate to runUnsafe without reconnect/retry, and assert the exactly-once conn.unsafe + cancel-only catch in runUnsafe. Also extends coverage to executeRawDirect so the lock hot-path can't reintroduce a retry wrapper either. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
f11d56cfca
commit
f868257405
@@ -262,28 +262,46 @@ describe('Eng-review D3 — executeRaw has no per-call retry wrapper', () => {
|
||||
it('PostgresEngine.executeRaw is a single-statement passthrough (no try/catch on connection errors)', () => {
|
||||
const src = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8');
|
||||
|
||||
// Find the executeRaw method in the class (not the helper inside withReservedConnection)
|
||||
// v0.41.18.0 (T5/A20): signature extended with optional `opts?: { signal?: AbortSignal }`
|
||||
// 3rd arg + multi-line shape for real query cancellation. Regex updated to
|
||||
// tolerate both the legacy single-line and the new multi-line signatures.
|
||||
// v0.42.24.0 (eng-review D1): the cancellation plumbing shared by executeRaw
|
||||
// and executeRawDirect was extracted into a private `runUnsafe(conn, ...)`
|
||||
// helper. executeRaw / executeRawDirect now pick a connection and delegate;
|
||||
// the single `conn.unsafe(` call lives in runUnsafe. The D3 invariant (no
|
||||
// per-call retry wrapper) is unchanged — it just spans the delegate now, so
|
||||
// this guard checks both the public methods AND the shared helper.
|
||||
|
||||
// Find executeRaw in the class (not the helper inside withReservedConnection).
|
||||
// v0.41.18.0 (T5/A20): signature extended with optional `opts?: { signal?: AbortSignal }`.
|
||||
const fnMatch = src.match(/async executeRaw<T = Record<string, unknown>>\(\s*sql: string,\s*params\?: unknown\[\][^)]*\):\s*Promise<T\[\]>\s*\{([\s\S]*?)\n \}/);
|
||||
expect(fnMatch).not.toBeNull();
|
||||
const body = fnMatch![1];
|
||||
|
||||
// Must not call reconnect() from this method (D3 intent: no per-call
|
||||
// retry — recovery is supervisor-driven via reconnect()).
|
||||
// executeRaw must not retry: no reconnect, no inline re-issue, and it must
|
||||
// delegate to runUnsafe rather than re-implementing the query path.
|
||||
expect(body).not.toContain('this.reconnect()');
|
||||
// Must call conn.unsafe directly, exactly ONCE (no retry re-issue).
|
||||
expect(body).toContain('conn.unsafe(');
|
||||
const unsafeCallCount = (body.match(/conn\.unsafe\(/g) || []).length;
|
||||
expect(unsafeCallCount).toBe(1);
|
||||
// The try/catch present here is ONLY for AbortSignal cancellation
|
||||
// swallow (v0.41.18.0 A20), NOT for connection retry. Confirm by checking
|
||||
// the swallowed throws are .cancel() not network re-issue.
|
||||
if (body.includes('catch')) {
|
||||
expect(body).toContain('this.runUnsafe');
|
||||
expect((body.match(/conn\.unsafe\(/g) || []).length).toBe(0);
|
||||
|
||||
// executeRawDirect (the Minion lock hot-path sibling) routes to the direct
|
||||
// session pool but must NOT introduce a retry wrapper either — same delegate.
|
||||
const directMatch = src.match(/async executeRawDirect<T = Record<string, unknown>>\(\s*sql: string,\s*params\?: unknown\[\][^)]*\):\s*Promise<T\[\]>\s*\{([\s\S]*?)\n \}/);
|
||||
expect(directMatch).not.toBeNull();
|
||||
const directBody = directMatch![1];
|
||||
expect(directBody).not.toContain('this.reconnect()');
|
||||
expect(directBody).toContain('this.runUnsafe');
|
||||
expect((directBody.match(/conn\.unsafe\(/g) || []).length).toBe(0);
|
||||
|
||||
// The shared helper issues conn.unsafe EXACTLY ONCE (no retry re-issue) and
|
||||
// never reconnects. Its try/catch is ONLY the AbortSignal cancellation
|
||||
// swallow (v0.41.18.0 A20), NOT a connection retry.
|
||||
const helperMatch = src.match(/private runUnsafe<T>\(\s*conn:[^)]*\):\s*Promise<T\[\]>\s*\{([\s\S]*?)\n \}/);
|
||||
expect(helperMatch).not.toBeNull();
|
||||
const helperBody = helperMatch![1];
|
||||
expect(helperBody).not.toContain('this.reconnect()');
|
||||
expect((helperBody.match(/conn\.unsafe\(/g) || []).length).toBe(1);
|
||||
if (helperBody.includes('catch')) {
|
||||
// If catch exists, it must be the cancel-swallow shape, NOT a retry shape.
|
||||
expect(body).not.toMatch(/catch[^{]*\{[\s\S]*?conn\.unsafe/);
|
||||
expect(body).not.toMatch(/catch[^{]*\{[\s\S]*?setTimeout/);
|
||||
expect(helperBody).not.toMatch(/catch[^{]*\{[\s\S]*?conn\.unsafe/);
|
||||
expect(helperBody).not.toMatch(/catch[^{]*\{[\s\S]*?setTimeout/);
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
import { describe, expect, test } from 'bun:test';
|
||||
import { PostgresEngine } from '../src/core/postgres-engine.ts';
|
||||
|
||||
/**
|
||||
* executeRawDirect routing decision (PR #1816 lock hot-path fix).
|
||||
*
|
||||
* The point of executeRawDirect is to send the Minion lock heartbeat
|
||||
* (claim/renewLock) to the DIRECT session-mode pool (port 5432) instead of the
|
||||
* transaction pooler (6543) that reaps connections mid-hold. That routing
|
||||
* branch only fires against a real Supabase dual-pool, which CI doesn't have —
|
||||
* so the decision itself (which connection gets the statement) went untested.
|
||||
*
|
||||
* This exercises the pure routing logic with a stubbed ConnectionManager and
|
||||
* fake Sql handles, covering all three shapes:
|
||||
*
|
||||
* ┌─────────────────────────────┬──────────────┬───────────────┐
|
||||
* │ engine shape │ dual-pool │ conn chosen │
|
||||
* ├─────────────────────────────┼──────────────┼───────────────┤
|
||||
* │ worker, not in tx │ active │ ddl() direct │
|
||||
* │ tx clone (_sql = tx conn) │ active │ this.sql (tx) │ ← atomicity
|
||||
* │ worker, not in tx │ inactive │ this.sql read │
|
||||
* └─────────────────────────────┴──────────────┴───────────────┘
|
||||
*/
|
||||
|
||||
type FakeSql = { unsafe: (sql: string, params?: unknown[]) => Promise<unknown[]> };
|
||||
|
||||
/** A fake postgres.js handle whose unsafe() tags rows with its label. */
|
||||
function fakeSql(label: string): FakeSql {
|
||||
return {
|
||||
unsafe: async () => [{ via: label }],
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a PostgresEngine with its connection internals stubbed so
|
||||
* executeRawDirect can be driven without a live database.
|
||||
*
|
||||
* - readConn → the read pool (what `this.sql` returns when _sql === readPool)
|
||||
* - directConn → what connectionManager.ddl() resolves to
|
||||
* - sqlOverride → forces the `get sql()` getter (used to model a tx clone whose
|
||||
* `this.sql` is the tx connection, distinct from peekReadPool()).
|
||||
*/
|
||||
function makeEngine(opts: {
|
||||
dualPoolActive: boolean;
|
||||
readConn: FakeSql;
|
||||
directConn: FakeSql;
|
||||
// When set, models a tx clone: _sql is the tx conn (!== peekReadPool()).
|
||||
txConn?: FakeSql;
|
||||
}): PostgresEngine {
|
||||
const engine = new PostgresEngine();
|
||||
const e = engine as unknown as Record<string, unknown>;
|
||||
|
||||
// _sql: tx conn for a clone, otherwise the read pool itself (the worker case
|
||||
// where connect() does setReadPool(this._sql)).
|
||||
e._sql = opts.txConn ?? opts.readConn;
|
||||
|
||||
// The `get sql()` getter on the prototype returns _sql when set, so we don't
|
||||
// need to override it — _sql already drives it. For a tx clone _sql is txConn,
|
||||
// so this.sql === txConn, exactly as Object.defineProperty does in transaction().
|
||||
|
||||
e.connectionManager = {
|
||||
isDualPoolActive: () => opts.dualPoolActive,
|
||||
peekReadPool: () => opts.readConn,
|
||||
ddl: async () => opts.directConn,
|
||||
};
|
||||
|
||||
return engine;
|
||||
}
|
||||
|
||||
describe('PostgresEngine.executeRawDirect — routing decision (PR #1816)', () => {
|
||||
test('dual-pool active + not in tx → routes to direct (ddl) pool', async () => {
|
||||
const readConn = fakeSql('read');
|
||||
const directConn = fakeSql('direct');
|
||||
const engine = makeEngine({ dualPoolActive: true, readConn, directConn });
|
||||
|
||||
const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1');
|
||||
expect(rows[0].via).toBe('direct');
|
||||
});
|
||||
|
||||
test('inside a transaction → honors the tx connection (never reroutes off it)', async () => {
|
||||
const readConn = fakeSql('read');
|
||||
const directConn = fakeSql('direct');
|
||||
const txConn = fakeSql('tx');
|
||||
// dual-pool is active, but _sql (tx) !== peekReadPool() (read) → inTransaction.
|
||||
const engine = makeEngine({ dualPoolActive: true, readConn, directConn, txConn });
|
||||
|
||||
const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1');
|
||||
expect(rows[0].via).toBe('tx');
|
||||
});
|
||||
|
||||
test('dual-pool inactive → falls back to the read pool', async () => {
|
||||
const readConn = fakeSql('read');
|
||||
const directConn = fakeSql('direct');
|
||||
const engine = makeEngine({ dualPoolActive: false, readConn, directConn });
|
||||
|
||||
const rows = await engine.executeRawDirect<{ via: string }>('UPDATE minion_jobs SET x=1');
|
||||
expect(rows[0].via).toBe('read');
|
||||
});
|
||||
|
||||
test('already-aborted signal short-circuits with AbortError before routing the query', async () => {
|
||||
const readConn = fakeSql('read');
|
||||
const directConn = fakeSql('direct');
|
||||
const engine = makeEngine({ dualPoolActive: true, readConn, directConn });
|
||||
|
||||
const ac = new AbortController();
|
||||
ac.abort();
|
||||
await expect(
|
||||
engine.executeRawDirect('UPDATE minion_jobs SET x=1', [], { signal: ac.signal }),
|
||||
).rejects.toThrow(/abort/i);
|
||||
});
|
||||
});
|
||||
@@ -58,7 +58,11 @@ describe('MinionQueue lock-path recovery (issue #1678)', () => {
|
||||
let calls = 0;
|
||||
const engine = {
|
||||
kind: 'postgres',
|
||||
executeRaw: async () => { calls++; throw connEndedError(); },
|
||||
// claim() routes through executeRawDirect (direct session pool) as of
|
||||
// the lock-hot-path fix; executeRaw is kept as a throwing guard to
|
||||
// prove claim never falls back to it.
|
||||
executeRawDirect: async () => { calls++; throw connEndedError(); },
|
||||
executeRaw: async () => { throw new Error('claim must not use executeRaw'); },
|
||||
reconnect: async () => {},
|
||||
} as unknown as ConstructorParameters<typeof MinionQueue>[0];
|
||||
|
||||
|
||||
Reference in New Issue
Block a user