mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
v0.42.29.0 fix(minions): long-job abort-honoring + attempt accounting + supervisor singleton; topic-aware voice (#1737, #1849, #1851) (#1943)
* fix(minions): honest attempt accounting + cooperative abort-honoring + per-handler timeouts (#1737) - Wall-clock and stall dead-letter paths now increment attempts_made (terminal, no retry — wall-clock fires at 2x cumulative timeout; retrying non-idempotent embed/subagent work would duplicate side effects). Surface stalled_counter in jobs get so 'started 3 / stalled 2 / attempts 0' reads true instead of looking like broken accounting. - Thread AbortSignal through embed-backfill/autopilot-cycle -> runPhaseEmbed -> runEmbedCore -> embedAll(Stale)/embedPage, checking it on BOTH --stale and --all paths and between embed batches. A timed-out embed phase now bails within a batch, so the cycle finally releases gbrain_cycle_locks instead of running the full 10-15 min after the job was killed (the daily cycle-wedge). New shared src/core/abort-check.ts (isAborted/throwIfAborted/anySignal). - Per-handler default wall-clock budget (handler-timeouts.ts) stamped at submit for long handlers without an explicit timeout_ms. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(minions): queue-scoped DB supervisor singleton + canonical pidfile + doctor max-rss check (#1849) - Acquire a queue-scoped DB lock (tryAcquireDbLock, keyed on the raw DB identity + queue) on supervisor.start(): a second supervisor on the same (db, queue) fails fast with exit 2 regardless of $HOME/--pid-file. Refresh on a dedicated timer; on refresh failure past the threshold, fail SAFE (exit non-zero) before the TTL could lapse and let a second supervisor take over. Release on shutdown. - Canonical default pidfile keyed on brain id (currentBrainId, config-only, no DB connect) so two brains under one HOME no longer share supervisor.pid. - doctor: new supervisor_singleton check surfaces the effective --max-rss (from the started audit event) and warns when the lock holder's (host,pid) differs from the local pidfile — comparing host+pid, not bare pid. Registered in doctor-categories. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(agent-voice): topic-aware persona context via server-resolved topicId (#1851) Summon Mars/Venus into a specific conversation topic so they boot already knowing the recent thread. A per-topic call link carries only topicId (+ optional display topicName); the server resolves the recent-conversation context from the brain (topics/<topicId>.md) — topic CONTENT is never accepted over the wire (that would be prompt injection + a leak into URLs/referrers/logs). topicId is a strict slug with a path-traversal guard. New '# Topic Context' prompt slot injected after the persona body so identity-first ordering still wins; persona identity unchanged. No topicId -> generic behavior. Contract doc + persona skill docs updated. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: file #1737 slot-reservation fair-scheduling follow-up TODO (F7) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: bump version and changelog (v0.42.29.0) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs: sync KEY_FILES.md for v0.42.29.0 minions + abort wave (#1737, #1849) Fold the #1737/#1849 behavior into the existing per-file entries and add the two new core files, keeping the doc at current-state truth: - queue.ts: honest attempt accounting on wall-clock + stall dead-letter paths; defaultTimeoutMsFor stamping at submit. - supervisor.ts: queue-scoped DB singleton lock (supervisorLockId, classifySupervisorSingleton, LOCK_LOST, refresh-fail-safe, brain-id pidfile, max_rss_mb audit). - worker-registry.ts: currentDbIdentity(). - New entries: src/core/abort-check.ts, src/core/minions/handler-timeouts.ts. - New doctor.ts extension: supervisor_singleton check. - cycle.ts / embed.ts extensions: AbortSignal threading note. 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
f7f8512b14
commit
613da94093
@@ -0,0 +1,72 @@
|
||||
/**
|
||||
* Tests for src/core/abort-check.ts (#1737 cooperative-abort helper).
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { isAborted, throwIfAborted, anySignal, AbortError } from '../src/core/abort-check.ts';
|
||||
|
||||
describe('abort-check: isAborted', () => {
|
||||
test('false for undefined / null / not-yet-fired signal', () => {
|
||||
expect(isAborted(undefined)).toBe(false);
|
||||
expect(isAborted(null)).toBe(false);
|
||||
expect(isAborted(new AbortController().signal)).toBe(false);
|
||||
});
|
||||
|
||||
test('true once the controller aborts', () => {
|
||||
const ac = new AbortController();
|
||||
ac.abort();
|
||||
expect(isAborted(ac.signal)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('abort-check: throwIfAborted', () => {
|
||||
test('no-op when signal absent or unfired', () => {
|
||||
expect(() => throwIfAborted(undefined)).not.toThrow();
|
||||
expect(() => throwIfAborted(new AbortController().signal)).not.toThrow();
|
||||
});
|
||||
|
||||
test('throws AbortError carrying the signal reason', () => {
|
||||
const ac = new AbortController();
|
||||
ac.abort(new Error('wall-clock'));
|
||||
try {
|
||||
throwIfAborted(ac.signal, '[cycle]');
|
||||
throw new Error('should have thrown');
|
||||
} catch (e) {
|
||||
expect(e).toBeInstanceOf(AbortError);
|
||||
expect((e as Error).name).toBe('AbortError');
|
||||
expect((e as Error).message).toBe('[cycle]: wall-clock');
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('abort-check: anySignal', () => {
|
||||
test('returns the internal signal unchanged when no external signal', () => {
|
||||
const internal = new AbortController().signal;
|
||||
expect(anySignal(internal, undefined)).toBe(internal);
|
||||
expect(anySignal(internal, null)).toBe(internal);
|
||||
});
|
||||
|
||||
test('combined signal fires when the EXTERNAL signal fires', () => {
|
||||
const internal = new AbortController();
|
||||
const external = new AbortController();
|
||||
const combined = anySignal(internal.signal, external.signal);
|
||||
expect(combined.aborted).toBe(false);
|
||||
external.abort();
|
||||
expect(combined.aborted).toBe(true);
|
||||
});
|
||||
|
||||
test('combined signal fires when the INTERNAL signal fires', () => {
|
||||
const internal = new AbortController();
|
||||
const external = new AbortController();
|
||||
const combined = anySignal(internal.signal, external.signal);
|
||||
internal.abort();
|
||||
expect(combined.aborted).toBe(true);
|
||||
});
|
||||
|
||||
test('combined is already aborted if external was pre-aborted', () => {
|
||||
const internal = new AbortController();
|
||||
const external = new AbortController();
|
||||
external.abort();
|
||||
const combined = anySignal(internal.signal, external.signal);
|
||||
expect(combined.aborted).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -89,13 +89,15 @@ describe('embed.ts → worker-pool migration (T3)', () => {
|
||||
expect(callSites?.length ?? 0).toBeGreaterThanOrEqual(2);
|
||||
});
|
||||
|
||||
test('embedAllStale path still threads budgetSignal into pool', () => {
|
||||
// The pre-migration code checked `!budgetSignal.aborted` in the
|
||||
// worker loop. The migration moves that check into the helper via
|
||||
// the `signal` option. If a future refactor drops the signal, the
|
||||
// wall-clock budget cancellation regresses.
|
||||
test('embedAllStale path still threads the cancellation signal into pool', () => {
|
||||
// The pre-migration code checked `!budgetSignal.aborted` in the worker
|
||||
// loop. The migration moves that check into the helper via the `signal`
|
||||
// option. #1737 then composed the wall-clock budget with the caller's
|
||||
// abort into `effectiveSignal` (anySignal(budgetSignal, externalSignal)) so
|
||||
// a killed job stops the pool too. If a future refactor drops the signal,
|
||||
// both wall-clock budget AND cooperative abort cancellation regress.
|
||||
expect(EMBED_SOURCE).toMatch(
|
||||
/runSlidingPool\(\s*\{[\s\S]*?signal:\s*budgetSignal[\s\S]*?\}\)/,
|
||||
/runSlidingPool\(\s*\{[\s\S]*?signal:\s*effectiveSignal[\s\S]*?\}\)/,
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -40,7 +40,7 @@ mock.module('../src/core/embedding.ts', () => ({
|
||||
}));
|
||||
|
||||
// Import AFTER mocking.
|
||||
const { runEmbed } = await import('../src/commands/embed.ts');
|
||||
const { runEmbed, runEmbedCore } = await import('../src/commands/embed.ts');
|
||||
|
||||
// v0.41.6.0 D1: runEmbedCore now preflights embedding credentials. This
|
||||
// test stack uses the LEGACY embedBatch mock path, not the gateway,
|
||||
@@ -133,6 +133,50 @@ describe('runEmbed --all (parallel)', () => {
|
||||
expect(stampCalls[0].args[1]).toEqual({ sourceId: 'default', signature: 'test:model:1536' });
|
||||
});
|
||||
|
||||
// #1737: cooperative abort. A pre-aborted signal must stop the embed loop
|
||||
// BEFORE any embedBatch call, so a job killed by wall-clock/lock-loss frees
|
||||
// the worker (and lets the cycle's finally release gbrain_cycle_locks)
|
||||
// instead of grinding through the full 10-15 min embed phase.
|
||||
test('#1737 --all: pre-aborted signal embeds nothing (no embedBatch call)', async () => {
|
||||
const pages = Array.from({ length: 10 }, (_, i) => ({ slug: `page-${i}`, source_id: 'default' }));
|
||||
const chunksBySlug = new Map(
|
||||
pages.map(p => [
|
||||
p.slug,
|
||||
[{ chunk_index: 0, chunk_text: `text ${p.slug}`, chunk_source: 'compiled_truth', embedded_at: null, token_count: 4 }],
|
||||
]),
|
||||
);
|
||||
const engine = mockEngine({
|
||||
listPages: async () => pages,
|
||||
getChunks: async (slug: string) => chunksBySlug.get(slug) || [],
|
||||
upsertChunks: async () => {},
|
||||
});
|
||||
|
||||
const ac = new AbortController();
|
||||
ac.abort(new Error('wall-clock'));
|
||||
const result = await runEmbedCore(engine, { all: true, signal: ac.signal });
|
||||
|
||||
expect(totalEmbedCalls).toBe(0);
|
||||
expect(result.embedded).toBe(0);
|
||||
});
|
||||
|
||||
test('#1737 --stale: pre-aborted signal breaks the loop before listStaleChunks', async () => {
|
||||
let listStaleCalls = 0;
|
||||
const engine = mockEngine({
|
||||
countStaleChunks: async () => 5, // non-zero so we pass the early return
|
||||
listStaleChunks: async () => { listStaleCalls++; return []; },
|
||||
invalidateStaleSignatureEmbeddings: async () => 0,
|
||||
});
|
||||
|
||||
const ac = new AbortController();
|
||||
ac.abort(new Error('lock-lost'));
|
||||
const result = await runEmbedCore(engine, { stale: true, signal: ac.signal });
|
||||
|
||||
// The top-of-loop abort check fires before the first listStaleChunks page load.
|
||||
expect(listStaleCalls).toBe(0);
|
||||
expect(totalEmbedCalls).toBe(0);
|
||||
expect(result.embedded).toBe(0);
|
||||
});
|
||||
|
||||
test('respects GBRAIN_EMBED_CONCURRENCY=1 (serial)', async () => {
|
||||
const pages = Array.from({ length: 5 }, (_, i) => ({ slug: `page-${i}` }));
|
||||
const chunksBySlug = new Map(
|
||||
|
||||
Vendored
+8
@@ -23,9 +23,17 @@ import type { BrainEngine } from '../../src/core/engine.ts';
|
||||
|
||||
// Mock engine: healthCheck() calls engine.executeRaw; return empty rows so
|
||||
// the query path exercises without needing Postgres.
|
||||
//
|
||||
// #1849: start() now acquires the queue-scoped DB singleton lock via
|
||||
// tryAcquireDbLock, which uses the postgres `sql` tagged-template escape hatch.
|
||||
// The stub returns a single row from every call so acquire succeeds (length 1
|
||||
// → acquired) and refresh/release are no-ops. Each spawned runner is a fresh
|
||||
// process, so there's no cross-test lock state to clean up.
|
||||
const sqlStub = (..._args: unknown[]) => Promise.resolve([{ id: 'supervisor-lock' }]);
|
||||
const mockEngine: Partial<BrainEngine> = {
|
||||
kind: 'postgres' as const,
|
||||
executeRaw: async () => [],
|
||||
sql: sqlStub,
|
||||
} as unknown as BrainEngine;
|
||||
|
||||
const pidFile = process.env.SUP_PID_FILE;
|
||||
|
||||
@@ -270,6 +270,101 @@ describe('MinionQueue: Stall Detection', () => {
|
||||
});
|
||||
});
|
||||
|
||||
// --- #1737 — honest attempt accounting on terminal dead-letter paths ---
|
||||
|
||||
describe('MinionQueue: #1737 attempt accounting on dead-letter', () => {
|
||||
test('wall-clock dead-letter increments attempts_made (no more 0/N (started:M))', async () => {
|
||||
const job = await queue.add('sync', {}, { max_attempts: 2 });
|
||||
// Per-job timeout so the wall-clock threshold is timeout_ms * 2 = 2s.
|
||||
await engine.executeRaw('UPDATE minion_jobs SET timeout_ms = 1000 WHERE id = $1', [job.id]);
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
expect((await queue.getJob(job.id))!.attempts_made).toBe(0); // bug repro: started but not made
|
||||
|
||||
// Force cumulative wall-clock past timeout_ms * 2.
|
||||
await engine.executeRaw(
|
||||
"UPDATE minion_jobs SET started_at = now() - interval '10 seconds' WHERE id = $1",
|
||||
[job.id]
|
||||
);
|
||||
const dead = await queue.handleWallClockTimeouts(30000);
|
||||
expect(dead.length).toBe(1);
|
||||
expect(dead[0].status).toBe('dead');
|
||||
expect(dead[0].error_text).toBe('wall-clock timeout exceeded');
|
||||
// The fix: a job killed by wall-clock consumed an attempt.
|
||||
expect(dead[0].attempts_made).toBe(1);
|
||||
// Constraint chk_attempts_order (attempts_made <= attempts_started) holds.
|
||||
expect(dead[0].attempts_made).toBeLessThanOrEqual(dead[0].attempts_started);
|
||||
});
|
||||
|
||||
test('wall-clock dead-letter is terminal — does NOT retry even with attempts remaining', async () => {
|
||||
const job = await queue.add('sync', {}, { max_attempts: 5 });
|
||||
await engine.executeRaw('UPDATE minion_jobs SET timeout_ms = 1000 WHERE id = $1', [job.id]);
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await engine.executeRaw(
|
||||
"UPDATE minion_jobs SET started_at = now() - interval '10 seconds' WHERE id = $1",
|
||||
[job.id]
|
||||
);
|
||||
const dead = await queue.handleWallClockTimeouts(30000);
|
||||
expect(dead.length).toBe(1);
|
||||
// Even with 4 attempts left, wall-clock is terminal (non-idempotent handler safety).
|
||||
expect(dead[0].status).toBe('dead');
|
||||
expect(dead[0].status).not.toBe('delayed');
|
||||
});
|
||||
|
||||
test('stall dead-letter increments attempts_made; requeue does NOT', async () => {
|
||||
const job = await queue.add('sync', {}, { max_attempts: 3 });
|
||||
await engine.executeRaw('UPDATE minion_jobs SET max_stalled = 2 WHERE id = $1', [job.id]);
|
||||
|
||||
// First stall: requeued, attempts_made stays 0 (lease-loss recovery, not an app attempt).
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await engine.executeRaw(
|
||||
"UPDATE minion_jobs SET lock_until = now() - interval '1 second' WHERE id = $1",
|
||||
[job.id]
|
||||
);
|
||||
const r1 = await queue.handleStalled();
|
||||
expect(r1.requeued.length).toBe(1);
|
||||
expect(r1.requeued[0].attempts_made).toBe(0);
|
||||
|
||||
// Second stall: dead-lettered, attempts_made now increments.
|
||||
await queue.claim('tok2', 30000, 'default', ['sync']);
|
||||
await engine.executeRaw(
|
||||
"UPDATE minion_jobs SET lock_until = now() - interval '1 second' WHERE id = $1",
|
||||
[job.id]
|
||||
);
|
||||
const r2 = await queue.handleStalled();
|
||||
expect(r2.dead.length).toBe(1);
|
||||
expect(r2.dead[0].error_text).toBe('max stalled count exceeded');
|
||||
expect(r2.dead[0].attempts_made).toBe(1);
|
||||
expect(r2.dead[0].attempts_made).toBeLessThanOrEqual(r2.dead[0].attempts_started);
|
||||
});
|
||||
});
|
||||
|
||||
// --- #1737 — per-handler default wall-clock budget at submit ---
|
||||
|
||||
describe('MinionQueue: #1737 per-handler default timeout', () => {
|
||||
test('long handler with no explicit timeout_ms gets the 30-min default stamped', async () => {
|
||||
const job = await queue.add('embed-backfill', { sourceId: 'x' });
|
||||
expect(job.timeout_ms).toBe(30 * 60 * 1000);
|
||||
});
|
||||
|
||||
test('autopilot-cycle + subagent also get the long default', async () => {
|
||||
const cycle = await queue.add('autopilot-cycle', {});
|
||||
// subagent is a protected name → needs the trusted-submit flag (4th arg).
|
||||
const sub = await queue.add('subagent', {}, undefined, { allowProtectedSubmit: true });
|
||||
expect(cycle.timeout_ms).toBe(30 * 60 * 1000);
|
||||
expect(sub.timeout_ms).toBe(30 * 60 * 1000);
|
||||
});
|
||||
|
||||
test('explicit timeout_ms always wins over the default', async () => {
|
||||
const job = await queue.add('embed-backfill', { sourceId: 'x' }, { timeout_ms: 5000 });
|
||||
expect(job.timeout_ms).toBe(5000);
|
||||
});
|
||||
|
||||
test('short handler keeps null timeout_ms (tight wall-clock default applies)', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
expect(job.timeout_ms).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// --- v0.13.1 #219 — max_stalled default + input surface ---
|
||||
|
||||
describe('MinionQueue: v0.13.1 max_stalled schema default (#219)', () => {
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
/**
|
||||
* #1849: queue-scoped DB supervisor singleton.
|
||||
*
|
||||
* The pidfile guard is mutually exclusive only per pidfile PATH; the DB lock
|
||||
* makes the (database, queue) pair the mutex domain so two supervisors with
|
||||
* different $HOME / --pid-file can't both run on one queue. These tests pin:
|
||||
* - the lock id keys on DB identity + queue (T2)
|
||||
* - a second acquire of the same (db, queue) lock is refused (the singleton)
|
||||
* - different queues don't collide
|
||||
* - refresh-failure past the threshold fails SAFE (exits non-zero) (F1A)
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach, spyOn } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { tryAcquireDbLock } from '../src/core/db-lock.ts';
|
||||
import { MinionSupervisor, ExitCodes, supervisorLockId, classifySupervisorSingleton } from '../src/core/minions/supervisor.ts';
|
||||
import type { DbLockHandle } from '../src/core/db-lock.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id LIKE 'gbrain-supervisor:%'`);
|
||||
});
|
||||
|
||||
describe('#1849 supervisorLockId', () => {
|
||||
test('keys on DB identity AND queue', () => {
|
||||
expect(supervisorLockId('default', 'postgres://x')).toBe('gbrain-supervisor:postgres://x:default');
|
||||
expect(supervisorLockId('shell', 'postgres://x')).toBe('gbrain-supervisor:postgres://x:shell');
|
||||
// Different DB identity → different lock even for the same queue.
|
||||
expect(supervisorLockId('default', 'postgres://a'))
|
||||
.not.toBe(supervisorLockId('default', 'postgres://b'));
|
||||
});
|
||||
});
|
||||
|
||||
describe('#1849 classifySupervisorSingleton (doctor)', () => {
|
||||
test('no live lock → no_lock', () => {
|
||||
expect(classifySupervisorSingleton({
|
||||
lockLive: false, lockHolderHost: 'h', lockHolderPid: 1, localHost: 'h', localPid: 1,
|
||||
})).toBe('no_lock');
|
||||
});
|
||||
|
||||
test('live lock held by the local (host,pid) → single', () => {
|
||||
expect(classifySupervisorSingleton({
|
||||
lockLive: true, lockHolderHost: 'box', lockHolderPid: 42, localHost: 'box', localPid: 42,
|
||||
})).toBe('single');
|
||||
});
|
||||
|
||||
test('live lock held by a DIFFERENT pid → mismatch (rogue second supervisor)', () => {
|
||||
expect(classifySupervisorSingleton({
|
||||
lockLive: true, lockHolderHost: 'box', lockHolderPid: 99, localHost: 'box', localPid: 42,
|
||||
})).toBe('mismatch');
|
||||
});
|
||||
|
||||
test('same pid but DIFFERENT host → mismatch (bare pid is meaningless cross-host)', () => {
|
||||
expect(classifySupervisorSingleton({
|
||||
lockLive: true, lockHolderHost: 'other', lockHolderPid: 42, localHost: 'box', localPid: 42,
|
||||
})).toBe('mismatch');
|
||||
});
|
||||
|
||||
test('live lock but no local pidfile → mismatch', () => {
|
||||
expect(classifySupervisorSingleton({
|
||||
lockLive: true, lockHolderHost: 'box', lockHolderPid: 42, localHost: 'box', localPid: null,
|
||||
})).toBe('mismatch');
|
||||
});
|
||||
});
|
||||
|
||||
describe('#1849 DB lock is the real singleton', () => {
|
||||
test('second acquire of the same (db, queue) lock is refused', async () => {
|
||||
const id = supervisorLockId('default', 'pglite:test');
|
||||
const first = await tryAcquireDbLock(engine, id, 5);
|
||||
expect(first).not.toBeNull();
|
||||
// A second supervisor (different pidfile, same db+queue) gets null → exit 2.
|
||||
const second = await tryAcquireDbLock(engine, id, 5);
|
||||
expect(second).toBeNull();
|
||||
// After release, a fresh supervisor can take over.
|
||||
await first!.release();
|
||||
const third = await tryAcquireDbLock(engine, id, 5);
|
||||
expect(third).not.toBeNull();
|
||||
await third!.release();
|
||||
});
|
||||
|
||||
test('different queues on the same DB do not collide', async () => {
|
||||
const a = await tryAcquireDbLock(engine, supervisorLockId('default', 'pglite:test'), 5);
|
||||
const b = await tryAcquireDbLock(engine, supervisorLockId('shell', 'pglite:test'), 5);
|
||||
expect(a).not.toBeNull();
|
||||
expect(b).not.toBeNull();
|
||||
await a!.release();
|
||||
await b!.release();
|
||||
});
|
||||
});
|
||||
|
||||
describe('#1849 refresh-failure fails safe (F1A)', () => {
|
||||
test('exits LOCK_LOST after the failure threshold; tolerates a single blip', async () => {
|
||||
const sup = new MinionSupervisor(engine, { cliPath: '/bin/sh', healthInterval: 0, json: true });
|
||||
const exitSpy = spyOn(process, 'exit').mockImplementation(((_code?: number) => {
|
||||
throw new Error(`exit:${_code}`); // stop execution like the real exit would
|
||||
}) as never);
|
||||
|
||||
let refreshCalls = 0;
|
||||
const failingLock: DbLockHandle = {
|
||||
id: 'x',
|
||||
refresh: async () => { refreshCalls++; throw new Error('pooler down'); },
|
||||
release: async () => {},
|
||||
};
|
||||
sup._setDbLockForTests(failingLock);
|
||||
|
||||
try {
|
||||
// First two failures: tolerated (counter climbs, no exit).
|
||||
await sup._refreshDbLockForTests();
|
||||
await sup._refreshDbLockForTests();
|
||||
expect(exitSpy).not.toHaveBeenCalled();
|
||||
// Third failure crosses the threshold → shutdown → process.exit(LOCK_LOST).
|
||||
try { await sup._refreshDbLockForTests(); } catch { /* exit stub throws */ }
|
||||
expect(exitSpy).toHaveBeenCalledWith(ExitCodes.LOCK_LOST);
|
||||
expect(refreshCalls).toBe(3);
|
||||
} finally {
|
||||
exitSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
test('a successful refresh resets the failure counter', async () => {
|
||||
const sup = new MinionSupervisor(engine, { cliPath: '/bin/sh', healthInterval: 0, json: true });
|
||||
const exitSpy = spyOn(process, 'exit').mockImplementation(((_code?: number) => {
|
||||
throw new Error(`exit:${_code}`);
|
||||
}) as never);
|
||||
|
||||
let mode: 'fail' | 'ok' = 'fail';
|
||||
const flakyLock: DbLockHandle = {
|
||||
id: 'x',
|
||||
refresh: async () => { if (mode === 'fail') throw new Error('blip'); },
|
||||
release: async () => {},
|
||||
};
|
||||
sup._setDbLockForTests(flakyLock);
|
||||
|
||||
try {
|
||||
await sup._refreshDbLockForTests(); // fail 1
|
||||
await sup._refreshDbLockForTests(); // fail 2
|
||||
mode = 'ok';
|
||||
await sup._refreshDbLockForTests(); // success → reset
|
||||
mode = 'fail';
|
||||
await sup._refreshDbLockForTests(); // fail 1 again
|
||||
await sup._refreshDbLockForTests(); // fail 2
|
||||
// Counter was reset, so we are NOT past threshold yet.
|
||||
expect(exitSpy).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
exitSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user