fix(open-knowledge): serialize shadow gc against the write path (#2743)

.changeset/shadow-gc-write-race.md

@@ -0,0 +1,5 @@
+---
+"@inkeep/open-knowledge": patch
+---
+
+Background shadow-repo garbage collection no longer races the write path. Previously, the maintenance gc (which packs and prunes the attribution journal's loose objects) could run concurrently with an in-flight auto-save commit; in the race window, git's object-directory cleanup could make the commit fail transiently (`unable to create temporary file` / `failed to insert into database`), logged as a per-writer shadow commit failure and dropping that flush's version-history entry until the next auto-save. Shadow mutations and the gc leg are now mutually exclusive: gc waits for in-flight commits to drain and briefly holds new ones until it finishes, so auto-saves can no longer fail because maintenance happened to run at the same time.

packages/server/src/maintenance-coordinator.test.ts

@@ -140,6 +140,38 @@ describe('MaintenanceCoordinator.runGc (PRD-6972 FR4)', () => {
     expect(head).toBe(concurrentSha);
   }, 60_000);
 
+  // A1 amplification: object creation stays active across gc's ENTIRE window
+  // (a stream of commits, not one). Pre-fix this raced `git add`'s loose-object
+  // writes against gc's object-dir deletion at high probability; post-fix the
+  // shadow op gate serializes the stream against the exclusive gc leg.
+  test('A2: gc is safe against a sustained stream of concurrent commits', async () => {
+    await seedReachableLooseObjects(1500);
+    const coord = createMaintenanceCoordinator({ getShadow: () => shadow });
+
+    const shas: string[] = [];
+    const [gcResult] = await Promise.all([
+      coord.runGc('test'),
+      (async () => {
+        for (let i = 0; i < 20; i++) {
+          writeFileSync(resolve(contentDir, 'intro.md'), `# concurrent edit ${i}\n`);
+          shas.push(await commitWip(shadow, human, 'content/docs', `WIP: during gc ${i}`));
+        }
+      })(),
+    ]);
+
+    expect(gcResult.ran).toBe(true);
+
+    const sg = shadowGit(shadow);
+    const fsck = await sg.raw('fsck', '--full', '--strict');
+    expect(fsck).not.toContain('error');
+    expect(fsck).not.toContain('missing');
+
+    // Every commit in the stream survived; the ref lands on the last one.
+    const head = (await sg.raw('rev-parse', 'refs/wip/main/human-ada')).trim();
+    expect(shas).toHaveLength(20);
+    expect(head).toBe(shas[shas.length - 1]);
+  }, 60_000);
+
   test('detects + surfaces a gc.log latch', async () => {
     // Simulate a prior gc failure: a recent gc.log makes `git gc --auto` decline
     // to run and leaves the latch in place.

packages/server/src/maintenance-coordinator.ts

@@ -20,6 +20,7 @@ import {
   recordGcLatch,
   recordMaintenanceRun,
 } from './shadow-maintenance-telemetry.ts';
+import { shadowOpGateFor } from './shadow-op-gate.ts';
 import type { ShadowHandle, WriterIdentity } from './shadow-repo.ts';
 import {
   enumerateWipChains,
@@ -380,6 +381,18 @@ export class MaintenanceCoordinator {
     const shadow = this.deps.getShadow();
     if (!shadow) return { ran: false, skipped: 'no-shadow' };
 
+    // `git gc` deletes newly-packed loose objects and their emptied fan-out
+    // directories, which races concurrent object writes (`git add` /
+    // `hash-object -w`) into transient failures. Take the shadow op gate
+    // EXCLUSIVELY: wait for in-flight shadow mutators to drain, and hold new
+    // ones until gc completes. The coordinator's `running` flag serializes
+    // maintenance ops against each other; this gate serializes gc against the
+    // write path (see shadow-op-gate.ts).
+    return shadowOpGateFor(shadow).withExclusive(() => this.gcRun(trigger, shadow));
+  }
+
+  /** The actual gc run (caller holds the exclusive shadow-op-gate hold). */
+  private async gcRun(trigger: string, shadow: ShadowHandle): Promise<GcRunResult> {
     const start = performance.now();
     try {
       const before = await countShadowObjects(shadow);

packages/server/src/server-factory.ts

@@ -155,6 +155,7 @@ import {
 import { acquireServerLock, markServerLockDraining, releaseServerLock } from './server-lock.ts';
 import { createServerObserverExtension } from './server-observer-extension.ts';
 import type { PairedWriteOrigin } from './server-observers.ts';
+import { shadowOpGateFor } from './shadow-op-gate.ts';
 import {
   commitUpstreamImport,
   destroyShadowRepo,
@@ -3872,11 +3873,15 @@ export function createServer(options: ServerOptions): ServerInstance {
                   await sg.raw('for-each-ref', `refs/wip/${info.oldBranch}/`, '--format=%(refname)')
                 ).trim();
                 if (refs) {
-                  for (const ref of refs.split('\n')) {
-                    if (ref) {
-                      await sg.raw('update-ref', '-d', ref);
+                  // Ref deletion is a shadow mutation — take the op gate so it
+                  // cannot interleave with a maintenance gc run.
+                  await shadowOpGateFor(shadowRef.current).withMutator(async () => {
+                    for (const ref of refs.split('\n')) {
+                      if (ref) {
+                        await sg.raw('update-ref', '-d', ref);
+                      }
                     }
-                  }
+                  });
                   log.info(
                     { context: info.oldBranch },
                     `[branch-switch] cleaned up detached context ${info.oldBranch}`,

packages/server/src/shadow-op-gate.test.ts

@@ -0,0 +1,216 @@
+import { describe, expect, test } from 'bun:test';
+import { releaseShadowOpGate, ShadowOpGate, shadowOpGateFor } from './shadow-op-gate.ts';
+
+/** A promise whose settle order we can observe alongside manual triggers. */
+function deferred(): { promise: Promise<void>; resolve: () => void } {
+  let resolve!: () => void;
+  const promise = new Promise<void>((r) => {
+    resolve = r;
+  });
+  return { promise, resolve };
+}
+
+const tick = () => new Promise((r) => setTimeout(r, 0));
+
+describe('ShadowOpGate (forced interleavings, deterministic)', () => {
+  test('mutators run concurrently with each other', async () => {
+    const gate = new ShadowOpGate();
+    const a = deferred();
+    const b = deferred();
+    let aStarted = false;
+    let bStarted = false;
+    const pa = gate.withMutator(async () => {
+      aStarted = true;
+      await a.promise;
+    });
+    const pb = gate.withMutator(async () => {
+      bStarted = true;
+      await b.promise;
+    });
+    await tick();
+    expect(aStarted).toBe(true);
+    expect(bStarted).toBe(true);
+    expect(gate.activeMutators).toBe(2);
+    a.resolve();
+    b.resolve();
+    await Promise.all([pa, pb]);
+    expect(gate.activeMutators).toBe(0);
+  });
+
+  test('exclusive waits for an in-flight mutator to drain before running', async () => {
+    const gate = new ShadowOpGate();
+    const hold = deferred();
+    let gcRan = false;
+    const mutator = gate.withMutator(() => hold.promise);
+    await tick();
+    const gc = gate.withExclusive(async () => {
+      gcRan = true;
+    });
+    await tick();
+    // Mutator still holds — gc must not have started.
+    expect(gcRan).toBe(false);
+    expect(gate.isExclusiveHeld).toBe(false);
+    hold.resolve();
+    await gc;
+    expect(gcRan).toBe(true);
+    await mutator;
+  });
+
+  test('mutators queue while exclusive holds, run after release', async () => {
+    const gate = new ShadowOpGate();
+    const gcHold = deferred();
+    let mutatorRan = false;
+    const gc = gate.withExclusive(() => gcHold.promise);
+    await tick();
+    expect(gate.isExclusiveHeld).toBe(true);
+    const mutator = gate.withMutator(async () => {
+      mutatorRan = true;
+    });
+    await tick();
+    // gc still holds — the mutator must be queued, not running.
+    expect(mutatorRan).toBe(false);
+    gcHold.resolve();
+    await gc;
+    await mutator;
+    expect(mutatorRan).toBe(true);
+  });
+
+  test('new top-level mutator is not blocked by a waiting (not holding) exclusive', async () => {
+    const gate = new ShadowOpGate();
+    const mutatorHold = deferred();
+    const firstMutator = gate.withMutator(() => mutatorHold.promise);
+    await tick();
+    let gcRan = false;
+    const gc = gate.withExclusive(async () => {
+      gcRan = true;
+    });
+    await tick();
+    expect(gate.isExclusiveHeld).toBe(false); // gc is waiting, not holding
+    let newMutatorRan = false;
+    const newMutator = gate.withMutator(async () => {
+      newMutatorRan = true;
+    });
+    await tick();
+    // No barging: the new top-level mutator must NOT queue behind the waiting gc.
+    expect(newMutatorRan).toBe(true);
+    mutatorHold.resolve();
+    await Promise.all([firstMutator, newMutator, gc]);
+    expect(gcRan).toBe(true);
+  });
+
+  test('nested mutator holds never deadlock against a waiting exclusive', async () => {
+    const gate = new ShadowOpGate();
+    const outerHold = deferred();
+    let innerRan = false;
+    const outer = gate.withMutator(async () => {
+      await outerHold.promise;
+      // Nested acquisition while an exclusive is WAITING (not holding): must be
+      // granted immediately — a pending exclusive does not block new mutators.
+      await gate.withMutator(async () => {
+        innerRan = true;
+      });
+    });
+    await tick();
+    let gcRan = false;
+    const gc = gate.withExclusive(async () => {
+      gcRan = true;
+    });
+    await tick();
+    expect(gcRan).toBe(false); // waiting for the outer mutator to drain
+    outerHold.resolve();
+    await outer;
+    expect(innerRan).toBe(true);
+    await gc;
+    expect(gcRan).toBe(true);
+  });
+
+  test('exclusives serialize against each other', async () => {
+    const gate = new ShadowOpGate();
+    const firstHold = deferred();
+    const order: string[] = [];
+    const first = gate.withExclusive(async () => {
+      order.push('first-start');
+      await firstHold.promise;
+      order.push('first-end');
+    });
+    await tick();
+    const second = gate.withExclusive(async () => {
+      order.push('second-start');
+    });
+    await tick();
+    expect(order).toEqual(['first-start']);
+    firstHold.resolve();
+    await Promise.all([first, second]);
+    expect(order).toEqual(['first-start', 'first-end', 'second-start']);
+  });
+
+  test('two exclusives queued behind one mutator still serialize', async () => {
+    const gate = new ShadowOpGate();
+    const mutatorHold = deferred();
+    const firstHold = deferred();
+    const order: string[] = [];
+    const mutator = gate.withMutator(() => mutatorHold.promise);
+    await tick();
+    // Both exclusives now wait on the SAME drain event.
+    const first = gate.withExclusive(async () => {
+      order.push('first-start');
+      await firstHold.promise;
+      order.push('first-end');
+    });
+    const second = gate.withExclusive(async () => {
+      order.push('second-start');
+    });
+    await tick();
+    expect(order).toEqual([]);
+    mutatorHold.resolve();
+    await mutator;
+    await tick();
+    // Exactly one exclusive acquired; the other re-queued behind it.
+    expect(order).toEqual(['first-start']);
+    firstHold.resolve();
+    await Promise.all([first, second]);
+    expect(order).toEqual(['first-start', 'first-end', 'second-start']);
+  });
+
+  test('mutator errors release the hold (exclusive still proceeds)', async () => {
+    const gate = new ShadowOpGate();
+    await expect(
+      gate.withMutator(async () => {
+        throw new Error('boom');
+      }),
+    ).rejects.toThrow('boom');
+    expect(gate.activeMutators).toBe(0);
+    let gcRan = false;
+    await gate.withExclusive(async () => {
+      gcRan = true;
+    });
+    expect(gcRan).toBe(true);
+  });
+
+  test('exclusive errors release the hold (mutators still proceed)', async () => {
+    const gate = new ShadowOpGate();
+    await expect(
+      gate.withExclusive(async () => {
+        throw new Error('gc-boom');
+      }),
+    ).rejects.toThrow('gc-boom');
+    expect(gate.isExclusiveHeld).toBe(false);
+    let mutatorRan = false;
+    await gate.withMutator(async () => {
+      mutatorRan = true;
+    });
+    expect(mutatorRan).toBe(true);
+  });
+
+  test('registry: same gitDir shares one gate; release drops it', () => {
+    const a = shadowOpGateFor({ gitDir: '/tmp/gate-test-a' });
+    const b = shadowOpGateFor({ gitDir: '/tmp/gate-test-a' });
+    const c = shadowOpGateFor({ gitDir: '/tmp/gate-test-c' });
+    expect(a).toBe(b);
+    expect(a).not.toBe(c);
+    releaseShadowOpGate('/tmp/gate-test-a');
+    expect(shadowOpGateFor({ gitDir: '/tmp/gate-test-a' })).not.toBe(a);
+    releaseShadowOpGate('/tmp/gate-test-a');
+    releaseShadowOpGate('/tmp/gate-test-c');
+  });
+});