From c74cf9b85d4bee5608a9d06821dfa25d61b7d980 Mon Sep 17 00:00:00 2001 From: Nick Gomez <122398915+nick-inkeep@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:06:28 -0700 Subject: [PATCH] fix(open-knowledge): serialize shadow gc against the write path (#2743) * fix(server): serialize shadow gc against the write path via shadow op gate git gc deletes newly-packed loose objects and emptied objects/xx fan-out dirs; a concurrent git add / hash-object -w writing into those dirs fails transiently (EINVAL observed on macOS, escaping git's ENOENT-only retry). The maintenance coordinator serialized maintenance legs against each other but not against shadow mutators, so a background gc could fail an in-flight flush commit (the A1 flake, 6-14 percent per run measured). Add ShadowOpGate (per shadow gitDir): mutators take shared holds, the gc leg takes an exclusive hold that drains in-flight mutators and briefly queues new ones. Deadlock-free by construction (a pending exclusive never blocks new shared entries, so nested mutator holds cannot deadlock). Wrap every shadow mutation entry point; A2 pins gc against a sustained commit stream. * test(server): review-response hardening for the shadow op gate Address claude-review suggestions on #2743: warn when an exclusive acquisition waits >60s on the mutator drain (starvation observability), document buildWipTree's reliance on the prune grace window across its split gate hold, count actual (not attempted) legacy-ref deletions in sweepLegacyShadowRefs, and add the no-barging top-level-mutator and exclusive-error-release gate tests. GitOrigin-RevId: 90f22c66540c9ce6cf03b7d7ba13f99c000af66d --- .changeset/shadow-gc-write-race.md | 5 + .../src/maintenance-coordinator.test.ts | 32 +++ .../server/src/maintenance-coordinator.ts | 13 ++ packages/server/src/server-factory.ts | 13 +- packages/server/src/shadow-op-gate.test.ts | 216 ++++++++++++++++++ packages/server/src/shadow-op-gate.ts | 141 ++++++++++++ packages/server/src/shadow-repo.ts | 68 +++++- 7 files changed, 472 insertions(+), 16 deletions(-) create mode 100644 .changeset/shadow-gc-write-race.md create mode 100644 packages/server/src/shadow-op-gate.test.ts create mode 100644 packages/server/src/shadow-op-gate.ts diff --git a/.changeset/shadow-gc-write-race.md b/.changeset/shadow-gc-write-race.md new file mode 100644 index 000000000..7cc9c2ae1 --- /dev/null +++ b/.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. diff --git a/packages/server/src/maintenance-coordinator.test.ts b/packages/server/src/maintenance-coordinator.test.ts index 3d967de66..3b3cdf7e5 100644 --- a/packages/server/src/maintenance-coordinator.test.ts +++ b/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. diff --git a/packages/server/src/maintenance-coordinator.ts b/packages/server/src/maintenance-coordinator.ts index 934c7c0e7..cdd402076 100644 --- a/packages/server/src/maintenance-coordinator.ts +++ b/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 { const start = performance.now(); try { const before = await countShadowObjects(shadow); diff --git a/packages/server/src/server-factory.ts b/packages/server/src/server-factory.ts index d94e0499d..7413d3a22 100644 --- a/packages/server/src/server-factory.ts +++ b/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}`, diff --git a/packages/server/src/shadow-op-gate.test.ts b/packages/server/src/shadow-op-gate.test.ts new file mode 100644 index 000000000..92a19ddd5 --- /dev/null +++ b/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; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((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'); + }); +}); diff --git a/packages/server/src/shadow-op-gate.ts b/packages/server/src/shadow-op-gate.ts new file mode 100644 index 000000000..0b0d51dce --- /dev/null +++ b/packages/server/src/shadow-op-gate.ts @@ -0,0 +1,141 @@ +/** + * Shadow-repo operation gate — mutual exclusion between shadow git MUTATIONS + * (object/ref writes: commitWip, buildWipTree, saveVersion, checkpoints, park, + * ref sweeps) and the maintenance GC leg. + * + * Why: `git gc` (repack + prune/prune-packed) deletes newly-packed loose + * objects and removes now-empty `objects/xx/` fan-out directories. A concurrent + * `git add`/`hash-object -w` creates its temporary object file inside those + * same directories; when gc removes a directory in the race window the write + * fails (`unable to create temporary file` → `fatal: updating files failed`). + * git only retries its internal mkdir fallback on ENOENT, and the raced state + * can surface other errnos (EINVAL observed on macOS), so the failure escapes + * to the caller — in production, a transiently failed user-content flush + * commit. The MaintenanceCoordinator's own gate serializes maintenance ops + * against each other but not against the write path; this gate closes that gap. + * + * Semantics (deadlock-free by construction): + * - `withMutator` (shared): waits only while an exclusive op HOLDS the gate, + * then runs concurrently with other mutators. A *pending* exclusive request + * does NOT block new mutators, so nested mutator calls (e.g. saveVersion + * internally driving further ref updates) can never deadlock against a + * waiting gc. Shared holders never wait while holding. + * - `withExclusive` (gc): serializes against other exclusives, then waits for + * in-flight mutators to drain to zero, then holds the gate — new mutators + * queue until it releases. The exclusive holder never waits while holding. + * + * Trade-offs: gc can be starved under a continuous mutator stream — acceptable + * because flush commits are debounced/bursty and gc is opportunistic + * (skip-and-retry-next-trigger posture). Mutators can queue behind a long gc; + * shadow flushes are background persistence (debounced off the CRDT hot path), + * so a paused flush delays durability, not user-visible editing. + * + * Gates are registered per shadow gitDir (not per handle object) so every + * handle pointing at the same shadow repo shares one gate, including handles + * recreated across boots within one process. + */ + +import { getLogger } from './logger.ts'; + +const log = getLogger('shadow-op-gate'); + +/** + * Warn when an exclusive (gc) acquisition waited longer than this for the + * mutator drain. Purely diagnostic — the documented starvation posture is + * unchanged; this makes it observable if the production write profile turns + * out to be less bursty than assumed. + */ +const EXCLUSIVE_WAIT_WARN_MS = 60_000; + +export class ShadowOpGate { + private mutatorCount = 0; + /** Non-null exactly while an exclusive op holds the gate. */ + private exclusiveHeld: Promise | null = null; + private drainWaiters: Array<() => void> = []; + + /** Number of in-flight mutator holds (diagnostics/tests). */ + get activeMutators(): number { + return this.mutatorCount; + } + + /** True while an exclusive (gc) op holds the gate (diagnostics/tests). */ + get isExclusiveHeld(): boolean { + return this.exclusiveHeld !== null; + } + + async withMutator(fn: () => Promise): Promise { + // Wait only while an exclusive op HOLDS the gate. Re-check after each wake: + // another exclusive may have acquired between the release and this + // continuation running. + while (this.exclusiveHeld) await this.exclusiveHeld; + this.mutatorCount += 1; + try { + return await fn(); + } finally { + this.mutatorCount -= 1; + if (this.mutatorCount === 0) { + for (const wake of this.drainWaiters.splice(0)) wake(); + } + } + } + + async withExclusive(fn: () => Promise): Promise { + // Acquire: no exclusive holder AND zero in-flight mutators. Both + // conditions re-check after every wake in ONE loop — a wake from either + // wait may race another exclusive acquiring first (the coordinator's own + // gate already ensures one maintenance op at a time; this keeps the gate + // safe standalone). New mutators may still enter while we wait for the + // drain (no barging — that is what makes nested mutator holds + // deadlock-free). + const waitStart = performance.now(); + for (;;) { + if (this.exclusiveHeld) { + await this.exclusiveHeld; + continue; + } + if (this.mutatorCount > 0) { + await new Promise((r) => { + this.drainWaiters.push(r); + }); + continue; + } + break; + } + const waitedMs = performance.now() - waitStart; + if (waitedMs > EXCLUSIVE_WAIT_WARN_MS) { + log.warn( + { waitedMs: Math.round(waitedMs) }, + '[shadow-op-gate] exclusive acquisition waited a long time for the mutator drain — sustained write stream is starving maintenance', + ); + } + // Both conditions held on a synchronous check just now — acquire before + // any other task can observe the gate un-held. + let release!: () => void; + this.exclusiveHeld = new Promise((r) => { + release = r; + }); + try { + return await fn(); + } finally { + this.exclusiveHeld = null; + release(); + } + } +} + +const gates = new Map(); + +/** The shared per-shadow-repo gate, keyed by the shadow's gitDir. */ +export function shadowOpGateFor(shadow: { gitDir: string }): ShadowOpGate { + let gate = gates.get(shadow.gitDir); + if (!gate) { + gate = new ShadowOpGate(); + gates.set(shadow.gitDir, gate); + } + return gate; +} + +/** Drop the registry entry for a shadow repo (graceful shutdown / tests). */ +export function releaseShadowOpGate(gitDir: string): void { + gates.delete(gitDir); +} diff --git a/packages/server/src/shadow-repo.ts b/packages/server/src/shadow-repo.ts index de2542651..14e96ab69 100644 --- a/packages/server/src/shadow-repo.ts +++ b/packages/server/src/shadow-repo.ts @@ -46,6 +46,7 @@ import { tracedMkdirSync, tracedRenameSync, tracedWriteFileSync } from './fs-tra import { listTreeLongEntries } from './git-paths.ts'; import { incrementShadowMigrationLegacyRefsDeleted } from './metrics.ts'; import { acquireLock, releaseLock } from './shadow-lock.ts'; +import { releaseShadowOpGate, shadowOpGateFor } from './shadow-op-gate.ts'; import { withSpan } from './telemetry.ts'; // ─── Types ─────────────────────────────────────────────────────────────────── @@ -272,6 +273,7 @@ export async function initShadowRepo(projectRoot: string): Promise */ export function destroyShadowRepo(shadow: ShadowHandle): void { releaseLock(shadow.gitDir); + releaseShadowOpGate(shadow.gitDir); } /** @@ -328,15 +330,17 @@ export async function sweepLegacyShadowRefs(shadow: ShadowHandle): Promise { + for (const ref of toDelete) { + try { + await sg.raw('update-ref', '-d', ref); + deleted++; + } catch (e) { + console.warn(`[shadow-migration] failed to delete legacy ref ${ref}:`, e); + } } - } - - const deleted = toDelete.length; + }); incrementShadowMigrationLegacyRefsDeleted(deleted); console.warn( `[shadow-migration] deleted ${deleted} legacy refs: server=${breakdown.server} human-=${breakdown['human-']} upstream=${breakdown.upstream}`, @@ -383,7 +387,10 @@ export async function commitWip( 'shadow.branch': branch, }, }, - async () => commitWipInner(shadow, writer, contentRoot, message, branch, opts?.date), + async () => + shadowOpGateFor(shadow).withMutator(() => + commitWipInner(shadow, writer, contentRoot, message, branch, opts?.date), + ), ); } @@ -502,6 +509,16 @@ function sweepOrphanedTmpIndexFiles(shadow: ShadowHandle): number { } export async function buildWipTree(shadow: ShadowHandle, contentRoot: string): Promise { + // The returned tree is dangling until a later commitWipFromTree references + // it, and that call takes a SEPARATE gate hold — a gc can interleave between + // the two. Safe today because gc.pruneExpire is left at git's 2-week default, + // so a milliseconds-old dangling tree is far inside the prune grace window. + // If prune config is ever tightened toward `now`, the flush fan-out must + // instead span tree-build + per-writer commits under ONE mutator hold. + return shadowOpGateFor(shadow).withMutator(() => buildWipTreeInner(shadow, contentRoot)); +} + +async function buildWipTreeInner(shadow: ShadowHandle, contentRoot: string): Promise { const tmpIndex = resolve(shadow.gitDir, `index-wip-fanout-${randomUUID()}`); const sg = shadowGit(shadow); const gitPathspec = contentRoot || '.'; @@ -546,7 +563,10 @@ export async function commitWipFromTree( 'shadow.tree': treeSha.slice(0, 8), }, }, - async () => commitWipFromTreeInner(shadow, writer, treeSha, message, branch), + async () => + shadowOpGateFor(shadow).withMutator(() => + commitWipFromTreeInner(shadow, writer, treeSha, message, branch), + ), ); } @@ -790,6 +810,16 @@ export async function saveInMemoryCheckpoint( shadow: ShadowHandle, contentRoot: string, params: InMemoryCheckpointParams, +): Promise { + return shadowOpGateFor(shadow).withMutator(() => + saveInMemoryCheckpointInner(shadow, contentRoot, params), + ); +} + +async function saveInMemoryCheckpointInner( + shadow: ShadowHandle, + contentRoot: string, + params: InMemoryCheckpointParams, ): Promise { const branch = params.branch ?? 'main'; const sg = shadowGit(shadow); @@ -1082,6 +1112,14 @@ export async function gcCheckpointRefs( shadow: ShadowHandle, branch = 'main', policy: CheckpointRetentionPolicy = DEFAULT_CHECKPOINT_RETENTION, +): Promise { + return shadowOpGateFor(shadow).withMutator(() => gcCheckpointRefsInner(shadow, branch, policy)); +} + +async function gcCheckpointRefsInner( + shadow: ShadowHandle, + branch = 'main', + policy: CheckpointRetentionPolicy = DEFAULT_CHECKPOINT_RETENTION, ): Promise { const result: CheckpointGcResult = { scanned: 0, @@ -1285,7 +1323,10 @@ export async function parkBranch( 'shadow.doc_count': documents.length, }, }, - async () => parkBranchInner(shadow, branch, writerId, documents, newBranch), + async () => + shadowOpGateFor(shadow).withMutator(() => + parkBranchInner(shadow, branch, writerId, documents, newBranch), + ), ); } @@ -1490,7 +1531,10 @@ export async function saveVersion( 'shadow.checkpoint_kind': options?.checkpointKind ? 'auto-consolidation' : 'user', }, }, - async () => saveVersionInner(shadow, contentRoot, writers, branch, summary, options), + async () => + shadowOpGateFor(shadow).withMutator(() => + saveVersionInner(shadow, contentRoot, writers, branch, summary, options), + ), ); }