v0.42.36.0 fix(sync): resumable, durable, single-flight sync — converges under pool exhaustion + repeated kills (#1794) (#1980)

* fix(retry): match EMAXCONNSESSION + SQLSTATE 53300 as retryable conn errors (#1794) * feat(schema): add op_checkpoint_paths append-only delta table (migration v115) (#1794) * refactor(op-checkpoint): append-only deltas via executeRawDirect + withRetry (#1794) * fix(sync): resumable-checkpoint durability + lock-thrash fix (#1794) Durable append-only checkpoint writes (executeRawDirect + retry), fail-loud consecutive-failure abort, first-file/10s flush cadence, race-safe pending-delta under parallel workers, guaranteed final flush on every exit path incl. SIGTERM (no-retry one-shot via registerCleanup), bankedFiles/reason observability, event-loop yield to keep the lock heartbeat alive, and routing the bare (no-source) sync through withRefreshingLock. * fix(db-lock): heartbeat-aware takeover + direct-pool refresh (#1794) * fix(cycle): treat SyncLockBusyError as skip, not a phase failure (#1794) * docs(sync): document the 5 checkpoint/lock env knobs (#1794) * v0.42.36.0 fix(sync): resumable, durable, single-flight sync — converges under pool exhaustion + repeated kills (#1794) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs(key-files): update sync.ts + op-checkpoint.ts entries to resumable-checkpoint current state (#1794) --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 22:15:33 +00:00 · 2026-06-08 06:16:34 -07:00
parent 612753f318
commit 959af1068d
18 changed files with 834 additions and 133 deletions
@@ -0,0 +1,162 @@
+/**
+ * #1794 — heartbeat-aware lock takeover + direct-pool refresh.
+ *
+ * The lock-thrash fix: a holder that refreshed within the steal-grace window is
+ * NOT stolen even if its TTL lapsed (starved-but-alive), while a holder that
+ * stopped refreshing past the grace IS stolen. These tests isolate the ON
+ * CONFLICT grace logic by holding with `process.pid` (alive) so the dead-pid
+ * auto-takeover path can't fire — a successful steal therefore proves the
+ * grace/last_refreshed_at predicate did it.
+ *
+ * Plus the commit-8 causal-mechanism check: setImmediate yields let a
+ * setInterval tick fire during a busy loop (the reason the import loop's
+ * maybeYield keeps the refresh heartbeat alive).
+ */
+
+import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
+import { hostname } from 'os';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import {
+  tryAcquireDbLock,
+  resolveStealGraceSeconds,
+  DEFAULT_STEAL_GRACE_SECONDS,
+} from '../src/core/db-lock.ts';
+
+let engine: PGLiteEngine;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({ database_url: '' });
+  await engine.initSchema();
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id LIKE 'test-hb-%'`);
+});
+
+const LOCAL = hostname();
+
+/**
+ * Seed a TTL-EXPIRED lock held by an ALIVE pid (process.pid), with a chosen
+ * last_refreshed_at age (or NULL). Alive holder → dead-pid auto-takeover can't
+ * fire, so the only reclaim path is the ON CONFLICT grace predicate.
+ */
+async function seedExpiredLock(id: string, refreshedSecondsAgo: number | null): Promise<void> {
+  if (refreshedSecondsAgo === null) {
+    await engine.executeRaw(
+      `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
+       VALUES ($1, $2, $3, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '5 minutes', NULL)`,
+      [id, process.pid, LOCAL],
+    );
+  } else {
+    await engine.executeRaw(
+      `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
+       VALUES ($1, $2, $3, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '5 minutes', NOW() - ($4 || ' seconds')::interval)`,
+      [id, process.pid, LOCAL, String(refreshedSecondsAgo)],
+    );
+  }
+}
+
+async function refreshedAge(id: string): Promise<Date | null> {
+  const rows = await engine.executeRaw<{ last_refreshed_at: string | null }>(
+    `SELECT last_refreshed_at FROM gbrain_cycle_locks WHERE id = $1`,
+    [id],
+  );
+  const v = rows[0]?.last_refreshed_at ?? null;
+  return v ? new Date(v) : null;
+}
+
+describe('resolveStealGraceSeconds', () => {
+  test('derives ~2 refresh ticks from the TTL (30min → 600s)', () => {
+    // refresh ~ttl/6 = 5min = 300s; *2 = 600s.
+    expect(resolveStealGraceSeconds(30)).toBe(DEFAULT_STEAL_GRACE_SECONDS);
+    expect(resolveStealGraceSeconds(30)).toBe(600);
+  });
+
+  test('floors at 60s for tiny TTLs', () => {
+    expect(resolveStealGraceSeconds(1)).toBe(60);
+  });
+
+  test('env override wins', () => {
+    process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = '123';
+    try {
+      expect(resolveStealGraceSeconds(30)).toBe(123);
+    } finally {
+      delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
+    }
+  });
+
+  test('bad env override falls back to derived', () => {
+    process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = 'nope';
+    try {
+      expect(resolveStealGraceSeconds(30)).toBe(600);
+    } finally {
+      delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
+    }
+  });
+});
+
+describe('heartbeat-aware takeover (ON CONFLICT grace predicate)', () => {
+  test('FRESH holder is NOT stolen even with an expired TTL', async () => {
+    // ttl expired 5min ago, but refreshed 1s ago → inside the 600s grace.
+    await seedExpiredLock('test-hb-fresh', 1);
+    const handle = await tryAcquireDbLock(engine, 'test-hb-fresh', 30);
+    expect(handle).toBeNull();
+  });
+
+  test('STALE-refresh holder IS stolen (refresh older than the grace)', async () => {
+    // ttl expired AND last refresh 20min ago (> 600s grace) → stealable, even
+    // though the holder pid is alive (proves the grace path, not auto-takeover).
+    await seedExpiredLock('test-hb-stale', 1200);
+    const handle = await tryAcquireDbLock(engine, 'test-hb-stale', 30);
+    expect(handle).not.toBeNull();
+    await handle!.release();
+  });
+
+  test('NULL last_refreshed_at (pre-v98 row) IS stolen on TTL expiry', async () => {
+    await seedExpiredLock('test-hb-null', null);
+    const handle = await tryAcquireDbLock(engine, 'test-hb-null', 30);
+    expect(handle).not.toBeNull();
+    await handle!.release();
+  });
+
+  test('a reclaimed handle refresh() bumps last_refreshed_at (direct pool path)', async () => {
+    await seedExpiredLock('test-hb-bump', 1200);
+    const handle = await tryAcquireDbLock(engine, 'test-hb-bump', 30);
+    expect(handle).not.toBeNull();
+    const before = await refreshedAge('test-hb-bump');
+    // Small real delay so NOW() advances measurably between acquire and refresh.
+    await new Promise((r) => setTimeout(r, 20));
+    await handle!.refresh();
+    const after = await refreshedAge('test-hb-bump');
+    expect(before).not.toBeNull();
+    expect(after).not.toBeNull();
+    expect(after!.getTime()).toBeGreaterThan(before!.getTime());
+    await handle!.release();
+  });
+});
+
+describe('event-loop yield keeps timers alive (commit 8 mechanism)', () => {
+  test('setTimeout(0) yields let a setInterval heartbeat fire during a busy loop', async () => {
+    let ticks = 0;
+    const iv = setInterval(() => { ticks++; }, 2);
+    try {
+      // Mirror the import loop's maybeYield: setTimeout(0) enters the timers
+      // phase, so the setInterval heartbeat can fire mid-loop. (A setImmediate
+      // loop starves the timers phase in Bun — the reason maybeYield uses
+      // setTimeout, not setImmediate.) Bound by wall-clock so the 2ms interval
+      // has real time to fire.
+      const start = Date.now();
+      while (Date.now() - start < 40) {
+        await new Promise<void>((r) => setTimeout(r, 0));
+      }
+    } finally {
+      clearInterval(iv);
+    }
+    expect(ticks).toBeGreaterThan(0);
+  });
+});
@@ -4,6 +4,7 @@ import { resetPgliteState } from './helpers/reset-pglite.ts';
 import {
  loadOpCheckpoint,
  recordCompleted,
+  appendCompleted,
  clearOpCheckpoint,
  resumeFilter,
  purgeStaleCheckpoints,
@@ -142,6 +143,88 @@ describe('loadOpCheckpoint / recordCompleted / clearOpCheckpoint', () => {
  });
 });

+// #1794: append-only delta storage (op_checkpoint_paths). recordCompleted keeps
+// REPLACE semantics for the 9 non-sync consumers; appendCompleted is the
+// additive path sync uses to avoid O(N²) full-set rewrites.
+describe('appendCompleted (delta) + union read', () => {
+  async function pathRowCount(op: string, fp: string): Promise<number> {
+    const rows = await engine.executeRaw<{ n: string | number }>(
+      `SELECT count(*)::text AS n FROM op_checkpoint_paths WHERE op = $1 AND fingerprint = $2`,
+      [op, fp],
+    );
+    return Number(rows[0]?.n ?? 0);
+  }
+
+  test('appendCompleted returns true and load reflects the delta', async () => {
+    const key = { op: 'sync', fingerprint: 'fp-append' };
+    expect(await appendCompleted(engine, key, ['a.md', 'b.md'])).toBe(true);
+    expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['a.md', 'b.md']);
+  });
+
+  test('re-appending an already-banked path inserts 0 new rows (delta, not full rewrite)', async () => {
+    const key = { op: 'sync', fingerprint: 'fp-delta' };
+    await appendCompleted(engine, key, ['a.md', 'b.md']);
+    expect(await pathRowCount('sync', 'fp-delta')).toBe(2);
+    // Second flush re-sends one banked + one new path; ON CONFLICT DO NOTHING
+    // means only the genuinely-new row lands.
+    await appendCompleted(engine, key, ['b.md', 'c.md']);
+    expect(await pathRowCount('sync', 'fp-delta')).toBe(3);
+    expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['a.md', 'b.md', 'c.md']);
+  });
+
+  test('empty delta is a no-op (returns true, writes nothing)', async () => {
+    const key = { op: 'sync', fingerprint: 'fp-empty' };
+    expect(await appendCompleted(engine, key, [])).toBe(true);
+    expect(await pathRowCount('sync', 'fp-empty')).toBe(0);
+  });
+
+  test('union read across legacy completed_keys array AND appended child rows', async () => {
+    // Simulates an in-flight upgrade: a pre-existing parent row carries the
+    // legacy array, then the new code appends child rows to the same key.
+    const key = { op: 'sync', fingerprint: 'fp-union' };
+    await engine.executeRaw(
+      `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at)
+       VALUES ('sync', 'fp-union', '["legacy-1","legacy-2"]'::jsonb, now())`,
+    );
+    await appendCompleted(engine, key, ['new-1']);
+    expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['legacy-1', 'legacy-2', 'new-1']);
+  });
+
+  test('clearOpCheckpoint cascades to child rows', async () => {
+    const key = { op: 'sync', fingerprint: 'fp-clear' };
+    await appendCompleted(engine, key, ['a.md', 'b.md']);
+    expect(await pathRowCount('sync', 'fp-clear')).toBe(2);
+    await clearOpCheckpoint(engine, key);
+    expect(await pathRowCount('sync', 'fp-clear')).toBe(0);
+    expect(await loadOpCheckpoint(engine, key)).toEqual([]);
+  });
+
+  test('recordCompleted still REPLACES (sync appendCompleted does not)', async () => {
+    // Guards V3: recordCompleted must remove stale keys, not append them.
+    const key = { op: 'embed', fingerprint: 'fp-replace' };
+    await recordCompleted(engine, key, ['x', 'y']);
+    await recordCompleted(engine, key, ['x']);
+    expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['x']);
+  });
+
+  test('purge of a stale parent cascades to its child rows', async () => {
+    // The FK guarantees children always have a parent, so deleting the stale
+    // parent cascade-drops its children. (A standalone orphan is impossible to
+    // create — the FK rejects it — so there is no separate orphan sweep.)
+    await engine.executeRaw(
+      `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at)
+       VALUES ('sync', 'fp-stale', '[]'::jsonb, now() - interval '10 days')`,
+    );
+    await engine.executeRaw(
+      `INSERT INTO op_checkpoint_paths (op, fingerprint, path, created_at)
+       VALUES ('sync', 'fp-stale', 'old.md', now() - interval '10 days')`,
+    );
+    const purged = await purgeStaleCheckpoints(engine, 7);
+    expect(purged).toBe(1); // counts the parent; child cascades silently
+    expect(await pathRowCount('sync', 'fp-stale')).toBe(0);
+  });
+});
+
 describe('resumeFilter (pure)', () => {
  test('empty completed returns all', () => {
    expect(resumeFilter(['a', 'b', 'c'], [])).toEqual(['a', 'b', 'c']);
@@ -93,6 +93,34 @@ describe('isRetryableConnError', () => {
    const err = { problem: 'No database connection', message: 'instance pool torn down' };
    expect(isRetryableConnError(err)).toBe(true);
  });
+
+  // #1794: Supavisor session-pool exhaustion (EMAXCONNSESSION) + Postgres
+  // SQLSTATE 53300 too_many_connections. Transient under load — must retry so
+  // the resumable-sync checkpoint write survives the spike instead of being
+  // dropped (which is how #1794 lost 100% of progress).
+  test('matches EMAXCONNSESSION via message', () => {
+    expect(isRetryableConnError(new Error('EMAXCONNSESSION: max clients in session mode'))).toBe(true);
+  });
+
+  test('matches SQLSTATE 53300 too_many_connections via code', () => {
+    expect(isRetryableConnError(pgError('53300', 'too many connections for role'))).toBe(true);
+  });
+
+  test('matches "too many clients already" message', () => {
+    expect(isRetryableConnError(new Error('sorry, too many clients already'))).toBe(true);
+  });
+
+  test('matches reserved-slots message', () => {
+    expect(
+      isRetryableConnError(new Error('remaining connection slots are reserved for non-replication superuser connections'))
+    ).toBe(true);
+  });
+
+  // 53300 must NOT accidentally widen to other 53xxx (e.g. 53400
+  // configuration_limit_exceeded is not a transient pool blip).
+  test('does NOT match unrelated 53xxx codes', () => {
+    expect(isRetryableConnError(pgError('53400', 'configuration limit exceeded'))).toBe(false);
+  });
 });

 describe('isRetryableError', () => {