v0.42.49.0 feat(pace): native DB-contention pacing for embed/sync backfills (#2240)

* feat(pace): composable DB-contention pacer primitive createDbPacer/createNoopPacer (concurrency permit + in-band EWMA + jittered cooperative sleep, abort-throws, fail-open) + named pace-mode bundles (env>config>bundle, default off) + shared embed-backfill lock key. * feat(embed): wire DB-pacing into embed paths + single-flight + bounded keyset re-entry embedStaleForSource + CLI embedAllStale/embedAll lower worker count to the resolved cap and observe()/pace() their DB ops; embed job + embed-backfill handler resolve env>config>bundle; CLI --pace flags; --background carries overrides into the job payload; single-flight via shared per-source lock; budget-timer re-arm around paced sleeps; EmbedResult.pacing telemetry. * feat(sync): shared DB-pacer permit across parallel worker engines One pacer spans the per-worker PostgresEngines (the multi-pool permit case); observe() import writes, pace() between files, dispose on all exit paths. * docs(pace): CLAUDE.md Pace Mode section + regenerated llms bundle * fix(pace): pre-landing review fixes (Codex P1/P2) - never unref() the cooperative-sleep timer (could exit mid-sleep) - pace() excludes the wall-clock budget + re-arm after each sleep - pacing only lowers concurrency, never raises above an operator cap - serialized job pace resolves at config tier so GBRAIN_PACE_* still wins - --pace-max-concurrency consumes its value token * chore: bump version and changelog (v0.42.48.0) Native DB-contention pacing for embed/sync backfills. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: bump version to v0.42.49.0 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 22:15:33 +00:00 · 2026-06-17 06:58:07 -07:00
parent 7ea92d602c
commit 7968f84077
16 changed files with 1550 additions and 43 deletions
@@ -0,0 +1,202 @@
+/**
+ * Pins the DB-pacer primitive: concurrency cap (the real lever), abort-throws,
+ * in-band EWMA, jittered cooperative sleep, fail-open, and the no-op path.
+ *
+ * Uses a fake sleep + deterministic rng so there's no real wall-clock wait and
+ * no flakiness — the whole suite runs in microtask time.
+ */
+import { describe, expect, test } from 'bun:test';
+import { createDbPacer, createNoopPacer, type DbPacer } from '../src/core/db-pacer.ts';
+import { AbortError } from '../src/core/abort-check.ts';
+import { PACE_BUNDLES, type PaceBundle } from '../src/core/pace-mode.ts';
+
+const BUNDLE: PaceBundle = {
+  enabled: true,
+  maxConcurrency: 2,
+  paceAtMs: 100,
+  maxSleepMs: 1000,
+  ewmaAlpha: 0.5,
+};
+
+/** A fake sleep that resolves immediately but still honors abort. */
+function fakeSleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) return Promise.reject(new AbortError('aborted'));
+  return Promise.resolve();
+}
+
+function pacer(over: Partial<PaceBundle> = {}, rng = () => 0.5): DbPacer {
+  return createDbPacer({ bundle: { ...BUNDLE, ...over }, sleep: fakeSleep, rng });
+}
+
+describe('concurrency cap (the real lever)', () => {
+  test('caps simultaneous permits at maxConcurrency; release hands off FIFO', async () => {
+    const p = pacer({ maxConcurrency: 2 });
+    const a = await p.acquire();
+    const b = await p.acquire();
+    expect(p.snapshot().active).toBe(2);
+
+    let cResolved = false;
+    const cPromise = p.acquire().then((permit) => {
+      cResolved = true;
+      return permit;
+    });
+    // c is blocked: still 2 active, 1 waiter.
+    await Promise.resolve();
+    expect(cResolved).toBe(false);
+    expect(p.snapshot().active).toBe(2);
+    expect(p.snapshot().maxWaiters).toBe(1);
+
+    a.release(); // hands the slot to c
+    const c = await cPromise;
+    expect(cResolved).toBe(true);
+    expect(p.snapshot().active).toBe(2); // handoff kept the cap, never 3
+
+    b.release();
+    c.release();
+    expect(p.snapshot().active).toBe(0);
+  });
+
+  test('double release is idempotent', async () => {
+    const p = pacer({ maxConcurrency: 1 });
+    const a = await p.acquire();
+    a.release();
+    a.release();
+    expect(p.snapshot().active).toBe(0);
+  });
+});
+
+describe('abort throws (never falls into a DB call)', () => {
+  test('acquire throws AbortError when the signal fires while waiting', async () => {
+    const p = pacer({ maxConcurrency: 1 });
+    await p.acquire(); // exhausts the single slot
+    const ac = new AbortController();
+    const blocked = p.acquire(ac.signal);
+    ac.abort();
+    await expect(blocked).rejects.toBeInstanceOf(AbortError);
+    expect(p.snapshot().maxWaiters).toBe(1);
+  });
+
+  test('acquire throws immediately when already aborted', async () => {
+    const p = pacer();
+    const ac = new AbortController();
+    ac.abort();
+    await expect(p.acquire(ac.signal)).rejects.toBeInstanceOf(AbortError);
+  });
+
+  test('pace throws AbortError when aborted during sleep', async () => {
+    // sleep that rejects with AbortError when the signal is set.
+    const p = createDbPacer({
+      bundle: BUNDLE,
+      sleep: (_ms, signal) =>
+        signal?.aborted ? Promise.reject(new AbortError('aborted')) : Promise.resolve(),
+      rng: () => 0.5,
+    });
+    p.observe(1000); // EWMA well above paceAtMs → will sleep
+    const ac = new AbortController();
+    ac.abort();
+    await expect(p.pace(ac.signal)).rejects.toBeInstanceOf(AbortError);
+  });
+});
+
+describe('in-band EWMA + observe guard', () => {
+  test('first sample seeds EWMA; later samples smooth it', () => {
+    const p = pacer({ ewmaAlpha: 0.5 });
+    expect(p.snapshot().ewmaMs).toBeNull();
+    p.observe(200);
+    expect(p.snapshot().ewmaMs).toBe(200);
+    p.observe(400);
+    expect(p.snapshot().ewmaMs).toBe(300); // 0.5*400 + 0.5*200
+    expect(p.snapshot().sampleCount).toBe(2);
+  });
+
+  test('NaN / negative samples are ignored', () => {
+    const p = pacer();
+    p.observe(Number.NaN);
+    p.observe(-5);
+    p.observe(Infinity);
+    expect(p.snapshot().ewmaMs).toBeNull();
+    expect(p.snapshot().sampleCount).toBe(0);
+  });
+});
+
+describe('cooperative sleep', () => {
+  test('no sleep when EWMA is below paceAtMs', async () => {
+    const p = pacer({ paceAtMs: 100 });
+    p.observe(50);
+    await p.pace();
+    expect(p.snapshot().sleepCount).toBe(0);
+    expect(p.snapshot().totalSleptMs).toBe(0);
+  });
+
+  test('sleeps when EWMA exceeds paceAtMs; jittered to 50-100% of base, capped', async () => {
+    // rng=0 → jitter factor 0.5 → ms = round(base * 0.5).
+    const p = pacer({ paceAtMs: 100, maxSleepMs: 1000 }, () => 0);
+    p.observe(800); // base = min(1000, 800) = 800; ms = 400
+    await p.pace();
+    expect(p.snapshot().sleepCount).toBe(1);
+    expect(p.snapshot().totalSleptMs).toBe(400);
+  });
+
+  test('sleep base is capped at maxSleepMs', async () => {
+    const p = pacer({ paceAtMs: 100, maxSleepMs: 300 }, () => 1); // jitter factor 1.0
+    p.observe(5000); // base = min(300, 5000) = 300; ms = 300
+    await p.pace();
+    expect(p.snapshot().totalSleptMs).toBe(300);
+  });
+
+  test('different rng values decorrelate sleep durations (anti-thundering-herd)', async () => {
+    const lo = pacer({ paceAtMs: 100 }, () => 0); // factor 0.5
+    const hi = pacer({ paceAtMs: 100 }, () => 1); // factor 1.0
+    lo.observe(800);
+    hi.observe(800);
+    await lo.pace();
+    await hi.pace();
+    expect(lo.snapshot().totalSleptMs).toBe(400);
+    expect(hi.snapshot().totalSleptMs).toBe(800);
+  });
+});
+
+describe('fail-open', () => {
+  test('an unexpected (non-abort) sleep failure does not throw', async () => {
+    const p = createDbPacer({
+      bundle: BUNDLE,
+      sleep: () => Promise.reject(new Error('boom')),
+      rng: () => 0.5,
+    });
+    p.observe(1000);
+    await expect(p.pace()).resolves.toBeUndefined();
+    // The failed sleep is not counted.
+    expect(p.snapshot().sleepCount).toBe(0);
+  });
+});
+
+describe('no-op pacer (off mode / PGLite)', () => {
+  test('createNoopPacer: unbounded acquire, zero sleeps', async () => {
+    const p = createNoopPacer();
+    const permits = await Promise.all([p.acquire(), p.acquire(), p.acquire()]);
+    expect(permits).toHaveLength(3);
+    p.observe(99999);
+    await p.pace();
+    expect(p.snapshot().enabled).toBe(false);
+    expect(p.snapshot().totalSleptMs).toBe(0);
+  });
+
+  test('an off bundle yields a no-op pacer', async () => {
+    const p = createDbPacer({ bundle: PACE_BUNDLES.off });
+    expect(p.snapshot().enabled).toBe(false);
+    await p.acquire();
+    await p.acquire(); // never blocks despite maxConcurrency 0
+    expect(p.snapshot().active).toBe(0);
+  });
+});
+
+describe('dispose', () => {
+  test('dispose releases blocked acquirers with a no-op permit', async () => {
+    const p = pacer({ maxConcurrency: 1 });
+    await p.acquire();
+    const blocked = p.acquire();
+    p.dispose();
+    const permit = await blocked; // resolves instead of hanging
+    expect(permit).toBeDefined();
+  });
+});
@@ -0,0 +1,120 @@
+/**
+ * Pins the DB-pacing mode bundles + resolution chain.
+ * The load-bearing claim: env beats config (incident escape hatch), per-call
+ * beats env, and `off` resolves to a disabled bundle.
+ */
+import { describe, expect, test } from 'bun:test';
+import {
+  PACE_BUNDLES,
+  PACE_MODES,
+  DEFAULT_PACE_MODE,
+  isPaceMode,
+  resolvePaceMode,
+  loadOverridesFromConfig,
+  readPaceEnv,
+} from '../src/core/pace-mode.ts';
+
+describe('pace-mode bundles', () => {
+  test('off is the default and is disabled', () => {
+    expect(DEFAULT_PACE_MODE).toBe('off');
+    expect(PACE_BUNDLES.off.enabled).toBe(false);
+  });
+
+  test('enabled bundles cap concurrency below the unpaced default (20)', () => {
+    for (const m of ['gentle', 'balanced', 'aggressive'] as const) {
+      expect(PACE_BUNDLES[m].enabled).toBe(true);
+      expect(PACE_BUNDLES[m].maxConcurrency).toBeGreaterThanOrEqual(1);
+      expect(PACE_BUNDLES[m].maxConcurrency).toBeLessThan(20);
+    }
+  });
+
+  test('isPaceMode guards the union', () => {
+    expect(PACE_MODES.every(isPaceMode)).toBe(true);
+    expect(isPaceMode('turbo')).toBe(false);
+    expect(isPaceMode(undefined)).toBe(false);
+  });
+});
+
+describe('resolvePaceMode precedence', () => {
+  test('unknown mode falls back to off (disabled), mode_valid=false', () => {
+    const r = resolvePaceMode({ mode: 'turbo' });
+    expect(r.resolved_mode).toBe('off');
+    expect(r.mode_valid).toBe(false);
+    expect(r.enabled).toBe(false);
+  });
+
+  test('config mode applies its bundle', () => {
+    const r = resolvePaceMode({ mode: 'balanced' });
+    expect(r.resolved_mode).toBe('balanced');
+    expect(r.maxConcurrency).toBe(PACE_BUNDLES.balanced.maxConcurrency);
+  });
+
+  test('env mode beats config mode (incident escape hatch)', () => {
+    const r = resolvePaceMode({ mode: 'gentle', envMode: 'aggressive' });
+    expect(r.resolved_mode).toBe('aggressive');
+  });
+
+  test('per-call mode beats env and config', () => {
+    const r = resolvePaceMode({ mode: 'gentle', envMode: 'balanced', perCallMode: 'aggressive' });
+    expect(r.resolved_mode).toBe('aggressive');
+  });
+
+  test('env knob override beats config knob override', () => {
+    const r = resolvePaceMode({
+      mode: 'balanced',
+      configOverrides: { maxConcurrency: 6 },
+      envOverrides: { maxConcurrency: 2 },
+    });
+    expect(r.maxConcurrency).toBe(2);
+  });
+
+  test('per-call knob beats env and config', () => {
+    const r = resolvePaceMode({
+      mode: 'balanced',
+      configOverrides: { maxConcurrency: 6 },
+      envOverrides: { maxConcurrency: 2 },
+      perCall: { maxConcurrency: 12 },
+    });
+    expect(r.maxConcurrency).toBe(12);
+  });
+
+  test('clamps an out-of-range maxConcurrency back to the bundle when enabled', () => {
+    const r = resolvePaceMode({ mode: 'balanced', perCall: { maxConcurrency: 0 } });
+    expect(r.enabled).toBe(true);
+    expect(r.maxConcurrency).toBe(PACE_BUNDLES.balanced.maxConcurrency);
+  });
+
+  test('clamps a bad ewmaAlpha back to a sane default', () => {
+    const r = resolvePaceMode({ mode: 'balanced', perCall: { ewmaAlpha: 5 } });
+    expect(r.ewmaAlpha).toBeGreaterThan(0);
+    expect(r.ewmaAlpha).toBeLessThanOrEqual(1);
+  });
+});
+
+describe('config + env parsing', () => {
+  test('loadOverridesFromConfig parses present keys only', () => {
+    const ov = loadOverridesFromConfig({
+      'pace.max_concurrency': '5',
+      'pace.pace_at_ms': '400',
+    });
+    expect(ov.maxConcurrency).toBe(5);
+    expect(ov.paceAtMs).toBe(400);
+    expect(ov.maxSleepMs).toBeUndefined();
+  });
+
+  test('readPaceEnv reads GBRAIN_PACE_* including mode', () => {
+    const { envMode, envOverrides } = readPaceEnv({
+      GBRAIN_PACE_MODE: 'gentle',
+      GBRAIN_PACE_MAX_CONCURRENCY: '3',
+      GBRAIN_PACE_ENABLED: 'true',
+    });
+    expect(envMode).toBe('gentle');
+    expect(envOverrides.maxConcurrency).toBe(3);
+    expect(envOverrides.enabled).toBe(true);
+  });
+
+  test('rejects out-of-range env concurrency (falls through)', () => {
+    const { envOverrides } = readPaceEnv({ GBRAIN_PACE_MAX_CONCURRENCY: '99999' });
+    expect(envOverrides.maxConcurrency).toBeUndefined();
+  });
+});