mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-31 04:07:52 +00:00
Two infrastructure modules the subagent handler spine depends on:
rate-leases.ts — lease-based concurrency cap for outbound providers
(anthropic:messages, openai:*, etc.). Counter-based limiters leak capacity
on worker crash; leases are owner-tagged rows with expires_at that
auto-prune on the next acquire. Two-phase: txn-scoped pg_advisory_xact_lock
guards the check-then-insert so concurrent acquires can't both win the
"last slot". renewLeaseWithBackoff retries 3x (250/500/1000ms) for mid-
call DB blips — on persistent failure the LLM-loop caller aborts with a
renewable error so the worker re-claims and the rate invariant is
preserved. Owner FK cascades clean up leases on job deletion.
wait-for-completion.ts — poll-until-terminal helper for CLI callers.
Minions' NOTIFY is worker-side only; `gbrain agent run --follow` polls
getJob() until status is {completed, failed, dead, cancelled}. TimeoutError
carries jobId + elapsedMs and does NOT cancel the job — the user can
inspect via `gbrain jobs get <id>` later. Supports AbortSignal for Ctrl-C
without throwing. Default pollMs is 1000 on Postgres, 250 on PGLite (inline
CLI has no network RTT).
21 unit tests cover: single/multi acquire under cap, rejection past cap,
release frees slot, different keys are independent, stale prune, cascade
on owner delete, renew bumps expires_at, renew on missing is false,
backoff path success + pruned short-circuit. waitForCompletion: fast-path
terminal, transitions mid-wait (completed/failed/cancelled), TimeoutError
shape, abort-signal early exit, non-existent job error.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
149 lines
5.1 KiB
TypeScript
149 lines
5.1 KiB
TypeScript
/**
|
|
* Lease-based rate limiter tests. Runs against PGLite in-memory.
|
|
*/
|
|
|
|
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
|
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
|
import { MinionQueue } from '../src/core/minions/queue.ts';
|
|
import {
|
|
acquireLease,
|
|
renewLease,
|
|
releaseLease,
|
|
renewLeaseWithBackoff,
|
|
} from '../src/core/minions/rate-leases.ts';
|
|
|
|
let engine: PGLiteEngine;
|
|
let queue: MinionQueue;
|
|
let owner: number; // a minion_jobs.id to own leases (FK target)
|
|
|
|
beforeAll(async () => {
|
|
engine = new PGLiteEngine();
|
|
await engine.connect({ databaseUrl: '' });
|
|
await engine.initSchema();
|
|
queue = new MinionQueue(engine);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await engine.disconnect();
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
await engine.executeRaw('DELETE FROM subagent_rate_leases');
|
|
await engine.executeRaw('DELETE FROM minion_jobs');
|
|
const j = await queue.add('owner', {});
|
|
owner = j.id;
|
|
});
|
|
|
|
describe('acquireLease / releaseLease', () => {
|
|
test('single acquire under cap returns lease id', async () => {
|
|
const r = await acquireLease(engine, 'anthropic:messages', owner, 2);
|
|
expect(r.acquired).toBe(true);
|
|
expect(r.leaseId).toBeGreaterThan(0);
|
|
expect(r.activeCount).toBe(1);
|
|
});
|
|
|
|
test('acquires up to max_concurrent', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 2);
|
|
const b = await acquireLease(engine, 'k', owner, 2);
|
|
expect(a.acquired).toBe(true);
|
|
expect(b.acquired).toBe(true);
|
|
expect(b.activeCount).toBe(2);
|
|
});
|
|
|
|
test('rejects beyond max_concurrent', async () => {
|
|
await acquireLease(engine, 'k', owner, 2);
|
|
await acquireLease(engine, 'k', owner, 2);
|
|
const third = await acquireLease(engine, 'k', owner, 2);
|
|
expect(third.acquired).toBe(false);
|
|
expect(third.leaseId).toBeUndefined();
|
|
expect(third.activeCount).toBe(2);
|
|
});
|
|
|
|
test('releaseLease frees a slot', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1);
|
|
expect(a.acquired).toBe(true);
|
|
const blocked = await acquireLease(engine, 'k', owner, 1);
|
|
expect(blocked.acquired).toBe(false);
|
|
|
|
await releaseLease(engine, a.leaseId!);
|
|
|
|
const after = await acquireLease(engine, 'k', owner, 1);
|
|
expect(after.acquired).toBe(true);
|
|
});
|
|
|
|
test('different keys have independent capacity', async () => {
|
|
const a = await acquireLease(engine, 'k1', owner, 1);
|
|
const b = await acquireLease(engine, 'k2', owner, 1);
|
|
expect(a.acquired).toBe(true);
|
|
expect(b.acquired).toBe(true);
|
|
});
|
|
|
|
test('stale leases auto-prune on next acquire', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1, { ttlMs: 10 });
|
|
expect(a.acquired).toBe(true);
|
|
// Force the lease to be stale.
|
|
await engine.executeRaw(
|
|
`UPDATE subagent_rate_leases SET expires_at = now() - interval '1 minute' WHERE id = $1`,
|
|
[a.leaseId!],
|
|
);
|
|
const b = await acquireLease(engine, 'k', owner, 1);
|
|
expect(b.acquired).toBe(true);
|
|
// Only the fresh lease should remain.
|
|
const rows = await engine.executeRaw<{ count: string }>(
|
|
`SELECT count(*)::text AS count FROM subagent_rate_leases WHERE key = $1`,
|
|
['k'],
|
|
);
|
|
expect(parseInt(rows[0]!.count, 10)).toBe(1);
|
|
});
|
|
|
|
test('owner job deletion cascades lease rows', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1);
|
|
expect(a.acquired).toBe(true);
|
|
await engine.executeRaw(`DELETE FROM minion_jobs WHERE id = $1`, [owner]);
|
|
const rows = await engine.executeRaw<{ count: string }>(
|
|
`SELECT count(*)::text AS count FROM subagent_rate_leases WHERE key = $1`,
|
|
['k'],
|
|
);
|
|
expect(parseInt(rows[0]!.count, 10)).toBe(0);
|
|
});
|
|
|
|
test('releaseLease on a missing id is a no-op (idempotent)', async () => {
|
|
await expect(releaseLease(engine, 99_999)).resolves.toBeUndefined();
|
|
});
|
|
});
|
|
|
|
describe('renewLease', () => {
|
|
test('renewLease bumps expires_at and returns true', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1, { ttlMs: 50 });
|
|
const before = await engine.executeRaw<{ expires_at: string }>(
|
|
`SELECT expires_at FROM subagent_rate_leases WHERE id = $1`,
|
|
[a.leaseId!],
|
|
);
|
|
await new Promise(r => setTimeout(r, 5));
|
|
const ok = await renewLease(engine, a.leaseId!, 120_000);
|
|
expect(ok).toBe(true);
|
|
const after = await engine.executeRaw<{ expires_at: string }>(
|
|
`SELECT expires_at FROM subagent_rate_leases WHERE id = $1`,
|
|
[a.leaseId!],
|
|
);
|
|
expect(new Date(after[0]!.expires_at).getTime()).toBeGreaterThan(new Date(before[0]!.expires_at).getTime());
|
|
});
|
|
|
|
test('renewLease on a missing lease returns false', async () => {
|
|
expect(await renewLease(engine, 99_999)).toBe(false);
|
|
});
|
|
});
|
|
|
|
describe('renewLeaseWithBackoff', () => {
|
|
test('returns true on live lease', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1);
|
|
expect(await renewLeaseWithBackoff(engine, a.leaseId!)).toBe(true);
|
|
});
|
|
|
|
test('returns false on pruned lease (no retry loop)', async () => {
|
|
const a = await acquireLease(engine, 'k', owner, 1);
|
|
await releaseLease(engine, a.leaseId!);
|
|
expect(await renewLeaseWithBackoff(engine, a.leaseId!)).toBe(false);
|
|
});
|
|
});
|