From 5f7c4c94f02126bcfae5b41336c5be6e0b38d1e8 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Thu, 23 Jul 2026 12:19:28 +0200 Subject: [PATCH 1/2] fix(sync): retry foreground catchup under backpressure --- packages/agent/src/dkg-agent-lifecycle.ts | 69 +++++-- packages/agent/src/index.ts | 11 ++ packages/agent/src/sync/catchup-policy.ts | 78 ++++++++ packages/agent/test/catchup-policy.test.ts | 108 +++++++++++ .../agent/test/sync-fetch-coalescing.test.ts | 168 +++++++++++++++++- .../cli/src/catchup-runner-worker-impl.ts | 33 ++-- packages/cli/src/catchup-runner.ts | 20 ++- .../test/catchup-runner-worker-impl.test.ts | 138 ++++++++++++-- 8 files changed, 581 insertions(+), 44 deletions(-) create mode 100644 packages/agent/src/sync/catchup-policy.ts create mode 100644 packages/agent/test/catchup-policy.test.ts diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 437af21db6..23e9781a9e 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -257,6 +257,10 @@ 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 { + runCatchupPlanesWithPolicy, + type CatchupMode, +} from './sync/catchup-policy.js'; import { classifyDurableProgress, createDurableSyncAccumulator, @@ -621,12 +625,14 @@ function contextGraphCatchupSingleFlightKey(params: { includeSharedMemory: boolean; maxPeers?: number; peerRotationKey?: string; + mode: CatchupMode; }): string { return syncSingleFlightKey('context-graph-catchup', { contextGraphId: params.contextGraphId, includeSharedMemory: params.includeSharedMemory, maxPeers: normalizedCatchupMaxPeers(params.maxPeers), peerRotationKey: params.peerRotationKey ?? null, + mode: params.mode, }); } @@ -640,6 +646,7 @@ function durableSyncSingleFlightKey(params: { hasAccessDeniedCallback: boolean; hasSinceBatchIdResolver: boolean; exactAssetUals?: readonly string[]; + priority?: number; }): string | null { if (params.hasPhaseCallback || params.hasAccessDeniedCallback || params.hasSinceBatchIdResolver) { return null; @@ -651,6 +658,7 @@ function durableSyncSingleFlightKey(params: { totalTimeoutMs: params.totalTimeoutMs, syncAgentsMeta: params.syncAgentsMeta, exactAssetUals: params.exactAssetUals ?? null, + priority: params.priority ?? null, }); } @@ -660,6 +668,7 @@ function sharedMemorySyncSingleFlightKey(params: { stopOnBackoffWorthyFailure?: boolean; publicContextGraphIds: readonly string[]; privateRecoverFromCurator: readonly string[]; + priority?: number; }): string { return syncSingleFlightKey('shared-memory-sync', { remotePeerId: params.remotePeerId, @@ -667,6 +676,7 @@ function sharedMemorySyncSingleFlightKey(params: { stopOnBackoffWorthyFailure: params.stopOnBackoffWorthyFailure === true, publicContextGraphIds: params.publicContextGraphIds, privateRecoverFromCurator: params.privateRecoverFromCurator, + priority: params.priority ?? null, }); } @@ -768,6 +778,17 @@ interface RecoverContextGraphSwmFromPeerDependencies { type SyncReconcilerAttemptOutcome = SyncOnConnectOutcome | 'not-started' | 'deferred-backpressure'; +export interface ContextGraphCatchupOptions { + includeSharedMemory?: boolean; + maxPeers?: number; + peerRotationKey?: string; + /** + * Foreground mode receives scheduler priority and bounded local-deferral + * retries. Background mode remains best-effort and never waits for capacity. + */ + mode?: CatchupMode; +} + export type DurableSyncOptions = { stopOnBackoffWorthyFailure?: boolean; /** @@ -4002,7 +4023,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 && (changelogResult.deferredBackpressure ?? 0) > 0) { @@ -4113,6 +4140,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(); } @@ -4284,6 +4312,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)) { @@ -4324,6 +4353,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { item.lane, item.operationId, run, + priority, ), merge: mergeDurableSyncAccumulatorInto, markDeferred: (summary) => { @@ -4754,6 +4784,8 @@ export class LifecycleSyncMethods extends DKGAgentBase { options?: { stopOnBackoffWorthyFailure?: boolean; sharedMemorySyncPlan?: SharedMemorySyncContextGraphPlan; + /** Admission override for foreground catch-up. */ + priority?: number; }, ): Promise { const ctx = createOperationContext('sync'); @@ -4820,6 +4852,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { stopOnBackoffWorthyFailure, publicContextGraphIds, privateRecoverFromCurator, + priority: options?.priority, }); const runSync = async (): Promise => { @@ -4973,6 +5006,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { item.lane, item.operationId, run, + options?.priority, ), merge: mergeSharedMemorySyncResults, markDeferred: (summary) => ({ @@ -5077,7 +5111,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { */ async syncContextGraphFromConnectedPeers(this: DKGAgent, contextGraphId: string, - options?: { includeSharedMemory?: boolean; maxPeers?: number; peerRotationKey?: string }, + options?: ContextGraphCatchupOptions, ): Promise<{ /** Ordered connected peers before optional maxPeers windowing. */ connectedPeers: number; @@ -5122,6 +5156,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { }> { const ctx = createOperationContext('sync'); const includeSharedMemory = options?.includeSharedMemory ?? false; + const mode = options?.mode ?? 'background'; this.trackSyncContextGraph(contextGraphId); @@ -5130,6 +5165,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { includeSharedMemory, maxPeers: options?.maxPeers, peerRotationKey: options?.peerRotationKey, + mode, }); return runSyncSingleFlight(this, singleFlightKey, async (): Promise => { @@ -5179,6 +5215,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { ); return this.runCatchupOverPeers(contextGraphId, includeSharedMemory, peers, { totalPeers: orderedPeers.length, + mode, }); }); } @@ -5269,7 +5306,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { contextGraphId: string, includeSharedMemory: boolean, peers: Array<{ toString(): string }>, - stats?: { totalPeers?: number }, + stats?: { totalPeers?: number; mode?: CatchupMode }, ): Promise<{ /** Ordered connected peers before optional caller windowing. */ connectedPeers: number; @@ -5405,14 +5442,24 @@ export class LifecycleSyncMethods extends DKGAgentBase { syncCapable, CATCHUP_MAX_CONCURRENT_PEER_SYNCS, async (remotePeerId) => { - const durable = await this.syncFromPeerDetailed( - remotePeerId, - [contextGraphId], - ).catch(() => createFailedPeerDurableSyncResult()); - const shared = includeSharedMemory - ? await this.syncSharedMemoryFromPeerDetailed(remotePeerId, [contextGraphId]).catch(emptyShared) - : null; - return { durable, shared }; + const mode = stats?.mode ?? 'background'; + return runCatchupPlanesWithPolicy({ + mode, + includeSharedMemory, + syncDurable: ({ priority }) => this.syncFromPeerDetailed( + remotePeerId, + [contextGraphId], + undefined, + undefined, + undefined, + priority === undefined ? undefined : { priority }, + ).catch(() => createFailedPeerDurableSyncResult()), + syncSharedMemory: ({ priority }) => this.syncSharedMemoryFromPeerDetailed( + remotePeerId, + [contextGraphId], + priority === undefined ? undefined : { priority }, + ).catch(emptyShared), + }); }, ); let accessDeniedPeers = 0; diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index f218a79c7f..108e14d42a 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -271,6 +271,17 @@ 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, + catchupPriorityForMode, + runCatchupPlanesWithPolicy, + type CatchupMode, + type CatchupPlaneContext, + type CatchupPlanePolicyOptions, + type CatchupPlanePolicyResult, + type CatchupPlaneResult, +} from './sync/catchup-policy.js'; export { classifyDurableProgress, createFailedPeerDurableSyncResult, diff --git a/packages/agent/src/sync/catchup-policy.ts b/packages/agent/src/sync/catchup-policy.ts new file mode 100644 index 0000000000..97d001dd90 --- /dev/null +++ b/packages/agent/src/sync/catchup-policy.ts @@ -0,0 +1,78 @@ +export type CatchupMode = 'background' | 'foreground'; + +export const FOREGROUND_CATCHUP_SYNC_PRIORITY = 2_000; +export const CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS = [100, 250, 500] as const; + +export interface CatchupPlaneResult { + deferredBackpressure?: number; +} + +export interface CatchupPlaneContext { + priority?: number; +} + +export interface CatchupPlanePolicyOptions< + TDurable extends CatchupPlaneResult, + TShared extends CatchupPlaneResult, +> { + mode: CatchupMode; + includeSharedMemory: boolean; + syncDurable: (context: CatchupPlaneContext) => Promise; + syncSharedMemory: (context: CatchupPlaneContext) => Promise; + retryDelaysMs?: readonly number[]; + wait?: (delayMs: number) => Promise; +} + +export interface CatchupPlanePolicyResult< + TDurable extends CatchupPlaneResult, + TShared extends CatchupPlaneResult, +> { + durable: TDurable; + shared: TShared | null; +} + +export function catchupPriorityForMode(mode: CatchupMode): number | undefined { + return mode === 'foreground' ? FOREGROUND_CATCHUP_SYNC_PRIORITY : undefined; +} + +async function runCatchupPlane( + mode: CatchupMode, + run: (context: CatchupPlaneContext) => Promise, + options: Pick, 'retryDelaysMs' | 'wait'>, +): Promise { + const context = { priority: catchupPriorityForMode(mode) }; + let result = await run(context); + if (mode !== 'foreground') return result; + + const retryDelaysMs = options.retryDelaysMs ?? CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS; + const wait = options.wait ?? ((delayMs: number) => new Promise((resolve) => { + setTimeout(resolve, delayMs); + })); + for (const delayMs of retryDelaysMs) { + if ((result.deferredBackpressure ?? 0) === 0) return result; + await wait(delayMs); + result = await run(context); + } + return result; +} + +/** + * Canonical foreground/background catch-up policy shared by the in-agent and + * worker-backed runners. Durable metadata must settle before SWM starts; when + * only SWM is deferred, retries never refetch the already-completed durable + * plane. + */ +export async function runCatchupPlanesWithPolicy< + TDurable extends CatchupPlaneResult, + TShared extends CatchupPlaneResult, +>( + options: CatchupPlanePolicyOptions, +): Promise> { + const durable = await runCatchupPlane(options.mode, options.syncDurable, options); + if (!options.includeSharedMemory || (durable.deferredBackpressure ?? 0) > 0) { + return { durable, shared: null }; + } + + const shared = await runCatchupPlane(options.mode, options.syncSharedMemory, options); + return { durable, shared }; +} diff --git a/packages/agent/test/catchup-policy.test.ts b/packages/agent/test/catchup-policy.test.ts new file mode 100644 index 0000000000..59ca4505e1 --- /dev/null +++ b/packages/agent/test/catchup-policy.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + runCatchupPlanesWithPolicy, +} from '../src/sync/catchup-policy.js'; + +describe('runCatchupPlanesWithPolicy', () => { + it('derives foreground priority and retries durable before starting SWM', async () => { + const order: string[] = []; + const priorities: Array = []; + const waits: number[] = []; + const syncDurable = vi.fn(async ({ priority }: { priority?: number }) => { + priorities.push(priority); + order.push(`durable-${syncDurable.mock.calls.length}`); + return { deferredBackpressure: syncDurable.mock.calls.length === 1 ? 1 : 0 }; + }); + const syncSharedMemory = vi.fn(async ({ priority }: { priority?: number }) => { + priorities.push(priority); + order.push('shared'); + return { deferredBackpressure: 0 }; + }); + + const result = await runCatchupPlanesWithPolicy({ + mode: 'foreground', + includeSharedMemory: true, + syncDurable, + syncSharedMemory, + retryDelaysMs: [3, 5], + wait: async (delayMs) => { waits.push(delayMs); }, + }); + + expect(result).toEqual({ + durable: { deferredBackpressure: 0 }, + shared: { deferredBackpressure: 0 }, + }); + expect(order).toEqual(['durable-1', 'durable-2', 'shared']); + expect(priorities).toEqual([ + FOREGROUND_CATCHUP_SYNC_PRIORITY, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + ]); + expect(waits).toEqual([3]); + }); + + it('retries only SWM when durable already completed', async () => { + const syncDurable = vi.fn(async () => ({ deferredBackpressure: 0 })); + const syncSharedMemory = vi.fn() + .mockResolvedValueOnce({ deferredBackpressure: 1 }) + .mockResolvedValueOnce({ deferredBackpressure: 0 }); + + const result = await runCatchupPlanesWithPolicy({ + mode: 'foreground', + includeSharedMemory: true, + syncDurable, + syncSharedMemory, + retryDelaysMs: [1], + wait: async () => {}, + }); + + expect(result.shared?.deferredBackpressure).toBe(0); + expect(syncDurable).toHaveBeenCalledTimes(1); + expect(syncSharedMemory).toHaveBeenCalledTimes(2); + }); + + it('returns the final durable deferral without starting dependent SWM', async () => { + const syncDurable = vi.fn(async () => ({ deferredBackpressure: 1 })); + const syncSharedMemory = vi.fn(async () => ({ deferredBackpressure: 0 })); + + const result = await runCatchupPlanesWithPolicy({ + mode: 'foreground', + includeSharedMemory: true, + syncDurable, + syncSharedMemory, + wait: async () => {}, + }); + + expect(result.durable.deferredBackpressure).toBe(1); + expect(result.shared).toBeNull(); + expect(syncDurable).toHaveBeenCalledTimes( + CATCHUP_BACKPRESSURE_RETRY_DELAYS_MS.length + 1, + ); + expect(syncSharedMemory).not.toHaveBeenCalled(); + }); + + it('keeps background catch-up best-effort without retries or priority', async () => { + const priorities: Array = []; + const syncDurable = vi.fn(async ({ priority }: { priority?: number }) => { + priorities.push(priority); + return { deferredBackpressure: 1 }; + }); + const syncSharedMemory = vi.fn(async () => ({ deferredBackpressure: 0 })); + + const result = await runCatchupPlanesWithPolicy({ + mode: 'background', + includeSharedMemory: true, + syncDurable, + syncSharedMemory, + wait: async () => { throw new Error('background mode must not wait'); }, + }); + + expect(result.durable.deferredBackpressure).toBe(1); + expect(result.shared).toBeNull(); + expect(syncDurable).toHaveBeenCalledTimes(1); + expect(syncSharedMemory).not.toHaveBeenCalled(); + expect(priorities).toEqual([undefined]); + }); +}); diff --git a/packages/agent/test/sync-fetch-coalescing.test.ts b/packages/agent/test/sync-fetch-coalescing.test.ts index e9aadc0f72..215436450e 100644 --- a/packages/agent/test/sync-fetch-coalescing.test.ts +++ b/packages/agent/test/sync-fetch-coalescing.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from 'vitest'; import { createOperationContext, PROTOCOL_SYNC } from '@origintrail-official/dkg-core'; import { MockChainAdapter } from '@origintrail-official/dkg-chain'; -import { DKGAgent } from '../src/index.js'; +import { + DKGAgent, + FOREGROUND_CATCHUP_SYNC_PRIORITY, +} from '../src/index.js'; import { resolveSyncGlobalBackpressure, SyncBackpressureBusyError, withGlobalSyncBackpressure } from '../src/sync/backpressure.js'; import type { SyncPhase } from '../src/sync/auth/request-build.js'; import type { SyncPageResult } from '../src/sync/requester/page-fetch.js'; @@ -371,6 +374,50 @@ describe('DKGAgent sync fetch coalescing', () => { } }); + it('does not join direct durable syncs with different admission priorities', async () => { + let fetchCalls = 0; + const agent = await createAgentWithSend(async () => new Uint8Array(0)); + (agent as any).fetchSyncPages = async (...args: unknown[]) => { + fetchCalls++; + return emptySyncPage(String(args[4])); + }; + (agent as any).processDurableBatchInWorker = async () => ({ + verifiedData: [], + verifiedMeta: [], + totalFetchedDataQuads: 0, + totalFetchedMetaQuads: 0, + rejectedKcs: 0, + emptyResponses: 1, + metaOnlyResponses: 0, + dataRejectedMissingMeta: 0, + }); + + try { + const background = (agent as any).syncFromPeerDetailed( + PEER_A, + ['coalesced-cg'], + undefined, + undefined, + undefined, + { priority: 0 }, + ); + const foreground = (agent as any).syncFromPeerDetailed( + PEER_A, + ['coalesced-cg'], + undefined, + undefined, + undefined, + { priority: FOREGROUND_CATCHUP_SYNC_PRIORITY }, + ); + + const [backgroundResult, foregroundResult] = await Promise.all([background, foreground]); + expect(backgroundResult).not.toBe(foregroundResult); + expect(fetchCalls).toBe(4); + } finally { + await agent.stop().catch(() => {}); + } + }); + it('does not single-flight exact VM syncs with different asset batches', async () => { let fetchCalls = 0; const agent = await createAgentWithSend(async () => new Uint8Array(0)); @@ -543,6 +590,51 @@ describe('DKGAgent sync fetch coalescing', () => { } }); + it('does not join direct shared-memory syncs with different admission priorities', async () => { + let fetchCalls = 0; + const agent = await createAgentWithSend(async () => new Uint8Array(0)); + const sharedMemorySyncPlan = { + eligibleContextGraphIds: ['coalesced-cg'], + publicContextGraphIds: ['coalesced-cg'], + privateRecoverFromCurator: [], + }; + (agent as any).listSubGraphs = async () => []; + (agent as any).fetchSyncPages = async (...args: unknown[]) => { + fetchCalls++; + return emptySyncPage(String(args[4])); + }; + (agent as any).getOrCreateSyncVerifyWorker = () => ({ + processSharedMemoryBatch: async () => ({ + verifiedData: [], + verifiedMeta: [], + totalFetchedDataQuads: 0, + totalFetchedMetaQuads: 0, + droppedDataTriples: 0, + emptyResponses: 1, + entityCreators: [], + }), + }); + + try { + const background = (agent as any).syncSharedMemoryFromPeerDetailed( + PEER_A, + ['coalesced-cg'], + { sharedMemorySyncPlan, priority: 0 }, + ); + const foreground = (agent as any).syncSharedMemoryFromPeerDetailed( + PEER_A, + ['coalesced-cg'], + { sharedMemorySyncPlan, priority: FOREGROUND_CATCHUP_SYNC_PRIORITY }, + ); + + const [backgroundResult, foregroundResult] = await Promise.all([background, foreground]); + expect(backgroundResult).not.toBe(foregroundResult); + expect(fetchCalls).toBe(4); + } finally { + await agent.stop().catch(() => {}); + } + }); + it.each([ { name: 'private-only', @@ -686,6 +778,65 @@ describe('DKGAgent sync fetch coalescing', () => { } }); + it('retries foreground durable admission before starting SWM in the agent path', async () => { + const agent = await createAgentWithSend(async () => new Uint8Array(0)); + const remotePeer = { toString: () => PEER_A }; + const order: string[] = []; + const priorities: Array = []; + let durableCalls = 0; + + try { + await agent.start(); + (agent as any).waitForSyncProtocol = async () => true; + (agent as any).refreshMetaSyncedFlags = async () => undefined; + (agent as any).syncFromPeerDetailed = async ( + _peerId: string, + _contextGraphIds: string[], + _onPhase: unknown, + _onAccessDenied: unknown, + _sinceBatchIdFor: unknown, + options: { priority?: number } | undefined, + ) => { + durableCalls += 1; + priorities.push(options?.priority); + order.push(`durable-${durableCalls}`); + return durableCalls === 1 + ? { + ...cleanDurableSyncResult(), + completedPhases: 0, + deferredBackpressure: 1, + } + : cleanDurableSyncResult(); + }; + (agent as any).syncSharedMemoryFromPeerDetailed = async ( + _peerId: string, + _contextGraphIds: string[], + options: { priority?: number } | undefined, + ) => { + priorities.push(options?.priority); + order.push('shared'); + return cleanSharedMemorySyncResult(); + }; + + const result = await (agent as any).runCatchupOverPeers( + 'coalesced-cg', + true, + [remotePeer], + { mode: 'foreground' }, + ); + + expect(result.deferredBackpressure).toBe(0); + expect(order).toEqual(['durable-1', 'durable-2', 'shared']); + expect(priorities).toEqual([ + FOREGROUND_CATCHUP_SYNC_PRIORITY, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + FOREGROUND_CATCHUP_SYNC_PRIORITY, + ]); + } finally { + await agent.stop().catch(() => {}); + } + }); + it('does not promote catch-up readiness when durable integrity verification rejects a KA', async () => { const agent = await createAgentWithSend(async () => new Uint8Array(0)); const remotePeer = { toString: () => PEER_A }; @@ -757,12 +908,23 @@ describe('DKGAgent sync fetch coalescing', () => { it('does not join catch-up rounds with different shareable identity fields', async () => { const cases: Array<{ name: string; - firstOptions?: { includeSharedMemory?: boolean; maxPeers?: number; peerRotationKey?: string }; - secondOptions?: { includeSharedMemory?: boolean; maxPeers?: number; peerRotationKey?: string }; + firstOptions?: { + includeSharedMemory?: boolean; + maxPeers?: number; + peerRotationKey?: string; + mode?: 'background' | 'foreground'; + }; + secondOptions?: { + includeSharedMemory?: boolean; + maxPeers?: number; + peerRotationKey?: string; + mode?: 'background' | 'foreground'; + }; }> = [ { name: 'includeSharedMemory', firstOptions: {}, secondOptions: { includeSharedMemory: true } }, { name: 'maxPeers', firstOptions: { maxPeers: 1 }, secondOptions: { maxPeers: 2 } }, { name: 'peerRotationKey', firstOptions: { peerRotationKey: 'a' }, secondOptions: { peerRotationKey: 'b' } }, + { name: 'mode', firstOptions: { mode: 'background' }, secondOptions: { mode: 'foreground' } }, ]; for (const testCase of cases) { diff --git a/packages/cli/src/catchup-runner-worker-impl.ts b/packages/cli/src/catchup-runner-worker-impl.ts index f92dcd9f87..988e151493 100644 --- a/packages/cli/src/catchup-runner-worker-impl.ts +++ b/packages/cli/src/catchup-runner-worker-impl.ts @@ -3,6 +3,7 @@ import { CATCHUP_MAX_CONCURRENT_PEER_SYNCS, createFailedPeerDurableSyncResult, mapWithConcurrency, + runCatchupPlanesWithPolicy, } from '@origintrail-official/dkg-agent'; import { catchupPeerResponded, @@ -175,16 +176,28 @@ async function runCatchup(request: CatchupRunRequest): Promise syncCapable, CATCHUP_MAX_CONCURRENT_PEER_SYNCS, async (peerId) => { - const rawDurable = await invoke('syncDurable', peerId, request.contextGraphId) - .catch(() => createFailedPeerDurableSyncResult()); - const durable = { - ...rawDurable, - verifiedPrivateOnlyResponses: rawDurable.verifiedPrivateOnlyResponses ?? 0, - }; - const shared = request.includeSharedMemory - ? await invoke('syncSharedMemory', peerId, request.contextGraphId).catch(() => emptyShared()) - : null; - return { durable, shared }; + return runCatchupPlanesWithPolicy({ + mode: 'foreground', + includeSharedMemory: request.includeSharedMemory, + syncDurable: async ({ priority }) => { + const rawDurable = await invoke( + 'syncDurable', + peerId, + request.contextGraphId, + priority, + ).catch(() => createFailedPeerDurableSyncResult()); + return { + ...rawDurable, + verifiedPrivateOnlyResponses: rawDurable.verifiedPrivateOnlyResponses ?? 0, + }; + }, + syncSharedMemory: ({ priority }) => invoke( + 'syncSharedMemory', + peerId, + request.contextGraphId, + priority, + ).catch(() => emptyShared()), + }); }, ); for (const { durable, shared } of perPeerResults) { diff --git a/packages/cli/src/catchup-runner.ts b/packages/cli/src/catchup-runner.ts index de88a3418a..5407e20a30 100644 --- a/packages/cli/src/catchup-runner.ts +++ b/packages/cli/src/catchup-runner.ts @@ -550,12 +550,23 @@ class WorkerCatchupRunner implements CatchupRunner { return agent.waitForSyncProtocol({ toString: () => peerId }); } case 'syncDurable': { - const [peerId, contextGraphId] = args as [string, string]; - return agent.syncFromPeerDetailed(peerId, [contextGraphId]); + const [peerId, contextGraphId, priority] = args as [string, string, number | undefined]; + return agent.syncFromPeerDetailed( + peerId, + [contextGraphId], + undefined, + undefined, + undefined, + priority === undefined ? undefined : { priority }, + ); } case 'syncSharedMemory': { - const [peerId, contextGraphId] = args as [string, string]; - return agent.syncSharedMemoryFromPeerDetailed(peerId, [contextGraphId]); + const [peerId, contextGraphId, priority] = args as [string, string, number | undefined]; + return agent.syncSharedMemoryFromPeerDetailed( + peerId, + [contextGraphId], + priority === undefined ? undefined : { priority }, + ); } case 'finalizeCatchup': { const [contextGraphId] = args as [string, number, number]; @@ -578,6 +589,7 @@ class InlineCatchupRunner implements CatchupRunner { run(request: CatchupRunRequest): Promise { return this.agent.syncContextGraphFromConnectedPeers(request.contextGraphId, { includeSharedMemory: request.includeSharedMemory, + mode: 'foreground', }) 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 2244e4f37d..c8688098fa 100644 --- a/packages/cli/test/catchup-runner-worker-impl.test.ts +++ b/packages/cli/test/catchup-runner-worker-impl.test.ts @@ -9,7 +9,11 @@ // 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, + FOREGROUND_CATCHUP_SYNC_PRIORITY, +} 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 @@ -128,6 +132,7 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) let peakSyncs = 0; const durableOrder: string[] = []; const sharedSeen: string[] = []; + const syncPriorities: Array = []; const finalizeCalls: unknown[][] = []; const result = await runWorkerCatchup({ contextGraphId: 'cg-storm', includeSharedMemory: true }, async (method, args) => { @@ -143,6 +148,7 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) } case 'syncDurable': { durableOrder.push(args[0] as string); + syncPriorities.push(args[2] as number | undefined); inFlightSyncs += 1; peakSyncs = Math.max(peakSyncs, inFlightSyncs); await delay(4); @@ -151,6 +157,7 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) } case 'syncSharedMemory': { sharedSeen.push(args[0] as string); + syncPriorities.push(args[2] as number | undefined); inFlightSyncs += 1; peakSyncs = Math.max(peakSyncs, inFlightSyncs); await delay(2); @@ -175,6 +182,9 @@ describe('catchup-runner-worker-impl bounded fan-out (sync-storm mitigation C-1) // order (the bounded mapper's shared cursor hands out work in order). expect(durableOrder).toEqual(peerIds); expect([...sharedSeen].sort()).toEqual([...peerIds].sort()); + expect(syncPriorities).toEqual( + Array.from({ length: peerIds.length * 2 }, () => FOREGROUND_CATCHUP_SYNC_PRIORITY), + ); // Aggregation unchanged from the unbounded Promise.all shape. expect(result.selectedPeers).toBe(peerIds.length); @@ -233,8 +243,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': @@ -242,17 +254,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; @@ -262,11 +279,100 @@ 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([]); }); From e5e11209af198b2c2c38c2deed4093f9b8640232 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Thu, 23 Jul 2026 12:48:25 +0200 Subject: [PATCH 2/2] fix(sync): preserve background catchup call shape --- packages/agent/src/dkg-agent-lifecycle.ts | 26 +++++++++++++---------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 23e9781a9e..80aeeff886 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -5446,18 +5446,22 @@ export class LifecycleSyncMethods extends DKGAgentBase { return runCatchupPlanesWithPolicy({ mode, includeSharedMemory, - syncDurable: ({ priority }) => this.syncFromPeerDetailed( - remotePeerId, - [contextGraphId], - undefined, - undefined, - undefined, - priority === undefined ? undefined : { priority }, + syncDurable: ({ priority }) => ( + priority === undefined + ? this.syncFromPeerDetailed(remotePeerId, [contextGraphId]) + : this.syncFromPeerDetailed( + remotePeerId, + [contextGraphId], + undefined, + undefined, + undefined, + { priority }, + ) ).catch(() => createFailedPeerDurableSyncResult()), - syncSharedMemory: ({ priority }) => this.syncSharedMemoryFromPeerDetailed( - remotePeerId, - [contextGraphId], - priority === undefined ? undefined : { priority }, + syncSharedMemory: ({ priority }) => ( + priority === undefined + ? this.syncSharedMemoryFromPeerDetailed(remotePeerId, [contextGraphId]) + : this.syncSharedMemoryFromPeerDetailed(remotePeerId, [contextGraphId], { priority }) ).catch(emptyShared), }); },