diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 28860b2c2b..723138fda9 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -257,6 +257,7 @@ import { import { runSyncOnConnect, SyncOnConnectPostSyncError, type SyncOnConnectOutcome, type SyncOnConnectPeerOutcome } from './sync/on-connect/sync-on-connect.js'; import { mapWithConcurrency } from './map-with-concurrency.js'; import { CATCHUP_MAX_CONCURRENT_PEER_SYNCS } from './sync/catchup-concurrency.js'; +import { retryCatchupPlaneOnBackpressure } from './sync/catchup-backpressure-retry.js'; import { classifyDurableProgress } from './sync/durable-progress.js'; import { getSyncBackpressureSnapshot, @@ -607,12 +608,16 @@ function contextGraphCatchupSingleFlightKey(params: { includeSharedMemory: boolean; maxPeers?: number; peerRotationKey?: string; + priority?: number; + retryDeferredBackpressure?: boolean; }): string { return syncSingleFlightKey('context-graph-catchup', { contextGraphId: params.contextGraphId, includeSharedMemory: params.includeSharedMemory, maxPeers: normalizedCatchupMaxPeers(params.maxPeers), peerRotationKey: params.peerRotationKey ?? null, + priority: params.priority ?? null, + retryDeferredBackpressure: params.retryDeferredBackpressure === true, }); } @@ -626,6 +631,7 @@ function durableSyncSingleFlightKey(params: { hasAccessDeniedCallback: boolean; hasSinceBatchIdResolver: boolean; exactAssetUals?: readonly string[]; + priority?: number; }): string | null { if (params.hasPhaseCallback || params.hasAccessDeniedCallback || params.hasSinceBatchIdResolver) { return null; @@ -637,6 +643,7 @@ function durableSyncSingleFlightKey(params: { totalTimeoutMs: params.totalTimeoutMs, syncAgentsMeta: params.syncAgentsMeta, exactAssetUals: params.exactAssetUals ?? null, + priority: params.priority ?? null, }); } @@ -646,6 +653,7 @@ function sharedMemorySyncSingleFlightKey(params: { stopOnBackoffWorthyFailure?: boolean; publicContextGraphIds: readonly string[]; privateRecoverFromCurator: readonly string[]; + priority?: number; }): string { return syncSingleFlightKey('shared-memory-sync', { remotePeerId: params.remotePeerId, @@ -653,6 +661,7 @@ function sharedMemorySyncSingleFlightKey(params: { stopOnBackoffWorthyFailure: params.stopOnBackoffWorthyFailure === true, publicContextGraphIds: params.publicContextGraphIds, privateRecoverFromCurator: params.privateRecoverFromCurator, + priority: params.priority ?? null, }); } @@ -4042,7 +4051,13 @@ export class LifecycleSyncMethods extends DKGAgentBase { // the same flag that makes it a responder. Same signal SC4 uses to advertise the protocol. if (asChangelogReader(this.store) !== null && contextGraphIds.length > 0) { try { - const lane = await this.runChangelogLane(ctx, remotePeerId, contextGraphIds, onAccessDenied); + const lane = await this.runChangelogLane( + ctx, + remotePeerId, + contextGraphIds, + onAccessDenied, + options?.priority, + ); changelogResult = lane.result; legacyContextGraphIds = lane.remainingLegacyCgs; if ((changelogResult.deferredBackpressure ?? 0) > 0) return changelogResult; @@ -4142,6 +4157,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { hasAccessDeniedCallback: Boolean(onAccessDenied), hasSinceBatchIdResolver: Boolean(sinceBatchIdFor), exactAssetUals: options?.exactAssetUals, + priority: options?.priority, }); return singleFlightKey ? runSyncSingleFlight(this, singleFlightKey, runSync) : runSync(); } @@ -4293,6 +4309,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { remotePeerId: string, contextGraphIds: string[], onAccessDenied?: (contextGraphId: string) => void, + priority?: number, ): Promise<{ result: DurableSyncResult; remainingLegacyCgs: string[] }> { const peerProtocols = await this.getPeerProtocols(remotePeerId); if (!peerProtocols.includes(PROTOCOL_SYNC_CHANGELOG)) { @@ -4333,6 +4350,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { item.lane, item.operationId, run, + priority, ), merge: mergeDurableSyncResults, markDeferred: (summary) => ({ @@ -4761,6 +4779,8 @@ export class LifecycleSyncMethods extends DKGAgentBase { options?: { stopOnBackoffWorthyFailure?: boolean; sharedMemorySyncPlan?: SharedMemorySyncContextGraphPlan; + /** Admission override for foreground catch-up. */ + priority?: number; }, ): Promise { const ctx = createOperationContext('sync'); @@ -4827,6 +4847,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { stopOnBackoffWorthyFailure, publicContextGraphIds, privateRecoverFromCurator, + priority: options?.priority, }); const runSync = async (): Promise => { @@ -4980,6 +5001,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { item.lane, item.operationId, run, + options?.priority, ), merge: mergeSharedMemorySyncResults, markDeferred: (summary) => ({ @@ -5084,7 +5106,15 @@ export class LifecycleSyncMethods extends DKGAgentBase { */ async syncContextGraphFromConnectedPeers(this: DKGAgent, contextGraphId: string, - options?: { includeSharedMemory?: boolean; maxPeers?: number; peerRotationKey?: string }, + options?: { + includeSharedMemory?: boolean; + maxPeers?: number; + peerRotationKey?: string; + /** Admission override used by explicit foreground catch-up callers. */ + priority?: number; + /** Retry only locally-deferred planes; completed planes are not rerun. */ + retryDeferredBackpressure?: boolean; + }, ): Promise<{ /** Ordered connected peers before optional maxPeers windowing. */ connectedPeers: number; @@ -5137,6 +5167,8 @@ export class LifecycleSyncMethods extends DKGAgentBase { includeSharedMemory, maxPeers: options?.maxPeers, peerRotationKey: options?.peerRotationKey, + priority: options?.priority, + retryDeferredBackpressure: options?.retryDeferredBackpressure, }); return runSyncSingleFlight(this, singleFlightKey, async (): Promise => { @@ -5186,6 +5218,8 @@ export class LifecycleSyncMethods extends DKGAgentBase { ); return this.runCatchupOverPeers(contextGraphId, includeSharedMemory, peers, { totalPeers: orderedPeers.length, + priority: options?.priority, + retryDeferredBackpressure: options?.retryDeferredBackpressure, }); }); } @@ -5276,7 +5310,11 @@ export class LifecycleSyncMethods extends DKGAgentBase { contextGraphId: string, includeSharedMemory: boolean, peers: Array<{ toString(): string }>, - stats?: { totalPeers?: number }, + stats?: { + totalPeers?: number; + priority?: number; + retryDeferredBackpressure?: boolean; + }, ): Promise<{ /** Ordered connected peers before optional caller windowing. */ connectedPeers: number; @@ -5432,13 +5470,37 @@ export class LifecycleSyncMethods extends DKGAgentBase { syncCapable, CATCHUP_MAX_CONCURRENT_PEER_SYNCS, async (remotePeerId) => { - const durable = await this.syncFromPeerDetailed( + const runDurable = () => this.syncFromPeerDetailed( remotePeerId, [contextGraphId], + undefined, + undefined, + undefined, + stats?.priority === undefined ? undefined : { priority: stats.priority }, + ); + const durable = await ( + stats?.retryDeferredBackpressure + ? retryCatchupPlaneOnBackpressure(runDurable) + : runDurable() ).catch(emptyDurable); - const shared = includeSharedMemory - ? await this.syncSharedMemoryFromPeerDetailed(remotePeerId, [contextGraphId]).catch(emptyShared) - : null; + + // SWM authorization/materialization depends on durable metadata. If + // durable admission remains deferred, do not manufacture a premature + // SWM denial. Once durable completes, retry only SWM; a successful VM + // plane is never fetched again just because SWM hit local pressure. + let shared: SharedMemorySyncResult | null = null; + if (includeSharedMemory && (durable.deferredBackpressure ?? 0) === 0) { + const runShared = () => this.syncSharedMemoryFromPeerDetailed( + remotePeerId, + [contextGraphId], + stats?.priority === undefined ? undefined : { priority: stats.priority }, + ); + shared = await ( + stats?.retryDeferredBackpressure + ? retryCatchupPlaneOnBackpressure(runShared) + : runShared() + ).catch(emptyShared); + } return { durable, shared }; }, ); diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index a8496bf5ea..61c80cab4f 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -269,6 +269,12 @@ export { // deep-importing the compiled `dist/` module. export { mapWithConcurrency } from './map-with-concurrency.js'; export { CATCHUP_MAX_CONCURRENT_PEER_SYNCS } from './sync/catchup-concurrency.js'; +export { + CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + retryCatchupPlaneOnBackpressure, + type CatchupBackpressureResult, +} from './sync/catchup-backpressure-retry.js'; export { classifyDurableProgress, type DurableProgressClassification, diff --git a/packages/agent/src/sync/catchup-backpressure-retry.ts b/packages/agent/src/sync/catchup-backpressure-retry.ts new file mode 100644 index 0000000000..7b345a1b8d --- /dev/null +++ b/packages/agent/src/sync/catchup-backpressure-retry.ts @@ -0,0 +1,39 @@ +/** + * User-requested catch-up must outrank autonomous exact-VM repair (priority + * 1_000) and ordinary background sync (priority 0). This lets a subscribe or + * explicit catch-up displace queued background work instead of being marked + * deferred before it has fetched a byte. + */ +export const FOREGROUND_CATCHUP_SYNC_PRIORITY = 2_000; + +/** + * Admission can still race another foreground catch-up. Retry that local-only + * outcome briefly; transport, authorization, timeout, and integrity failures + * are deliberately not retried here. + */ +export const CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS = [100, 250, 500] as const; + +export interface CatchupBackpressureResult { + deferredBackpressure?: number; +} + +export async function retryCatchupPlaneOnBackpressure( + run: () => Promise, + options?: { + delaysMs?: readonly number[]; + wait?: (delayMs: number) => Promise; + }, +): Promise { + const delaysMs = options?.delaysMs ?? CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS; + const wait = options?.wait ?? ((delayMs: number) => new Promise((resolve) => { + setTimeout(resolve, delayMs); + })); + + let result = await run(); + for (const delayMs of delaysMs) { + if ((result.deferredBackpressure ?? 0) === 0) break; + await wait(delayMs); + result = await run(); + } + return result; +} diff --git a/packages/agent/test/catchup-backpressure-retry.test.ts b/packages/agent/test/catchup-backpressure-retry.test.ts new file mode 100644 index 0000000000..311811f434 --- /dev/null +++ b/packages/agent/test/catchup-backpressure-retry.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS, + retryCatchupPlaneOnBackpressure, +} from '../src/sync/catchup-backpressure-retry.js'; + +describe('retryCatchupPlaneOnBackpressure', () => { + it('retries only the local scheduler deferral result', async () => { + const run = vi.fn() + .mockResolvedValueOnce({ deferredBackpressure: 1, marker: 'deferred' }) + .mockResolvedValueOnce({ deferredBackpressure: 0, marker: 'complete' }); + const waits: number[] = []; + + const result = await retryCatchupPlaneOnBackpressure(run, { + delaysMs: [3, 5], + wait: async (delayMs) => { waits.push(delayMs); }, + }); + + expect(result).toEqual({ deferredBackpressure: 0, marker: 'complete' }); + expect(run).toHaveBeenCalledTimes(2); + expect(waits).toEqual([3]); + }); + + it('returns the final deferred result after the bounded retry budget', async () => { + const run = vi.fn(async () => ({ deferredBackpressure: 1 })); + + const result = await retryCatchupPlaneOnBackpressure(run, { + wait: async () => {}, + }); + + expect(result.deferredBackpressure).toBe(1); + expect(run).toHaveBeenCalledTimes(CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS.length + 1); + }); + + it('does not retry a clean result', async () => { + const run = vi.fn(async () => ({ deferredBackpressure: 0 })); + + await retryCatchupPlaneOnBackpressure(run, { + wait: async () => { throw new Error('must not wait'); }, + }); + + expect(run).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/agent/test/sync-requester-priority.test.ts b/packages/agent/test/sync-requester-priority.test.ts index 712507dbc8..43c451a50b 100644 --- a/packages/agent/test/sync-requester-priority.test.ts +++ b/packages/agent/test/sync-requester-priority.test.ts @@ -54,7 +54,11 @@ function durableContext(contextGraphIds: string[]) { function lifecycleAgent( priorities: Record, - runAdmission: (contextGraphId: string, work: () => Promise) => Promise, + runAdmission: ( + contextGraphId: string, + work: () => Promise, + priorityOverride?: number, + ) => Promise, ) { return { config: { syncContextGraphPriorities: priorities }, @@ -95,7 +99,8 @@ function lifecycleAgent( _lane: string, _label: string, work: () => Promise, - ) => runAdmission(contextGraphId, work), + priorityOverride?: number, + ) => runAdmission(contextGraphId, work, priorityOverride), log: { info: noop, warn: noop, debug: noop }, }; } @@ -121,6 +126,27 @@ describe('requester per-CG priority admission', () => { expect(admissions).toEqual(['high', 'default', 'low']); }); + it('passes a foreground durable priority override through admission', async () => { + const priorityOverrides: Array = []; + const agent = lifecycleAgent({}, async (_contextGraphId, work, priorityOverride) => { + priorityOverrides.push(priorityOverride); + return work(); + }); + + await (LifecycleSyncMethods.prototype.runLegacyDurableSync as any).call( + agent, + ctx, + 'peer', + ['foreground'], + undefined, + undefined, + undefined, + { priority: 2_000 }, + ); + + expect(priorityOverrides).toEqual([2_000]); + }); + it('preserves completed durable progress when a later admission is deferred', async () => { const admissions: string[] = []; const agent = lifecycleAgent({}, async (contextGraphId, work) => { @@ -148,6 +174,7 @@ describe('requester per-CG priority admission', () => { it('marks changelog admission pressure deferred without routing that graph to legacy', async () => { const admissions: string[] = []; + const priorityOverrides: Array = []; const emptyResult = { insertedTriples: 0, fetchedMetaTriples: 0, @@ -179,8 +206,10 @@ describe('requester per-CG priority admission', () => { _lane: string, _label: string, work: () => Promise, + priorityOverride?: number, ) => { admissions.push(contextGraphId); + priorityOverrides.push(priorityOverride); if (contextGraphId === 'second') { throw new SyncBackpressureBusyError('queue full'); } @@ -195,16 +224,20 @@ describe('requester per-CG priority admission', () => { ctx, 'peer', ['first', 'second', 'third'], + undefined, + 2_000, ); expect(admissions).toEqual(['first', 'second']); expect(lane.result.completedPhases).toBe(1); expect(lane.result.deferredBackpressure).toBe(1); expect(lane.remainingLegacyCgs).toEqual([]); + expect(priorityOverrides).toEqual([2_000, 2_000]); }); it('preserves completed shared-memory progress when a later admission is deferred', async () => { const admissions: string[] = []; + const priorityOverrides: Array = []; const warnings: string[] = []; const contextGraphIds = ['first', 'second', 'third']; const agent = { @@ -239,8 +272,10 @@ describe('requester per-CG priority admission', () => { _lane: string, _label: string, work: () => Promise, + priorityOverride?: number, ) => { admissions.push(contextGraphId); + priorityOverrides.push(priorityOverride); if (contextGraphId === 'second') { throw new SyncBackpressureBusyError('queue full'); } @@ -263,6 +298,7 @@ describe('requester per-CG priority admission', () => { privateRecoverFromCurator: [], eligibleContextGraphIds: contextGraphIds, }, + priority: 2_000, }); expect(admissions).toEqual(['first', 'second']); @@ -272,6 +308,7 @@ describe('requester per-CG priority admission', () => { expect(summary.deferredBackpressure).toBe(1); expect(summary.failedPeers).toBe(0); expect(summary.backoffWorthyFailures).toBe(0); + expect(priorityOverrides).toEqual([2_000, 2_000]); }); it('counts several failed Context Graphs from one remote as one failed peer', async () => { diff --git a/packages/cli/src/catchup-runner-worker-impl.ts b/packages/cli/src/catchup-runner-worker-impl.ts index af66fb1d70..013783ddbf 100644 --- a/packages/cli/src/catchup-runner-worker-impl.ts +++ b/packages/cli/src/catchup-runner-worker-impl.ts @@ -1,5 +1,9 @@ import { parentPort } from 'node:worker_threads'; -import { CATCHUP_MAX_CONCURRENT_PEER_SYNCS, mapWithConcurrency } from '@origintrail-official/dkg-agent'; +import { + CATCHUP_MAX_CONCURRENT_PEER_SYNCS, + mapWithConcurrency, + retryCatchupPlaneOnBackpressure, +} from '@origintrail-official/dkg-agent'; import { catchupPeerResponded, catchupPeerSucceeded, @@ -192,13 +196,22 @@ async function runCatchup(request: CatchupRunRequest): Promise syncCapable, CATCHUP_MAX_CONCURRENT_PEER_SYNCS, async (peerId) => { - const rawDurable = await invoke('syncDurable', peerId, request.contextGraphId).catch(() => emptyDurable()); + const rawDurable = await retryCatchupPlaneOnBackpressure( + () => invoke('syncDurable', peerId, request.contextGraphId), + ).catch(() => emptyDurable()); const durable = { ...rawDurable, verifiedPrivateOnlyResponses: rawDurable.verifiedPrivateOnlyResponses ?? 0, }; - const shared = request.includeSharedMemory - ? await invoke('syncSharedMemory', peerId, request.contextGraphId).catch(() => emptyShared()) + + // Durable metadata is required to authorize/materialize SWM. If VM is + // still locally deferred after bounded retries, leave SWM untouched for + // this peer. If only SWM is deferred, retry just SWM so the already- + // completed durable plane is never fetched a second time. + const shared = request.includeSharedMemory && (durable.deferredBackpressure ?? 0) === 0 + ? await retryCatchupPlaneOnBackpressure( + () => invoke('syncSharedMemory', peerId, request.contextGraphId), + ).catch(() => emptyShared()) : null; return { durable, shared }; }, diff --git a/packages/cli/src/catchup-runner.ts b/packages/cli/src/catchup-runner.ts index 9be274fab7..62dbec4a3d 100644 --- a/packages/cli/src/catchup-runner.ts +++ b/packages/cli/src/catchup-runner.ts @@ -3,6 +3,7 @@ import { existsSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import { classifyDurableProgress, + FOREGROUND_CATCHUP_SYNC_PRIORITY, type DKGAgent, type DurableProgressSummary, } from '@origintrail-official/dkg-agent'; @@ -291,11 +292,22 @@ class WorkerCatchupRunner implements CatchupRunner { } case 'syncDurable': { const [peerId, contextGraphId] = args as [string, string]; - return agent.syncFromPeerDetailed(peerId, [contextGraphId]); + return agent.syncFromPeerDetailed( + peerId, + [contextGraphId], + undefined, + undefined, + undefined, + { priority: FOREGROUND_CATCHUP_SYNC_PRIORITY }, + ); } case 'syncSharedMemory': { const [peerId, contextGraphId] = args as [string, string]; - return agent.syncSharedMemoryFromPeerDetailed(peerId, [contextGraphId]); + return agent.syncSharedMemoryFromPeerDetailed( + peerId, + [contextGraphId], + { priority: FOREGROUND_CATCHUP_SYNC_PRIORITY }, + ); } case 'finalizeCatchup': { const [contextGraphId] = args as [string, number, number]; @@ -318,6 +330,8 @@ class InlineCatchupRunner implements CatchupRunner { run(request: CatchupRunRequest): Promise { return this.agent.syncContextGraphFromConnectedPeers(request.contextGraphId, { includeSharedMemory: request.includeSharedMemory, + priority: FOREGROUND_CATCHUP_SYNC_PRIORITY, + retryDeferredBackpressure: true, }) as Promise; } diff --git a/packages/cli/test/catchup-runner-worker-impl.test.ts b/packages/cli/test/catchup-runner-worker-impl.test.ts index dd68c49b4e..2a68311727 100644 --- a/packages/cli/test/catchup-runner-worker-impl.test.ts +++ b/packages/cli/test/catchup-runner-worker-impl.test.ts @@ -9,7 +9,10 @@ // aggregation keeps its one-result-per-peer input-order shape, and one peer's // failure stays isolated instead of failing the whole run. import { describe, expect, it, vi } from 'vitest'; -import { CATCHUP_MAX_CONCURRENT_PEER_SYNCS } from '@origintrail-official/dkg-agent'; +import { + CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS, + CATCHUP_MAX_CONCURRENT_PEER_SYNCS, +} from '@origintrail-official/dkg-agent'; import type { CatchupJobResult, CatchupRunRequest } from '../src/catchup-runner.js'; // The worker impl wires itself to `parentPort` at module load, so a @@ -232,8 +235,10 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) expect(result.diagnostics?.durable.failedPeers).toBe(1); }); - it('surfaces partial progress followed by local deferral without finalizing the catch-up', async () => { + it('retries only SWM after durable progress and finalizes when local pressure clears', async () => { const finalizeCalls: unknown[][] = []; + let durableCalls = 0; + let sharedCalls = 0; const result = await runWorkerCatchup({ contextGraphId: 'cg-deferred', includeSharedMemory: true }, async (method) => { switch (method) { case 'prepareCatchup': @@ -241,17 +246,22 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) case 'waitForSyncProtocol': return true; case 'syncDurable': + durableCalls += 1; return durableResult(); - case 'syncSharedMemory': - return { - ...sharedResult(), - insertedTriples: 0, - fetchedDataTriples: 0, - insertedDataTriples: 0, - bytesReceived: 0, - completedPhases: 0, - deferredBackpressure: 1, - }; + case 'syncSharedMemory': { + sharedCalls += 1; + return sharedCalls === 1 + ? { + ...sharedResult(), + insertedTriples: 0, + fetchedDataTriples: 0, + insertedDataTriples: 0, + bytesReceived: 0, + completedPhases: 0, + deferredBackpressure: 1, + } + : sharedResult(); + } case 'finalizeCatchup': finalizeCalls.push([]); return null; @@ -261,11 +271,94 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) }); expect(result.peersResponded).toBe(1); - expect(result.peersSucceeded).toBe(0); - expect(result.deferredBackpressure).toBe(1); + expect(result.peersSucceeded).toBe(1); + expect(result.deferredBackpressure).toBe(0); expect(result.dataSynced).toBe(1); - expect(result.sharedMemorySynced).toBe(0); - expect(result.diagnostics?.sharedMemory.deferredBackpressure).toBe(1); + expect(result.sharedMemorySynced).toBe(1); + expect(result.diagnostics?.sharedMemory.deferredBackpressure).toBe(0); + expect(durableCalls).toBe(1); + expect(sharedCalls).toBe(2); + expect(finalizeCalls).toEqual([[]]); + }); + + it('finishes deferred durable sync before starting SWM', async () => { + let durableCalls = 0; + let sharedCalls = 0; + const callOrder: string[] = []; + + const result = await runWorkerCatchup( + { contextGraphId: 'cg-durable-deferred', includeSharedMemory: true }, + async (method) => { + switch (method) { + case 'prepareCatchup': + return { preferredPeerId: undefined, isPrivateContextGraph: false, peerIds: ['peer-1'], connectedPeers: 1 }; + case 'waitForSyncProtocol': + return true; + case 'syncDurable': + durableCalls += 1; + callOrder.push(`durable-${durableCalls}`); + return durableCalls === 1 + ? { ...durableResult(), insertedTriples: 0, insertedDataTriples: 0, completedPhases: 0, deferredBackpressure: 1 } + : durableResult(); + case 'syncSharedMemory': + sharedCalls += 1; + callOrder.push('shared'); + return sharedResult(); + case 'finalizeCatchup': + return null; + default: + throw new Error(`unexpected invoke: ${method}`); + } + }, + ); + + expect(result.deferredBackpressure).toBe(0); + expect(durableCalls).toBe(2); + expect(sharedCalls).toBe(1); + expect(callOrder).toEqual(['durable-1', 'durable-2', 'shared']); + }); + + it('returns deferred after a bounded durable retry budget and never starts dependent SWM', async () => { + let durableCalls = 0; + let sharedCalls = 0; + const finalizeCalls: unknown[][] = []; + + const result = await runWorkerCatchup( + { contextGraphId: 'cg-persistently-deferred', includeSharedMemory: true }, + async (method) => { + switch (method) { + case 'prepareCatchup': + return { preferredPeerId: undefined, isPrivateContextGraph: false, peerIds: ['peer-1'], connectedPeers: 1 }; + case 'waitForSyncProtocol': + return true; + case 'syncDurable': + durableCalls += 1; + return { + ...durableResult(), + insertedTriples: 0, + fetchedDataTriples: 0, + insertedDataTriples: 0, + bytesReceived: 0, + completedPhases: 0, + deferredBackpressure: 1, + }; + case 'syncSharedMemory': + sharedCalls += 1; + return sharedResult(); + case 'finalizeCatchup': + finalizeCalls.push([]); + return null; + default: + throw new Error(`unexpected invoke: ${method}`); + } + }, + ); + + expect(durableCalls).toBe(CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS.length + 1); + expect(sharedCalls).toBe(0); + expect(result.deferredBackpressure).toBe(1); + expect(result.peersResponded).toBe(0); + expect(result.peersSucceeded).toBe(0); expect(finalizeCalls).toEqual([]); });