From e7c85f5f42cc36ef0ca06b979418f5654cc57104 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 03:37:58 +0200 Subject: [PATCH 01/12] feat(sync): resolve role-aware capacity policy --- packages/agent/src/sync/capacity-runtime.ts | 198 ++++++++++++++++++ .../agent/test/sync-capacity-runtime.test.ts | 94 +++++++++ 2 files changed, 292 insertions(+) create mode 100644 packages/agent/src/sync/capacity-runtime.ts create mode 100644 packages/agent/test/sync-capacity-runtime.test.ts diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts new file mode 100644 index 000000000..548578554 --- /dev/null +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -0,0 +1,198 @@ +import type { TripleStore } from '@origintrail-official/dkg-storage'; +import type { SyncAdaptiveCapacityConfig } from '../dkg-agent-types.js'; +import { + AdaptiveCapacityController, + MAX_SYNC_ADAPTIVE_INFLIGHT, + resolveAdaptiveCapacityBounds, + type AdaptiveCapacityDecision, + type AdaptiveCapacityState, +} from './adaptive-capacity.js'; +import { + AdaptiveCapacitySampler, + deriveAdaptiveInflightHardMax, + type AdaptiveCapacitySamplerDependencies, +} from './adaptive-capacity-sampler.js'; +import { + notifyGlobalSyncBackpressureCapacityChanged, + parseBooleanEnv, + resolveNonNegativeIntegerSwitch, + resolvePositiveIntegerSwitch, + resolveSyncGlobalBackpressure, + type SyncGlobalBackpressureConfig, + type SyncGlobalBackpressurePolicy, +} from './backpressure.js'; +import { resolveCorePublicSyncBatchSize } from './core-public-coverage-scheduler.js'; + +export interface SyncCapacityRuntimeConfig extends SyncGlobalBackpressureConfig { + nodeRole?: 'core' | 'edge'; + syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig; + syncCorePublicBatchSize?: number; +} + +export interface SyncCapacityStatus { + mode: 'static' | 'adaptive'; + state: AdaptiveCapacityState | 'healthy'; + currentInflight: number | null; + minInflight: number | null; + maxInflight: number | null; + currentCoverageBatch: number; + configuredCoverageBatch: number; + storePressureTelemetryAvailable: boolean; + lastDecision: Pick< + AdaptiveCapacityDecision, + 'action' | 'reason' | 'atMs' + > | null; +} + +export interface SyncCapacityRuntimeOptions { + parallelism?: number; + samplerDependencies?: AdaptiveCapacitySamplerDependencies; + now?: () => number; +} + +function resolvedExplicitGlobalLimit( + config: SyncGlobalBackpressureConfig, +): number | undefined { + return resolveNonNegativeIntegerSwitch( + config.syncGlobalMaxInflight, + 'DKG_SYNC_GLOBAL_MAX_INFLIGHT', + ) ?? resolveNonNegativeIntegerSwitch( + config.syncGlobalLimit, + 'DKG_SYNC_GLOBAL_LIMIT', + ); +} + +function readStorePressure(store: TripleStore) { + try { + return store.getPressureSnapshot?.(); + } catch { + return undefined; + } +} + +/** One agent-owned capacity policy; static callers keep the exact old path. */ +export class SyncCapacityRuntime { + private readonly controller?: AdaptiveCapacityController; + private readonly sampler?: AdaptiveCapacitySampler; + + private constructor( + readonly policy: SyncGlobalBackpressurePolicy, + private readonly configuredCoverageBatch: number, + controller?: AdaptiveCapacityController, + sampler?: AdaptiveCapacitySampler, + ) { + this.controller = controller; + this.sampler = sampler; + } + + static create( + config: SyncCapacityRuntimeConfig, + store: TripleStore, + options: SyncCapacityRuntimeOptions = {}, + ): SyncCapacityRuntime { + const configuredCoverageBatch = resolveCorePublicSyncBatchSize( + config.syncCorePublicBatchSize, + ); + const staticPolicy = resolveSyncGlobalBackpressure(config); + const explicitGlobalLimit = resolvedExplicitGlobalLimit(config); + const explicitlyEnabled = parseBooleanEnv('DKG_SYNC_ADAPTIVE_CAPACITY_ENABLED') + ?? config.syncAdaptiveCapacity?.enabled; + const adaptive = (config.nodeRole ?? 'edge') === 'core' + && staticPolicy.limit !== undefined + && (explicitlyEnabled ?? explicitGlobalLimit === undefined); + if (!adaptive) { + return new SyncCapacityRuntime(staticPolicy, configuredCoverageBatch); + } + + const requestedMax = resolvePositiveIntegerSwitch( + config.syncAdaptiveCapacity?.maxInflight, + 'DKG_SYNC_ADAPTIVE_MAX_INFLIGHT', + ); + const minInflight = resolvePositiveIntegerSwitch( + config.syncAdaptiveCapacity?.minInflight, + 'DKG_SYNC_ADAPTIVE_MIN_INFLIGHT', + ); + const hardMax = deriveAdaptiveInflightHardMax({ + operatorMax: Math.min( + requestedMax ?? MAX_SYNC_ADAPTIVE_INFLIGHT, + explicitGlobalLimit ?? MAX_SYNC_ADAPTIVE_INFLIGHT, + ), + parallelism: options.parallelism, + storePressure: readStorePressure(store), + }); + const bounds = resolveAdaptiveCapacityBounds({ + initialInflight: Math.min(2, hardMax), + minInflight, + maxInflight: hardMax, + configuredCoverageBatch, + }); + if (bounds.mode === 'unbounded') { + return new SyncCapacityRuntime(staticPolicy, configuredCoverageBatch); + } + + const adaptivePolicy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: explicitGlobalLimit ?? hardMax, + syncGlobalQueueLimit: config.syncGlobalQueueLimit, + }); + return new SyncCapacityRuntime( + adaptivePolicy, + configuredCoverageBatch, + new AdaptiveCapacityController(bounds, { now: options.now }), + new AdaptiveCapacitySampler(store, options.samplerDependencies), + ); + } + + isAdaptive(): boolean { + return this.controller !== undefined; + } + + getCurrentInflight(): number | undefined { + return this.controller?.getCurrentInflight(); + } + + getEffectiveCoverageBatch(): number { + return this.controller?.getEffectiveCoverageBatch() ?? this.configuredCoverageBatch; + } + + sample(demand: boolean): void { + if (!this.controller || !this.sampler) return; + const previousInflight = this.controller.getCurrentInflight(); + this.controller.observe(this.sampler.sample(demand)); + if (this.controller.getCurrentInflight() !== previousInflight) { + notifyGlobalSyncBackpressureCapacityChanged(); + } + } + + getStatus(): SyncCapacityStatus { + if (!this.controller) { + const currentInflight = this.policy.limit ?? null; + return { + mode: 'static', + state: 'healthy', + currentInflight, + minInflight: currentInflight, + maxInflight: currentInflight, + currentCoverageBatch: this.configuredCoverageBatch, + configuredCoverageBatch: this.configuredCoverageBatch, + storePressureTelemetryAvailable: false, + lastDecision: null, + }; + } + const status = this.controller.getStatus(); + return { + mode: 'adaptive', + state: status.state, + currentInflight: status.currentInflight, + minInflight: status.minInflight, + maxInflight: status.maxInflight, + currentCoverageBatch: status.effectiveCoverageBatch, + configuredCoverageBatch: status.configuredCoverageBatch, + storePressureTelemetryAvailable: status.storePressureTelemetryAvailable, + lastDecision: { + action: status.lastDecision.action, + reason: status.lastDecision.reason, + atMs: status.lastDecision.atMs, + }, + }; + } +} diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts new file mode 100644 index 000000000..222a6c74c --- /dev/null +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -0,0 +1,94 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { StorePressureSnapshot, TripleStore } from '@origintrail-official/dkg-storage'; +import { SyncCapacityRuntime } from '../src/sync/capacity-runtime.js'; + +function fakeStore(pressure?: StorePressureSnapshot): TripleStore { + return { + getPressureSnapshot: () => pressure, + } as unknown as TripleStore; +} + +const STORE_PRESSURE: StorePressureSnapshot = { + ackInflight: 0, + healthInflight: 0, + normalInflight: 0, + backgroundInflight: 0, + ackQueued: 0, + healthQueued: 0, + normalQueued: 0, + backgroundQueued: 0, + maxConcurrent: 5, + ackReservedSlots: 1, + healthReservedSlots: 1, +}; + +afterEach(() => vi.unstubAllEnvs()); + +describe('sync capacity runtime resolution', () => { + it('keeps Edge nodes on the exact static policy even if adaptive is requested', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'edge', + syncGlobalMaxInflight: 4, + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(runtime.isAdaptive()).toBe(false); + expect(runtime.policy.limit).toBe(4); + expect(runtime.getStatus()).toMatchObject({ mode: 'static', currentInflight: 4 }); + }); + + it('defaults Core nodes without an explicit global limit to adaptive mode', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncCorePublicBatchSize: 7, + }, fakeStore(STORE_PRESSURE), { parallelism: 16, now: () => 100 }); + + expect(runtime.isAdaptive()).toBe(true); + expect(runtime.policy.limit).toBe(3); + expect(runtime.getStatus()).toMatchObject({ + mode: 'adaptive', + currentInflight: 2, + minInflight: 1, + maxInflight: 3, + currentCoverageBatch: 7, + configuredCoverageBatch: 7, + }); + }); + + it('keeps an explicit Core limit static unless adaptation is explicitly enabled', () => { + const staticRuntime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncGlobalMaxInflight: 6, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + const adaptiveRuntime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncGlobalMaxInflight: 6, + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(staticRuntime.isAdaptive()).toBe(false); + expect(staticRuntime.policy.limit).toBe(6); + expect(adaptiveRuntime.isAdaptive()).toBe(true); + expect(adaptiveRuntime.policy.limit).toBe(6); + expect(adaptiveRuntime.getStatus().maxInflight).toBe(3); + }); + + it('retains explicit zero as unbounded and disables adaptation', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncGlobalMaxInflight: 0, + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(runtime.isAdaptive()).toBe(false); + expect(runtime.policy.limit).toBeUndefined(); + expect(runtime.getStatus().currentInflight).toBeNull(); + }); + + it('honors the adaptive environment disable over inferred Core defaults', () => { + vi.stubEnv('DKG_SYNC_ADAPTIVE_CAPACITY_ENABLED', '0'); + const runtime = SyncCapacityRuntime.create({ nodeRole: 'core' }, fakeStore(STORE_PRESSURE)); + expect(runtime.isAdaptive()).toBe(false); + }); +}); + From 5c91a344e50309d17c6746289e9b6969d5f8dfcf Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 03:41:40 +0200 Subject: [PATCH 02/12] feat(sync): apply adaptive Core capacity --- packages/agent/src/dkg-agent-base.ts | 13 ++++++++++ packages/agent/src/dkg-agent-lifecycle.ts | 28 ++++++++++++++++++--- packages/agent/src/dkg-agent.ts | 4 +++ packages/agent/src/sync/capacity-runtime.ts | 2 ++ packages/cli/src/daemon/routes/status.ts | 16 ++++++++++++ packages/cli/test/status-route-rpc.test.ts | 17 +++++++++++++ 6 files changed, 76 insertions(+), 4 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index e65192931..87633e1d1 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -164,6 +164,10 @@ import { resolveCorePublicSyncBatchSize, type CorePublicSyncCoverageStatus, } from './sync/core-public-coverage-scheduler.js'; +import { + SyncCapacityRuntime, + type SyncCapacityStatus, +} from './sync/capacity-runtime.js'; import { bindRandomSampling, type RandomSamplingDisabledReason, type RandomSamplingHandle, type RandomSamplingStatus } from './random-sampling-bind.js'; import { connectToMultiaddr, ensurePeerConnected as ensurePeerConnectedAtom, primeCatchupConnections as primeCatchupConnectionsAtom } from './p2p/peer-connect.js'; import { Messenger, type SloProtocolStats } from './p2p/messenger.js'; @@ -1060,6 +1064,8 @@ export class DKGAgentBase { * capped by this scheduler; Edge nodes never register automatic coverage. */ protected readonly corePublicSyncCoverageScheduler: CorePublicSyncCoverageScheduler; + /** Role-aware requester admission and Core automatic-coverage capacity. */ + protected readonly syncCapacityRuntime: SyncCapacityRuntime; protected started = false; /** * One OT-RFC-64 persistence owner for the inventory lease and every resource @@ -1555,6 +1561,7 @@ export class DKGAgentBase { */ protected readonly syncReconcilerBackoff = new Map(); protected syncReconcilerTimer: ReturnType | null = null; + protected syncAdaptiveCapacityTimer: ReturnType | null = null; /** A.4-lite+: periodic warm/pinned Core-connection reconcile (opt-in). */ protected warmCoreTimer: ReturnType | null = null; /** Cores keep-alive-pinned on the last warm-core pass, so the next pass can @@ -1630,6 +1637,7 @@ export class DKGAgentBase { this.wallet = wallet; this.node = node; this.store = store; + this.syncCapacityRuntime = SyncCapacityRuntime.create(config, store); this.contextGraphMetaProjection = new ContextGraphMetaProjection(store); this.publisher = publisher; this.queryEngine = queryEngine; @@ -1721,6 +1729,7 @@ export class DKGAgentBase { selected, this.config.syncContextGraphPriorities, remotePeer, + this.syncCapacityRuntime.getEffectiveCoverageBatch(), ) : []; const initialDurableContextGraphIds = [...new Set([ @@ -1743,6 +1752,10 @@ export class DKGAgentBase { ); } + getSyncCapacityStatus(): SyncCapacityStatus { + return this.syncCapacityRuntime.getStatus(); + } + /** * Acquire the RFC-64 inventory, finish bounded stale-candidate cleanup, and * open the inherited-owner control-object tree before network consumers. diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 0c2320947..499114e4a 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -307,9 +307,9 @@ import { getSyncBackpressureBusyError, resolveBooleanSwitch, resolveNonNegativeIntegerSwitch, - resolveSyncGlobalBackpressure, withGlobalSyncBackpressure, } from './sync/backpressure.js'; +import { DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS } from './sync/capacity-runtime.js'; import { contextGraphPriority, countSyncPriorityClasses, @@ -1235,7 +1235,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { try { return await withGlobalSyncBackpressure( { - policy: resolveSyncGlobalBackpressure(this.config), + policy: this.syncCapacityRuntime.policy, ctx, label, contextGraphId, @@ -1245,6 +1245,9 @@ export class LifecycleSyncMethods extends DKGAgentBase { source, signal: admissionBoundary.signal, logInfo: (opCtx, message) => this.log.info(opCtx, message), + ...(this.syncCapacityRuntime.isAdaptive() + ? { currentLimit: () => this.syncCapacityRuntime.getCurrentInflight()! } + : {}), }, work, ); @@ -2021,7 +2024,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { ); }, onDecline: (details) => { - const syncPressure = getSyncBackpressureSnapshot(resolveSyncGlobalBackpressure(this.config)); + const syncPressure = getSyncBackpressureSnapshot(this.syncCapacityRuntime.policy); const syncPressureLabel = `syncGlobalInflight=${syncPressure.inflight} ` + `syncGlobalQueued=${syncPressure.queued} ` + @@ -2544,7 +2547,8 @@ export class LifecycleSyncMethods extends DKGAgentBase { process.env, (message) => this.log.warn(ctx, message), ); - const syncGlobalPolicy = resolveSyncGlobalBackpressure(this.config); + const syncGlobalPolicy = this.syncCapacityRuntime.policy; + const syncCapacity = this.syncCapacityRuntime.getStatus(); const configuredPriorityCounts = countSyncPriorityClasses(this.config.syncContextGraphPriorities); this.log.info(ctx, `Resolved sync policy ${JSON.stringify({ snapshotGlobalRows: snapshotPolicy.budget.maxRows, @@ -2553,6 +2557,9 @@ export class LifecycleSyncMethods extends DKGAgentBase { snapshotLocalBytesEstimate: snapshotPolicy.budget.maxSnapshotBytesEstimate, syncGlobalInflightLimit: syncGlobalPolicy.limit ?? 0, syncGlobalQueueLimit: syncGlobalPolicy.queueLimit ?? 0, + syncCapacityMode: syncCapacity.mode, + syncCapacityCurrentInflight: syncCapacity.currentInflight ?? 0, + syncCapacityCoverageBatch: syncCapacity.currentCoverageBatch, configuredPriorities: configuredPriorityCounts, snapshotLocalClamped: snapshotPolicy.localRowsClamped || snapshotPolicy.localBytesEstimateClamped, })}`); @@ -3249,6 +3256,19 @@ export class LifecycleSyncMethods extends DKGAgentBase { this.log.warn(ctx, `Skipping periodic sync reconciler startup (DKG_SYNC_RECONCILER_ENABLED=0)`); } + if (this.syncCapacityRuntime.isAdaptive()) { + this.syncAdaptiveCapacityTimer = setInterval(() => { + try { + const pressure = getSyncBackpressureSnapshot(this.syncCapacityRuntime.policy); + this.syncCapacityRuntime.sample(pressure.inflight > 0 || pressure.queued > 0); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.log.warn(ctx, `Adaptive sync capacity sample failed: ${message}`); + } + }, DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS); + if (this.syncAdaptiveCapacityTimer.unref) this.syncAdaptiveCapacityTimer.unref(); + } + // A.4-lite+: keep a small set of Core nodes warm (connection pinned + // auto-redialed by libp2p) so catch-up / chain reconciliation never pays // a cold circuit-relay dial to reach a Core. Opt-in via diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index faf5e20df..81e8ede44 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -1720,6 +1720,10 @@ export class DKGAgent extends DKGAgentBase { async stop(): Promise { if (!this.started) return; + if (this.syncAdaptiveCapacityTimer) { + clearInterval(this.syncAdaptiveCapacityTimer); + this.syncAdaptiveCapacityTimer = null; + } if (this.chainPoller) { // Await so any in-flight poll (and its HTTP keep-alive socket) settles // BEFORE we tear down the chain adapter — otherwise the RPC connection diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 548578554..69564e75f 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -23,6 +23,8 @@ import { } from './backpressure.js'; import { resolveCorePublicSyncBatchSize } from './core-public-coverage-scheduler.js'; +export const DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS = 5_000; + export interface SyncCapacityRuntimeConfig extends SyncGlobalBackpressureConfig { nodeRole?: 'core' | 'edge'; syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig; diff --git a/packages/cli/src/daemon/routes/status.ts b/packages/cli/src/daemon/routes/status.ts index 3a8264138..1e0bd52fa 100644 --- a/packages/cli/src/daemon/routes/status.ts +++ b/packages/cli/src/daemon/routes/status.ts @@ -718,6 +718,21 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { trackedContextGraphs: 0, planningLanes: 0, }; + const fallbackSyncLimit = config.syncGlobalMaxInflight ?? config.syncGlobalLimit ?? 2; + const fallbackCoverageBatch = config.syncCorePublicBatchSize ?? 8; + const syncCapacity = typeof agent.getSyncCapacityStatus === 'function' + ? agent.getSyncCapacityStatus() + : { + mode: 'static' as const, + state: 'healthy' as const, + currentInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, + minInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, + maxInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, + currentCoverageBatch: fallbackCoverageBatch, + configuredCoverageBatch: fallbackCoverageBatch, + storePressureTelemetryAvailable: false, + lastDecision: null, + }; return jsonResponse(res, 200, { name: config.name, version: nodeVersion, @@ -783,6 +798,7 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { diagnosticsAvailable: '/api/diagnostics/backpressure', }, corePublicSyncCoverage, + syncCapacity, connectedPeers: uniquePeers.size, connections: { total: allConns.length, diff --git a/packages/cli/test/status-route-rpc.test.ts b/packages/cli/test/status-route-rpc.test.ts index ee4e21bd5..9d5c65f1d 100644 --- a/packages/cli/test/status-route-rpc.test.ts +++ b/packages/cli/test/status-route-rpc.test.ts @@ -235,12 +235,29 @@ describe('/api/status Core public synchronization coverage', () => { totalContextGraphs: 10, }, }; + const syncCapacity = { + mode: 'adaptive' as const, + state: 'cooldown' as const, + currentInflight: 3, + minInflight: 1, + maxInflight: 6, + currentCoverageBatch: 4, + configuredCoverageBatch: 8, + storePressureTelemetryAvailable: true, + lastDecision: { + action: 'increase' as const, + reason: 'healthy_hysteresis' as const, + atMs: 123, + }, + }; const response = await requestStatusWithAgent({ getCorePublicSyncCoverageStatus: () => corePublicSyncCoverage, + getSyncCapacityStatus: () => syncCapacity, }); expect(response.status).toBe(200); expect(response.body.corePublicSyncCoverage).toEqual(corePublicSyncCoverage); + expect(response.body.syncCapacity).toEqual(syncCapacity); }); }); From a45bf32c50a7923a60b75583d9f1e8a9dbfacb28 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 03:53:32 +0200 Subject: [PATCH 03/12] fix(sync): reject invalid adaptive bounds --- packages/agent/src/sync/capacity-runtime.ts | 44 ++++++++++------- .../agent/test/sync-capacity-runtime.test.ts | 47 ++++++++++++++++++- packages/agent/vitest.unit.config.ts | 1 + 3 files changed, 74 insertions(+), 18 deletions(-) diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 69564e75f..147c8c90b 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -15,8 +15,7 @@ import { import { notifyGlobalSyncBackpressureCapacityChanged, parseBooleanEnv, - resolveNonNegativeIntegerSwitch, - resolvePositiveIntegerSwitch, + resolveExplicitSyncGlobalLimit, resolveSyncGlobalBackpressure, type SyncGlobalBackpressureConfig, type SyncGlobalBackpressurePolicy, @@ -52,18 +51,6 @@ export interface SyncCapacityRuntimeOptions { now?: () => number; } -function resolvedExplicitGlobalLimit( - config: SyncGlobalBackpressureConfig, -): number | undefined { - return resolveNonNegativeIntegerSwitch( - config.syncGlobalMaxInflight, - 'DKG_SYNC_GLOBAL_MAX_INFLIGHT', - ) ?? resolveNonNegativeIntegerSwitch( - config.syncGlobalLimit, - 'DKG_SYNC_GLOBAL_LIMIT', - ); -} - function readStorePressure(store: TripleStore) { try { return store.getPressureSnapshot?.(); @@ -72,6 +59,27 @@ function readStorePressure(store: TripleStore) { } } +function requirePositiveInteger(value: unknown, label: string): number { + const parsed = typeof value === 'string' ? Number(value.trim()) : value; + if (typeof parsed !== 'number' || !Number.isInteger(parsed) || parsed <= 0) { + throw new Error(`${label} must be a positive integer`); + } + return parsed; +} + +function resolveAdaptivePositiveInteger( + configValue: number | undefined, + envName: string, + configName: string, +): number | undefined { + const envValue = process.env[envName]; + if (envValue !== undefined) return requirePositiveInteger(envValue, envName); + if (configValue !== undefined) { + return requirePositiveInteger(configValue, `syncAdaptiveCapacity.${configName}`); + } + return undefined; +} + /** One agent-owned capacity policy; static callers keep the exact old path. */ export class SyncCapacityRuntime { private readonly controller?: AdaptiveCapacityController; @@ -96,7 +104,7 @@ export class SyncCapacityRuntime { config.syncCorePublicBatchSize, ); const staticPolicy = resolveSyncGlobalBackpressure(config); - const explicitGlobalLimit = resolvedExplicitGlobalLimit(config); + const explicitGlobalLimit = resolveExplicitSyncGlobalLimit(config); const explicitlyEnabled = parseBooleanEnv('DKG_SYNC_ADAPTIVE_CAPACITY_ENABLED') ?? config.syncAdaptiveCapacity?.enabled; const adaptive = (config.nodeRole ?? 'edge') === 'core' @@ -106,13 +114,15 @@ export class SyncCapacityRuntime { return new SyncCapacityRuntime(staticPolicy, configuredCoverageBatch); } - const requestedMax = resolvePositiveIntegerSwitch( + const requestedMax = resolveAdaptivePositiveInteger( config.syncAdaptiveCapacity?.maxInflight, 'DKG_SYNC_ADAPTIVE_MAX_INFLIGHT', + 'maxInflight', ); - const minInflight = resolvePositiveIntegerSwitch( + const minInflight = resolveAdaptivePositiveInteger( config.syncAdaptiveCapacity?.minInflight, 'DKG_SYNC_ADAPTIVE_MIN_INFLIGHT', + 'minInflight', ); const hardMax = deriveAdaptiveInflightHardMax({ operatorMax: Math.min( diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 222a6c74c..379d75688 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -73,6 +73,52 @@ describe('sync capacity runtime resolution', () => { expect(adaptiveRuntime.getStatus().maxInflight).toBe(3); }); + it('uses the legacy environment limit ahead of newer config for adaptive policy and status', () => { + vi.stubEnv('DKG_SYNC_GLOBAL_LIMIT', '1'); + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncGlobalMaxInflight: 6, + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(runtime.isAdaptive()).toBe(true); + expect(runtime.policy.limit).toBe(1); + expect(runtime.getStatus()).toMatchObject({ + mode: 'adaptive', + currentInflight: 1, + minInflight: 1, + maxInflight: 1, + }); + }); + + it.each([0, -1, 1.5])('rejects invalid configured adaptive max %s', (maxInflight) => { + expect(() => SyncCapacityRuntime.create({ + nodeRole: 'core', + syncAdaptiveCapacity: { maxInflight }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + 'syncAdaptiveCapacity.maxInflight must be a positive integer', + ); + }); + + it('rejects an invalid adaptive max environment override instead of dropping a valid config ceiling', () => { + vi.stubEnv('DKG_SYNC_ADAPTIVE_MAX_INFLIGHT', '0'); + expect(() => SyncCapacityRuntime.create({ + nodeRole: 'core', + syncAdaptiveCapacity: { maxInflight: 1 }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + 'DKG_SYNC_ADAPTIVE_MAX_INFLIGHT must be a positive integer', + ); + }); + + it('rejects an invalid adaptive minimum instead of silently widening the controller bounds', () => { + vi.stubEnv('DKG_SYNC_ADAPTIVE_MIN_INFLIGHT', '1.5'); + expect(() => SyncCapacityRuntime.create({ + nodeRole: 'core', + }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + 'DKG_SYNC_ADAPTIVE_MIN_INFLIGHT must be a positive integer', + ); + }); + it('retains explicit zero as unbounded and disables adaptation', () => { const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', @@ -91,4 +137,3 @@ describe('sync capacity runtime resolution', () => { expect(runtime.isAdaptive()).toBe(false); }); }); - diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index e1e677ed5..4c2ed0ee4 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -88,6 +88,7 @@ export default defineConfig({ "test/core-public-coverage-scheduler.test.ts", "test/adaptive-capacity.test.ts", "test/adaptive-capacity-sampler.test.ts", + "test/sync-capacity-runtime.test.ts", "test/sync-requester-progress.test.ts", "test/rootless-durable-bounded-progress.test.ts", "test/rootless-durable-skips-legacy-partition.test.ts", From 1df88e12ce277d68a0ff3175c5683533836d99f9 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 04:35:32 +0200 Subject: [PATCH 04/12] fix(sync): bound adaptive admission backlog --- packages/agent/src/sync/capacity-runtime.ts | 6 +++++- .../agent/test/sync-capacity-runtime.test.ts | 17 ++++++++++++++++- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 147c8c90b..727a8b48c 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -143,7 +143,11 @@ export class SyncCapacityRuntime { } const adaptivePolicy = resolveSyncGlobalBackpressure({ - syncGlobalMaxInflight: explicitGlobalLimit ?? hardMax, + // The controller can never exceed hardMax, so use that same ceiling for + // queue sizing and admission observability. Basing the queue on a larger + // explicit operator limit would allow a backlog that the adaptive Core + // can never drain at the advertised policy capacity. + syncGlobalMaxInflight: hardMax, syncGlobalQueueLimit: config.syncGlobalQueueLimit, }); return new SyncCapacityRuntime( diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 379d75688..49212a502 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -69,10 +69,25 @@ describe('sync capacity runtime resolution', () => { expect(staticRuntime.isAdaptive()).toBe(false); expect(staticRuntime.policy.limit).toBe(6); expect(adaptiveRuntime.isAdaptive()).toBe(true); - expect(adaptiveRuntime.policy.limit).toBe(6); + expect(adaptiveRuntime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); expect(adaptiveRuntime.getStatus().maxInflight).toBe(3); }); + it('sizes the default adaptive queue from the effective hard maximum', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncGlobalMaxInflight: 100, + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); + + expect(runtime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); + expect(runtime.getStatus()).toMatchObject({ + mode: 'adaptive', + currentInflight: 2, + maxInflight: 3, + }); + }); + it('uses the legacy environment limit ahead of newer config for adaptive policy and status', () => { vi.stubEnv('DKG_SYNC_GLOBAL_LIMIT', '1'); const runtime = SyncCapacityRuntime.create({ From e53e3717835152b1a6ee5c88cd101dcdf8602f70 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 04:40:22 +0200 Subject: [PATCH 05/12] fix(sync): centralize adaptive runtime demand --- packages/agent/src/dkg-agent-base.ts | 1 - packages/agent/src/dkg-agent-lifecycle.ts | 30 ++--- packages/agent/src/dkg-agent.ts | 5 +- packages/agent/src/sync/capacity-runtime.ts | 52 ++++++- .../sync/core-public-coverage-scheduler.ts | 12 ++ .../core-public-coverage-scheduler.test.ts | 13 ++ .../agent/test/sync-capacity-runtime.test.ts | 127 +++++++++++++++++- packages/cli/src/daemon/routes/status.ts | 16 +-- packages/cli/test/status-route-rpc.test.ts | 14 ++ .../cli/test/status-route-store-quads.test.ts | 12 ++ 10 files changed, 242 insertions(+), 40 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index 87633e1d1..da964fa3e 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1561,7 +1561,6 @@ export class DKGAgentBase { */ protected readonly syncReconcilerBackoff = new Map(); protected syncReconcilerTimer: ReturnType | null = null; - protected syncAdaptiveCapacityTimer: ReturnType | null = null; /** A.4-lite+: periodic warm/pinned Core-connection reconcile (opt-in). */ protected warmCoreTimer: ReturnType | null = null; /** Cores keep-alive-pinned on the last warm-core pass, so the next pass can diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 499114e4a..fc7cf076b 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -309,7 +309,6 @@ import { resolveNonNegativeIntegerSwitch, withGlobalSyncBackpressure, } from './sync/backpressure.js'; -import { DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS } from './sync/capacity-runtime.js'; import { contextGraphPriority, countSyncPriorityClasses, @@ -1233,9 +1232,10 @@ export class LifecycleSyncMethods extends DKGAgentBase { operationSignal, ); try { + const capacityAdmission = this.syncCapacityRuntime.getAdmissionOptions(); return await withGlobalSyncBackpressure( { - policy: this.syncCapacityRuntime.policy, + ...capacityAdmission, ctx, label, contextGraphId, @@ -1245,9 +1245,6 @@ export class LifecycleSyncMethods extends DKGAgentBase { source, signal: admissionBoundary.signal, logInfo: (opCtx, message) => this.log.info(opCtx, message), - ...(this.syncCapacityRuntime.isAdaptive() - ? { currentLimit: () => this.syncCapacityRuntime.getCurrentInflight()! } - : {}), }, work, ); @@ -3256,18 +3253,17 @@ export class LifecycleSyncMethods extends DKGAgentBase { this.log.warn(ctx, `Skipping periodic sync reconciler startup (DKG_SYNC_RECONCILER_ENABLED=0)`); } - if (this.syncCapacityRuntime.isAdaptive()) { - this.syncAdaptiveCapacityTimer = setInterval(() => { - try { - const pressure = getSyncBackpressureSnapshot(this.syncCapacityRuntime.policy); - this.syncCapacityRuntime.sample(pressure.inflight > 0 || pressure.queued > 0); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - this.log.warn(ctx, `Adaptive sync capacity sample failed: ${message}`); - } - }, DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS); - if (this.syncAdaptiveCapacityTimer.unref) this.syncAdaptiveCapacityTimer.unref(); - } + this.syncCapacityRuntime.startSampling({ + hasSupplementalDemand: () => ( + this.corePublicSyncCoverageScheduler.hasAutomaticCoverageBacklog( + this.syncCapacityRuntime.getEffectiveCoverageBatch(), + ) + ), + onError: (error) => { + const message = error instanceof Error ? error.message : String(error); + this.log.warn(ctx, `Adaptive sync capacity sample failed: ${message}`); + }, + }); // A.4-lite+: keep a small set of Core nodes warm (connection pinned + // auto-redialed by libp2p) so catch-up / chain reconciliation never pays diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 81e8ede44..9a8dcbd95 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -1720,10 +1720,7 @@ export class DKGAgent extends DKGAgentBase { async stop(): Promise { if (!this.started) return; - if (this.syncAdaptiveCapacityTimer) { - clearInterval(this.syncAdaptiveCapacityTimer); - this.syncAdaptiveCapacityTimer = null; - } + this.syncCapacityRuntime.stopSampling(); if (this.chainPoller) { // Await so any in-flight poll (and its HTTP keep-alive socket) settles // BEFORE we tear down the chain adapter — otherwise the RPC connection diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 727a8b48c..7df5ebf8c 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -13,6 +13,7 @@ import { type AdaptiveCapacitySamplerDependencies, } from './adaptive-capacity-sampler.js'; import { + getSyncBackpressureSnapshot, notifyGlobalSyncBackpressureCapacityChanged, parseBooleanEnv, resolveExplicitSyncGlobalLimit, @@ -51,6 +52,13 @@ export interface SyncCapacityRuntimeOptions { now?: () => number; } +export interface SyncCapacitySamplingOptions { + /** Additional Core work that does not currently occupy requester admission. */ + hasSupplementalDemand?: () => boolean; + intervalMs?: number; + onError?: (error: unknown) => void; +} + function readStorePressure(store: TripleStore) { try { return store.getPressureSnapshot?.(); @@ -84,6 +92,7 @@ function resolveAdaptivePositiveInteger( export class SyncCapacityRuntime { private readonly controller?: AdaptiveCapacityController; private readonly sampler?: AdaptiveCapacitySampler; + private samplingTimer: ReturnType | undefined; private constructor( readonly policy: SyncGlobalBackpressurePolicy, @@ -162,8 +171,15 @@ export class SyncCapacityRuntime { return this.controller !== undefined; } - getCurrentInflight(): number | undefined { - return this.controller?.getCurrentInflight(); + /** Stable admission contract; callers do not need to branch on capacity mode. */ + getAdmissionOptions(): { + policy: SyncGlobalBackpressurePolicy; + currentLimit?: () => number; + } { + const controller = this.controller; + return controller + ? { policy: this.policy, currentLimit: () => controller.getCurrentInflight() } + : { policy: this.policy }; } getEffectiveCoverageBatch(): number { @@ -179,6 +195,38 @@ export class SyncCapacityRuntime { } } + /** + * Own the adaptive sampling lifecycle and requester-demand calculation. + * Static runtimes deliberately no-op so lifecycle callers stay mode-agnostic. + */ + startSampling(options: SyncCapacitySamplingOptions = {}): boolean { + if (!this.controller || !this.sampler || this.samplingTimer) return false; + const intervalMs = requirePositiveInteger( + options.intervalMs ?? DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS, + 'sync capacity sample interval', + ); + this.samplingTimer = setInterval(() => { + try { + const pressure = getSyncBackpressureSnapshot(this.policy); + const demand = pressure.inflight > 0 + || pressure.queued > 0 + || (options.hasSupplementalDemand?.() ?? false); + this.sample(demand); + } catch (error) { + options.onError?.(error); + } + }, intervalMs); + this.samplingTimer.unref?.(); + return true; + } + + stopSampling(): boolean { + if (!this.samplingTimer) return false; + clearInterval(this.samplingTimer); + this.samplingTimer = undefined; + return true; + } + getStatus(): SyncCapacityStatus { if (!this.controller) { const currentInflight = this.policy.limit ?? null; diff --git a/packages/agent/src/sync/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index 2b295f650..a9035bf17 100644 --- a/packages/agent/src/sync/core-public-coverage-scheduler.ts +++ b/packages/agent/src/sync/core-public-coverage-scheduler.ts @@ -90,6 +90,7 @@ export class CorePublicSyncCoverageScheduler { private readonly laneAnchors = new Map(); private lastPlanAt?: number; private lastPlan?: CorePublicSyncCoverageStatus['lastPlan']; + private lastCoverageCandidates?: number; constructor( private readonly batchSize: number, @@ -107,11 +108,13 @@ export class CorePublicSyncCoverageScheduler { if (!normalized) return false; const sizeBefore = this.tracked.size; this.tracked.add(normalized); + if (this.tracked.size !== sizeBefore) this.lastCoverageCandidates = undefined; return this.tracked.size !== sizeBefore; } unregister(contextGraphId: string): boolean { const removed = this.tracked.delete(contextGraphId); + if (removed) this.lastCoverageCandidates = undefined; if (this.tracked.size === 0) { this.laneAnchors.clear(); } else if (removed) { @@ -161,6 +164,7 @@ export class CorePublicSyncCoverageScheduler { const scheduledCoverage: string[] = []; const batchSize = resolveEffectiveBatchSize(this.batchSize, effectiveBatchSize); + this.lastCoverageCandidates = coverage.length; if (batchSize > 0 && coverage.length > 0) { const count = Math.min(batchSize, coverage.length); const previousAnchor = this.laneAnchors.get(planningLane); @@ -194,6 +198,14 @@ export class CorePublicSyncCoverageScheduler { return scheduledCoverage; } + /** True when automatic public coverage was truncated by the current batch. */ + hasAutomaticCoverageBacklog(effectiveBatchSize?: number): boolean { + if (this.batchSize === 0) return false; + const batchSize = resolveEffectiveBatchSize(this.batchSize, effectiveBatchSize); + const candidates = this.lastCoverageCandidates ?? this.tracked.size; + return candidates > batchSize; + } + getStatus(enabled: boolean): CorePublicSyncCoverageStatus { return { enabled: enabled && this.batchSize > 0, diff --git a/packages/agent/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index 489fd5407..f32bf88e7 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -137,6 +137,7 @@ describe('Core public Context Graph coverage scheduler', () => { scheduler.register('public-a'); expect(scheduler.planAutomaticCoverage(['selected'])).toEqual([]); + expect(scheduler.hasAutomaticCoverageBacklog(0)).toBe(false); expect(scheduler.getStatus(true)).toMatchObject({ enabled: false, batchSize: 0, @@ -189,6 +190,18 @@ describe('Core public Context Graph coverage scheduler', () => { expect(third).toEqual(['cg:d']); }); + it('reports automatic coverage demand only while the live batch truncates candidates', () => { + const scheduler = new CorePublicSyncCoverageScheduler(4); + for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d', 'cg:e']) { + scheduler.register(contextGraphId); + } + + expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); + scheduler.planAutomaticCoverage(['cg:a', 'cg:b'], undefined, 'peer-a', 2); + expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); + expect(scheduler.hasAutomaticCoverageBacklog(3)).toBe(false); + }); + it('rejects invalid live automatic-coverage batches', () => { const scheduler = new CorePublicSyncCoverageScheduler(3); scheduler.register('cg:a'); diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 49212a502..4242493f1 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -1,6 +1,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createOperationContext } from '@origintrail-official/dkg-core'; import type { StorePressureSnapshot, TripleStore } from '@origintrail-official/dkg-storage'; import { SyncCapacityRuntime } from '../src/sync/capacity-runtime.js'; +import { CorePublicSyncCoverageScheduler } from '../src/sync/core-public-coverage-scheduler.js'; +import { withGlobalSyncBackpressure } from '../src/sync/backpressure.js'; function fakeStore(pressure?: StorePressureSnapshot): TripleStore { return { @@ -22,7 +25,10 @@ const STORE_PRESSURE: StorePressureSnapshot = { healthReservedSlots: 1, }; -afterEach(() => vi.unstubAllEnvs()); +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllEnvs(); +}); describe('sync capacity runtime resolution', () => { it('keeps Edge nodes on the exact static policy even if adaptive is requested', () => { @@ -34,6 +40,7 @@ describe('sync capacity runtime resolution', () => { expect(runtime.isAdaptive()).toBe(false); expect(runtime.policy.limit).toBe(4); + expect(runtime.getAdmissionOptions()).toEqual({ policy: runtime.policy }); expect(runtime.getStatus()).toMatchObject({ mode: 'static', currentInflight: 4 }); }); @@ -53,6 +60,124 @@ describe('sync capacity runtime resolution', () => { currentCoverageBatch: 7, configuredCoverageBatch: 7, }); + expect(runtime.getAdmissionOptions().currentLimit?.()).toBe(2); + }); + + it('owns sampling and restores constrained coverage from supplemental Core demand', () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + let heapRatio = 0.82; + let cpuIdle = 0; + let cpuTotal = 0; + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncCorePublicBatchSize: 8, + }, fakeStore(STORE_PRESSURE), { + parallelism: 16, + samplerDependencies: { + readCpuTimes: () => { + cpuIdle += 80; + cpuTotal += 100; + return { idle: cpuIdle, total: cpuTotal }; + }, + readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), + eventLoopUtilizationDelta: () => 0.2, + readHeapRatio: () => heapRatio, + }, + }); + const coverage = new CorePublicSyncCoverageScheduler(8); + for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d', 'cg:e']) { + coverage.register(contextGraphId); + } + + runtime.sample(true); + expect(runtime.getStatus()).toMatchObject({ + currentInflight: 1, + currentCoverageBatch: 4, + lastDecision: { action: 'halve', reason: 'critical_heap' }, + }); + + heapRatio = 0.3; + expect(runtime.startSampling({ + hasSupplementalDemand: () => coverage.hasAutomaticCoverageBacklog( + runtime.getEffectiveCoverageBatch(), + ), + intervalMs: 5_000, + onError: (error) => { throw error; }, + })).toBe(true); + vi.advanceTimersByTime(30_000); + + expect(runtime.getStatus()).toMatchObject({ + currentInflight: 2, + currentCoverageBatch: 5, + lastDecision: { action: 'increase', reason: 'healthy_hysteresis' }, + }); + expect(runtime.stopSampling()).toBe(true); + expect(runtime.stopSampling()).toBe(false); + }); + + it('pumps queued requester work when a healthy sample raises live capacity', async () => { + let now = 0; + let heapRatio = 0.82; + let cpuIdle = 0; + let cpuTotal = 0; + const runtime = SyncCapacityRuntime.create({ nodeRole: 'core' }, fakeStore(STORE_PRESSURE), { + parallelism: 16, + now: () => now, + samplerDependencies: { + readCpuTimes: () => { + cpuIdle += 80; + cpuTotal += 100; + return { idle: cpuIdle, total: cpuTotal }; + }, + readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), + eventLoopUtilizationDelta: () => 0.2, + readHeapRatio: () => heapRatio, + }, + }); + runtime.sample(true); + expect(runtime.getStatus().currentInflight).toBe(1); + + let releaseFirst!: () => void; + let markFirstStarted!: () => void; + const firstBlock = new Promise((resolve) => { releaseFirst = resolve; }); + const firstStarted = new Promise((resolve) => { markFirstStarted = resolve; }); + const ctx = createOperationContext('test'); + const first = withGlobalSyncBackpressure({ + ...runtime.getAdmissionOptions(), + ctx, + label: 'durable:first', + }, async () => { + markFirstStarted(); + await firstBlock; + }); + await firstStarted; + + let markSecondStarted!: () => void; + const secondStarted = new Promise((resolve) => { markSecondStarted = resolve; }); + let secondRan = false; + const second = withGlobalSyncBackpressure({ + ...runtime.getAdmissionOptions(), + ctx, + label: 'durable:second', + }, async () => { + secondRan = true; + markSecondStarted(); + }); + await Promise.resolve(); + expect(secondRan).toBe(false); + + heapRatio = 0.3; + for (let sample = 1; sample <= 6; sample += 1) { + now = sample * 5_000; + runtime.sample(true); + } + await secondStarted; + expect(runtime.getStatus().currentInflight).toBe(2); + expect(secondRan).toBe(true); + + releaseFirst(); + await Promise.all([first, second]); }); it('keeps an explicit Core limit static unless adaptation is explicitly enabled', () => { diff --git a/packages/cli/src/daemon/routes/status.ts b/packages/cli/src/daemon/routes/status.ts index 1e0bd52fa..2ff56f61d 100644 --- a/packages/cli/src/daemon/routes/status.ts +++ b/packages/cli/src/daemon/routes/status.ts @@ -718,21 +718,7 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { trackedContextGraphs: 0, planningLanes: 0, }; - const fallbackSyncLimit = config.syncGlobalMaxInflight ?? config.syncGlobalLimit ?? 2; - const fallbackCoverageBatch = config.syncCorePublicBatchSize ?? 8; - const syncCapacity = typeof agent.getSyncCapacityStatus === 'function' - ? agent.getSyncCapacityStatus() - : { - mode: 'static' as const, - state: 'healthy' as const, - currentInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, - minInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, - maxInflight: fallbackSyncLimit === 0 ? null : fallbackSyncLimit, - currentCoverageBatch: fallbackCoverageBatch, - configuredCoverageBatch: fallbackCoverageBatch, - storePressureTelemetryAvailable: false, - lastDecision: null, - }; + const syncCapacity = agent.getSyncCapacityStatus(); return jsonResponse(res, 200, { name: config.name, version: nodeVersion, diff --git a/packages/cli/test/status-route-rpc.test.ts b/packages/cli/test/status-route-rpc.test.ts index 9d5c65f1d..2840dba9e 100644 --- a/packages/cli/test/status-route-rpc.test.ts +++ b/packages/cli/test/status-route-rpc.test.ts @@ -49,6 +49,17 @@ const DISABLED_PUBLISHER_STATE: RequestContext['publisherState'] = { operatorActionRequired: true, }, }; +const STATIC_SYNC_CAPACITY = { + mode: 'static' as const, + state: 'healthy' as const, + currentInflight: 2, + minInflight: 2, + maxInflight: 2, + currentCoverageBatch: 8, + configuredCoverageBatch: 8, + storePressureTelemetryAvailable: false, + lastDecision: null, +}; async function requestStatusWithAgent( agentOverrides: Record, @@ -76,6 +87,7 @@ async function requestStatusWithAgent( getRelayStats: () => null, }, publisher: { getIdentityId: () => 0n }, + getSyncCapacityStatus: () => STATIC_SYNC_CAPACITY, ...agentOverrides, }, nodeVersion: '0.0.0-test', @@ -290,6 +302,7 @@ describe('/api/status selected overlay details', () => { getRelayStats: () => null, }, publisher: { getIdentityId: () => 0n }, + getSyncCapacityStatus: () => STATIC_SYNC_CAPACITY, }, nodeVersion: '0.0.0-test', nodeCommit: '', @@ -358,6 +371,7 @@ describe('/api/status selected overlay details', () => { multiaddrs: [], node: { libp2p: { getConnections: () => [] }, getRelayStats: () => null }, publisher: { getIdentityId: () => 0n }, + getSyncCapacityStatus: () => STATIC_SYNC_CAPACITY, }, nodeVersion: '0.0.0-test', nodeCommit: '', diff --git a/packages/cli/test/status-route-store-quads.test.ts b/packages/cli/test/status-route-store-quads.test.ts index 2a07beec1..959692e48 100644 --- a/packages/cli/test/status-route-store-quads.test.ts +++ b/packages/cli/test/status-route-store-quads.test.ts @@ -16,6 +16,17 @@ const DISABLED_PUBLISHER_STATE: RequestContext['publisherState'] = { operatorActionRequired: true, }, }; +const STATIC_SYNC_CAPACITY = { + mode: 'static' as const, + state: 'healthy' as const, + currentInflight: 2, + minInflight: 2, + maxInflight: 2, + currentCoverageBatch: 8, + configuredCoverageBatch: 8, + storePressureTelemetryAvailable: false, + lastDecision: null, +}; interface Deferred { promise: Promise; @@ -65,6 +76,7 @@ async function startStatusServer(query: () => Promise): Promise<{ getRelayStats: () => null, }, publisher: { getIdentityId: () => 0n }, + getSyncCapacityStatus: () => STATIC_SYNC_CAPACITY, }, nodeVersion: '0.0.0-test', nodeCommit: '', From af7d062fb2087fe144876984200107a1558aa65e Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:08:24 +0200 Subject: [PATCH 06/12] feat(sync): expose adaptive capacity config --- packages/agent/src/dkg-agent-types.ts | 22 ++++++++++++++++++ packages/agent/src/dkg-agent.ts | 2 ++ packages/agent/src/index.ts | 1 + ...sync-adaptive-capacity-public.typecheck.ts | 23 +++++++++++++++++++ packages/cli/src/config.ts | 6 +++++ packages/cli/src/daemon/lifecycle.ts | 3 +++ .../daemon-sync-agents-meta-wiring.test.ts | 17 ++++++++++++++ 7 files changed, 74 insertions(+) create mode 100644 packages/agent/test/sync-adaptive-capacity-public.typecheck.ts diff --git a/packages/agent/src/dkg-agent-types.ts b/packages/agent/src/dkg-agent-types.ts index 92faf5a04..834de6ed1 100644 --- a/packages/agent/src/dkg-agent-types.ts +++ b/packages/agent/src/dkg-agent-types.ts @@ -1160,6 +1160,23 @@ export interface Rfc64PublicCatalogBootstrapConfigV1 { readonly retryIntervalMs?: number; } +/** + * Optional bounds for role-aware adaptive sync concurrency. + * + * These values are operator policy, not instantaneous capacity: + * `minInflight` is the floor the controller may reduce to and `maxInflight` is + * the ceiling it may grow to. Omitting the whole block preserves role-aware + * defaults; `enabled` can explicitly opt in or out without restating bounds. + */ +export interface SyncAdaptiveCapacityConfig { + /** Explicitly enable or disable adaptive sync capacity. Omit for the role-aware default. */ + readonly enabled?: boolean; + /** Minimum effective global sync concurrency the controller may select. */ + readonly minInflight?: number; + /** Maximum effective global sync concurrency the controller may select. */ + readonly maxInflight?: number; +} + export interface DKGAgentConfig { name: string; /** Selected genesis document. Defaults to the compatibility Base testnet genesis. */ @@ -1272,6 +1289,11 @@ export interface DKGAgentConfig { syncGlobalLimit?: number; /** Max sync jobs waiting behind the global cap. Defaults to 2x the inflight cap. */ syncGlobalQueueLimit?: number; + /** + * Optional role-aware adaptive sync-concurrency policy. Omission preserves + * role-aware defaults and is distinct from an explicit disable. + */ + syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig; /** * Maximum automatically discovered public CGs a Core adds to one peer-sync * round. Explicitly selected CGs are not capped. Defaults to 8; 0 disables diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 9a8dcbd95..89ce17c18 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -361,6 +361,7 @@ import { type DurableSyncResult, type SharedMemorySyncResult, type DKGAgentConfig, + type SyncAdaptiveCapacityConfig, type Rfc64CatalogAccessPolicyAuthorityConfigV1, type DKGAgentACKTransportOptions, type ImportedArtifactByteStore, @@ -473,6 +474,7 @@ export type { SharedMemorySyncDiagnostics, CatchupSyncDiagnostics, DKGAgentConfig, + SyncAdaptiveCapacityConfig, Rfc64CatalogAccessPolicyAuthorityConfigV1, DKGAgentACKTransportOptions, ImportedArtifactByteStore, diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 7ad98d3f8..07005727c 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -194,6 +194,7 @@ export { InvalidContentError, StaleSenderKeyTargetError, type DKGAgentConfig, + type SyncAdaptiveCapacityConfig, type Rfc64CatalogAccessPolicyAuthorityConfigV1, type DKGAgentACKTransportOptions, type ContextGraphSub, diff --git a/packages/agent/test/sync-adaptive-capacity-public.typecheck.ts b/packages/agent/test/sync-adaptive-capacity-public.typecheck.ts new file mode 100644 index 000000000..5ad1d1a24 --- /dev/null +++ b/packages/agent/test/sync-adaptive-capacity-public.typecheck.ts @@ -0,0 +1,23 @@ +import type { + DKGAgentConfig, + SyncAdaptiveCapacityConfig, +} from '@origintrail-official/dkg-agent'; + +// Keep the nested adaptive-capacity knobs available to package-root consumers. +// Real literals make every public field name load-bearing for this typecheck. +const adaptiveCapacity: SyncAdaptiveCapacityConfig = { + enabled: true, + minInflight: 2, + maxInflight: 8, +}; + +const agentConfig: Pick = { + syncAdaptiveCapacity: adaptiveCapacity, +}; + +// Omission is a distinct, supported configuration state: it preserves the +// role-aware default instead of manufacturing an empty policy object. +const omittedAgentConfig: Pick = {}; + +void agentConfig; +void omittedAgentConfig; diff --git a/packages/cli/src/config.ts b/packages/cli/src/config.ts index 181762c9c..e4bb2d8ab 100644 --- a/packages/cli/src/config.ts +++ b/packages/cli/src/config.ts @@ -5,6 +5,7 @@ import { fileURLToPath } from 'node:url'; import yaml from 'js-yaml'; import type { DKGAgentConfig, + SyncAdaptiveCapacityConfig, SyncContextGraphPriorityConfig, SyncResponderSnapshotLimitsConfig, } from '@origintrail-official/dkg-agent'; @@ -628,6 +629,11 @@ export interface DkgConfig { syncGlobalLimit?: number; /** Max sync jobs waiting behind the global cap. Defaults to 2x the inflight cap. */ syncGlobalQueueLimit?: number; + /** + * Optional role-aware adaptive sync-concurrency policy forwarded unchanged + * to the agent. Omit the block to preserve role-aware defaults. + */ + syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig; /** * Public-CG coverage admitted per Core peer-sync round. Explicit Edge/Core * selections are never capped. Default 8; 0 disables automatic Core catch-up. diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index f76f0b108..86446156f 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -1777,6 +1777,9 @@ export async function runDaemonInner( syncGlobalMaxInflight: config.syncGlobalMaxInflight, syncGlobalLimit: config.syncGlobalLimit, syncGlobalQueueLimit: config.syncGlobalQueueLimit, + ...(config.syncAdaptiveCapacity === undefined + ? {} + : { syncAdaptiveCapacity: config.syncAdaptiveCapacity }), syncCorePublicBatchSize: config.syncCorePublicBatchSize, syncResponderSnapshotLimits: config.syncResponderSnapshotLimits, syncContextGraphPriorities: config.syncContextGraphPriorities, diff --git a/packages/cli/test/daemon-sync-agents-meta-wiring.test.ts b/packages/cli/test/daemon-sync-agents-meta-wiring.test.ts index c6e0f74f8..ba1d99885 100644 --- a/packages/cli/test/daemon-sync-agents-meta-wiring.test.ts +++ b/packages/cli/test/daemon-sync-agents-meta-wiring.test.ts @@ -162,6 +162,23 @@ describe('runDaemonInner wires sync options into DKGAgent.create', () => { expect(createArg.syncGlobalQueueLimit).toBe(0); }); + it('passes the nested adaptive-capacity policy through unchanged', async () => { + const syncAdaptiveCapacity = { + enabled: true, + minInflight: 2, + maxInflight: 7, + } as const; + const createArg = await captureCreateArg({ syncAdaptiveCapacity }); + + expect(createArg.syncAdaptiveCapacity).toBe(syncAdaptiveCapacity); + }); + + it('keeps adaptive-capacity policy omitted when the operator omits it', async () => { + const createArg = await captureCreateArg(); + + expect(createArg).not.toHaveProperty('syncAdaptiveCapacity'); + }); + it('passes the Core public-CG peer-round batch size through unchanged', async () => { const createArg = await captureCreateArg({ syncCorePublicBatchSize: 13 }); From fb1d8830bf06b739dfbcb892c564303f804faad9 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:08:54 +0200 Subject: [PATCH 07/12] fix(sync): bind adaptive capacity to policy --- packages/agent/src/dkg-agent-base.ts | 22 ++++++++++++------- packages/agent/src/sync/capacity-runtime.ts | 15 +++++-------- .../core-public-coverage-scheduler.test.ts | 5 ++++- packages/agent/test/sync-backpressure.test.ts | 6 ++++- .../agent/test/sync-capacity-runtime.test.ts | 3 ++- 5 files changed, 30 insertions(+), 21 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index da964fa3e..2a0b411f5 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1723,14 +1723,20 @@ export class DKGAgentBase { /** Build one canonical peer-round scope with named snapshot/live phases. */ protected planCorePublicSyncPeerRound(remotePeer: string): PeerSyncScope { const selected = [...(this.config.syncContextGraphs ?? [])]; - const automaticContextGraphIds = (this.config.nodeRole ?? 'edge') === 'core' - ? this.corePublicSyncCoverageScheduler.planAutomaticCoverage( - selected, - this.config.syncContextGraphPriorities, - remotePeer, - this.syncCapacityRuntime.getEffectiveCoverageBatch(), - ) - : []; + let automaticContextGraphIds: string[] = []; + if ((this.config.nodeRole ?? 'edge') === 'core') { + automaticContextGraphIds = this.syncCapacityRuntime.isAdaptive() + ? this.corePublicSyncCoverageScheduler.planAutomaticCoverageWithOptions(selected, { + priorities: this.config.syncContextGraphPriorities, + planningLane: remotePeer, + effectiveBatchSize: this.syncCapacityRuntime.getEffectiveCoverageBatch(), + }) + : this.corePublicSyncCoverageScheduler.planAutomaticCoverage( + selected, + this.config.syncContextGraphPriorities, + remotePeer, + ); + } const initialDurableContextGraphIds = [...new Set([ ...selected, ...automaticContextGraphIds, diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 7df5ebf8c..9a13634c2 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -151,6 +151,7 @@ export class SyncCapacityRuntime { return new SyncCapacityRuntime(staticPolicy, configuredCoverageBatch); } + const controller = new AdaptiveCapacityController(bounds, { now: options.now }); const adaptivePolicy = resolveSyncGlobalBackpressure({ // The controller can never exceed hardMax, so use that same ceiling for // queue sizing and admission observability. Basing the queue on a larger @@ -158,11 +159,11 @@ export class SyncCapacityRuntime { // can never drain at the advertised policy capacity. syncGlobalMaxInflight: hardMax, syncGlobalQueueLimit: config.syncGlobalQueueLimit, - }); + }, () => controller.getCurrentInflight()); return new SyncCapacityRuntime( adaptivePolicy, configuredCoverageBatch, - new AdaptiveCapacityController(bounds, { now: options.now }), + controller, new AdaptiveCapacitySampler(store, options.samplerDependencies), ); } @@ -172,14 +173,8 @@ export class SyncCapacityRuntime { } /** Stable admission contract; callers do not need to branch on capacity mode. */ - getAdmissionOptions(): { - policy: SyncGlobalBackpressurePolicy; - currentLimit?: () => number; - } { - const controller = this.controller; - return controller - ? { policy: this.policy, currentLimit: () => controller.getCurrentInflight() } - : { policy: this.policy }; + getAdmissionOptions(): { policy: SyncGlobalBackpressurePolicy } { + return { policy: this.policy }; } getEffectiveCoverageBatch(): number { diff --git a/packages/agent/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index f32bf88e7..e1218f8b2 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -197,7 +197,10 @@ describe('Core public Context Graph coverage scheduler', () => { } expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); - scheduler.planAutomaticCoverage(['cg:a', 'cg:b'], undefined, 'peer-a', 2); + scheduler.planAutomaticCoverageWithOptions(['cg:a', 'cg:b'], { + planningLane: 'peer-a', + effectiveBatchSize: 2, + }); expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); expect(scheduler.hasAutomaticCoverageBacklog(3)).toBe(false); }); diff --git a/packages/agent/test/sync-backpressure.test.ts b/packages/agent/test/sync-backpressure.test.ts index b590c534e..b06ab57d5 100644 --- a/packages/agent/test/sync-backpressure.test.ts +++ b/packages/agent/test/sync-backpressure.test.ts @@ -396,10 +396,14 @@ describe('sync global backpressure', () => { // while every real admission reports `durable:unspecified` on // /api/diagnostics/backpressure, which is the attribution issue #2006 had to // reconstruct from daemon logs. + const config = { syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 1 }; const agentLike = { - config: { syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 1 }, + config, node: { stopSignal: undefined }, log: { info: () => {}, warn: () => {}, debug: () => {} }, + syncCapacityRuntime: { + getAdmissionOptions: () => ({ policy: resolveSyncGlobalBackpressure(config) }), + }, }; let releaseWork!: () => void; diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 4242493f1..5696f9829 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -60,7 +60,8 @@ describe('sync capacity runtime resolution', () => { currentCoverageBatch: 7, configuredCoverageBatch: 7, }); - expect(runtime.getAdmissionOptions().currentLimit?.()).toBe(2); + expect(runtime.getAdmissionOptions()).toEqual({ policy: runtime.policy }); + expect(runtime.policy.currentLimit?.()).toBe(2); }); it('owns sampling and restores constrained coverage from supplemental Core demand', () => { From 368a75bad14de6449a6838e75de0f5670ec92998 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:52:11 +0200 Subject: [PATCH 08/12] test(sync): initialize capacity runtime lifecycle fixture --- packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts index 14035787f..4a5e1d9c0 100644 --- a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts +++ b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts @@ -153,6 +153,7 @@ function minimalStartedAgent( }); Object.assign(agent, { started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, chainPoller: null, coreHostRecordingsClosed: false, drainCoreHostRecordings: vi.fn(async () => {}), @@ -504,6 +505,7 @@ describe('DKGAgent RFC-64 inventory lifecycle', () => { 'store', ]); expect(inventoryClose).toHaveBeenCalledOnce(); + expect(agent.syncCapacityRuntime.stopSampling).toHaveBeenCalledOnce(); expect(agent.finalizationRuntime.getRecoveryStore()).toBeUndefined(); expect(agent.rfc64PersistenceV1).toBeUndefined(); expect(agent.started).toBe(false); From b02fbbf65e774df4dfc53272111631c5c36e1e19 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 15:08:29 +0200 Subject: [PATCH 09/12] fix(sync): close adaptive capacity review gaps --- packages/agent/src/dkg-agent-base.ts | 2 + packages/agent/src/dkg-agent-lifecycle.ts | 1 + packages/agent/src/sync/backpressure.ts | 31 ++++++-- packages/agent/src/sync/capacity-runtime.ts | 10 ++- .../sync/core-public-coverage-scheduler.ts | 23 ++++-- .../core-public-coverage-scheduler.test.ts | 14 ++-- .../discovery-subscription-boundary.test.ts | 74 +++++++++++++++++-- .../agent/test/sync-capacity-runtime.test.ts | 42 +++++++++++ 8 files changed, 163 insertions(+), 34 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index 2a0b411f5..ab26c431c 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1725,6 +1725,8 @@ export class DKGAgentBase { const selected = [...(this.config.syncContextGraphs ?? [])]; let automaticContextGraphIds: string[] = []; if ((this.config.nodeRole ?? 'edge') === 'core') { + // Keep the established positional boundary exact for static callers; + // only adaptive activation crosses the named options seam. automaticContextGraphIds = this.syncCapacityRuntime.isAdaptive() ? this.corePublicSyncCoverageScheduler.planAutomaticCoverageWithOptions(selected, { priorities: this.config.syncContextGraphPriorities, diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index fc7cf076b..5816c0d06 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -3256,6 +3256,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { this.syncCapacityRuntime.startSampling({ hasSupplementalDemand: () => ( this.corePublicSyncCoverageScheduler.hasAutomaticCoverageBacklog( + this.config.syncContextGraphs ?? [], this.syncCapacityRuntime.getEffectiveCoverageBatch(), ) ), diff --git a/packages/agent/src/sync/backpressure.ts b/packages/agent/src/sync/backpressure.ts index 9186f8a53..2d102fa40 100644 --- a/packages/agent/src/sync/backpressure.ts +++ b/packages/agent/src/sync/backpressure.ts @@ -313,13 +313,16 @@ export function resolveExplicitSyncGlobalLimit( ?? nonNegativeInteger(config.syncGlobalLimit); } -/** Resolve one node-wide policy; the optional live limit is shared by every admission. */ -export function resolveSyncGlobalBackpressure( - config: SyncGlobalBackpressureConfig, +/** Build a policy from a final hard limit without re-reading global-limit env inputs. */ +export function createSyncGlobalBackpressurePolicy( + resolvedLimit: number, + configuredQueueLimit?: number, currentLimit?: SyncBackpressureCurrentLimit, ): SyncGlobalBackpressurePolicy { - const limit = resolveExplicitSyncGlobalLimit(config) - ?? DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT; + const limit = nonNegativeInteger(resolvedLimit); + if (limit === undefined) { + throw new TypeError('resolved sync global limit must be a non-negative integer'); + } if (limit === 0) { return Object.freeze({ limit: undefined, @@ -327,8 +330,10 @@ export function resolveSyncGlobalBackpressure( }) as SyncGlobalBackpressurePolicy; } + // Queue capacity remains an independent operator control. Only the already- + // resolved hard limit is protected from a second global-limit env lookup. const queueLimit = nonNegativeInteger(parseIntegerEnv('DKG_SYNC_GLOBAL_QUEUE_LIMIT')) - ?? nonNegativeInteger(config.syncGlobalQueueLimit) + ?? nonNegativeInteger(configuredQueueLimit) ?? limit * DEFAULT_SYNC_GLOBAL_QUEUE_LIMIT_MULTIPLIER; return Object.freeze({ limit, @@ -337,6 +342,20 @@ export function resolveSyncGlobalBackpressure( }) as SyncGlobalBackpressurePolicy; } +/** Resolve one node-wide policy; the optional live limit is shared by every admission. */ +export function resolveSyncGlobalBackpressure( + config: SyncGlobalBackpressureConfig, + currentLimit?: SyncBackpressureCurrentLimit, +): SyncGlobalBackpressurePolicy { + const limit = resolveExplicitSyncGlobalLimit(config) + ?? DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT; + return createSyncGlobalBackpressurePolicy( + limit, + config.syncGlobalQueueLimit, + currentLimit, + ); +} + export function getSyncBackpressureSnapshot( policy?: SyncGlobalBackpressurePolicy, now = Date.now(), diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index 9a13634c2..eff5edb4e 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -13,6 +13,7 @@ import { type AdaptiveCapacitySamplerDependencies, } from './adaptive-capacity-sampler.js'; import { + createSyncGlobalBackpressurePolicy, getSyncBackpressureSnapshot, notifyGlobalSyncBackpressureCapacityChanged, parseBooleanEnv, @@ -152,14 +153,15 @@ export class SyncCapacityRuntime { } const controller = new AdaptiveCapacityController(bounds, { now: options.now }); - const adaptivePolicy = resolveSyncGlobalBackpressure({ + const adaptivePolicy = createSyncGlobalBackpressurePolicy( // The controller can never exceed hardMax, so use that same ceiling for // queue sizing and admission observability. Basing the queue on a larger // explicit operator limit would allow a backlog that the adaptive Core // can never drain at the advertised policy capacity. - syncGlobalMaxInflight: hardMax, - syncGlobalQueueLimit: config.syncGlobalQueueLimit, - }, () => controller.getCurrentInflight()); + hardMax, + config.syncGlobalQueueLimit, + () => controller.getCurrentInflight(), + ); return new SyncCapacityRuntime( adaptivePolicy, configuredCoverageBatch, diff --git a/packages/agent/src/sync/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index a9035bf17..bd05ebeb0 100644 --- a/packages/agent/src/sync/core-public-coverage-scheduler.ts +++ b/packages/agent/src/sync/core-public-coverage-scheduler.ts @@ -90,7 +90,6 @@ export class CorePublicSyncCoverageScheduler { private readonly laneAnchors = new Map(); private lastPlanAt?: number; private lastPlan?: CorePublicSyncCoverageStatus['lastPlan']; - private lastCoverageCandidates?: number; constructor( private readonly batchSize: number, @@ -108,13 +107,11 @@ export class CorePublicSyncCoverageScheduler { if (!normalized) return false; const sizeBefore = this.tracked.size; this.tracked.add(normalized); - if (this.tracked.size !== sizeBefore) this.lastCoverageCandidates = undefined; return this.tracked.size !== sizeBefore; } unregister(contextGraphId: string): boolean { const removed = this.tracked.delete(contextGraphId); - if (removed) this.lastCoverageCandidates = undefined; if (this.tracked.size === 0) { this.laneAnchors.clear(); } else if (removed) { @@ -164,7 +161,6 @@ export class CorePublicSyncCoverageScheduler { const scheduledCoverage: string[] = []; const batchSize = resolveEffectiveBatchSize(this.batchSize, effectiveBatchSize); - this.lastCoverageCandidates = coverage.length; if (batchSize > 0 && coverage.length > 0) { const count = Math.min(batchSize, coverage.length); const previousAnchor = this.laneAnchors.get(planningLane); @@ -198,12 +194,23 @@ export class CorePublicSyncCoverageScheduler { return scheduledCoverage; } - /** True when automatic public coverage was truncated by the current batch. */ - hasAutomaticCoverageBacklog(effectiveBatchSize?: number): boolean { + /** Pure demand check for automatic public coverage beyond the current batch. */ + hasAutomaticCoverageBacklog( + selectedContextGraphIds: readonly string[], + effectiveBatchSize?: number, + ): boolean { if (this.batchSize === 0) return false; const batchSize = resolveEffectiveBatchSize(this.batchSize, effectiveBatchSize); - const candidates = this.lastCoverageCandidates ?? this.tracked.size; - return candidates > batchSize; + const selected = new Set( + selectedContextGraphIds.map((id) => id.trim()).filter(Boolean), + ); + let candidates = 0; + for (const contextGraphId of this.tracked) { + if (selected.has(contextGraphId)) continue; + candidates += 1; + if (candidates > batchSize) return true; + } + return false; } getStatus(enabled: boolean): CorePublicSyncCoverageStatus { diff --git a/packages/agent/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index e1218f8b2..341872e43 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -137,7 +137,7 @@ describe('Core public Context Graph coverage scheduler', () => { scheduler.register('public-a'); expect(scheduler.planAutomaticCoverage(['selected'])).toEqual([]); - expect(scheduler.hasAutomaticCoverageBacklog(0)).toBe(false); + expect(scheduler.hasAutomaticCoverageBacklog(['selected'], 0)).toBe(false); expect(scheduler.getStatus(true)).toMatchObject({ enabled: false, batchSize: 0, @@ -190,19 +190,15 @@ describe('Core public Context Graph coverage scheduler', () => { expect(third).toEqual(['cg:d']); }); - it('reports automatic coverage demand only while the live batch truncates candidates', () => { + it('computes automatic coverage demand from current selections without a prior plan', () => { const scheduler = new CorePublicSyncCoverageScheduler(4); for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d', 'cg:e']) { scheduler.register(contextGraphId); } - expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); - scheduler.planAutomaticCoverageWithOptions(['cg:a', 'cg:b'], { - planningLane: 'peer-a', - effectiveBatchSize: 2, - }); - expect(scheduler.hasAutomaticCoverageBacklog(2)).toBe(true); - expect(scheduler.hasAutomaticCoverageBacklog(3)).toBe(false); + expect(scheduler.hasAutomaticCoverageBacklog([], 2)).toBe(true); + expect(scheduler.hasAutomaticCoverageBacklog(['cg:a', 'cg:b'], 2)).toBe(true); + expect(scheduler.hasAutomaticCoverageBacklog(['cg:a', 'cg:b'], 3)).toBe(false); }); it('rejects invalid live automatic-coverage batches', () => { diff --git a/packages/agent/test/discovery-subscription-boundary.test.ts b/packages/agent/test/discovery-subscription-boundary.test.ts index 08327d858..b9ea1871f 100644 --- a/packages/agent/test/discovery-subscription-boundary.test.ts +++ b/packages/agent/test/discovery-subscription-boundary.test.ts @@ -19,6 +19,7 @@ import { type ContextGraphSubscriptionRecord, } from '../src/index.js'; import { normalizeLegacyContextGraphSubscriptionInput } from '../src/context-graph-subscription-policy.js'; +import { SyncCapacityRuntime } from '../src/sync/capacity-runtime.js'; describe('Context Graph discovery/subscription boundary', () => { it('normalizes the legacy public subscription shape before it enters live state', () => { @@ -402,15 +403,51 @@ describe('Context Graph discovery/subscription boundary', () => { } }, 30_000); - it('bounds store-discovered public coverage without capping explicit scope', async () => { - const publicIds = ['store-public-a', 'store-public-b', 'store-public-c']; + it('keeps static Core peer-round planning on the configured batch', async () => { const agent = await DKGAgent.create({ - name: 'BoundedCoreStoreDiscovery', + name: 'StaticCoreCoverageBoundary', listenHost: '127.0.0.1', nodeRole: 'core', chainAdapter: new MockChainAdapter(), syncContextGraphs: ['explicit-selection'], syncCorePublicBatchSize: 2, + syncGlobalMaxInflight: 3, + }); + + try { + await agent.start(); + expect(agent.getSyncCapacityStatus()).toMatchObject({ mode: 'static' }); + const scheduler = (agent as any).corePublicSyncCoverageScheduler; + for (const id of ['static-public-a', 'static-public-b', 'static-public-c']) { + scheduler.register(id); + } + + const scope = (agent as any).planCorePublicSyncPeerRound('static-peer'); + expect(scope.initialDurableContextGraphIds[0]).toBe('explicit-selection'); + expect(scope.automaticContextGraphIds).toHaveLength(2); + expect(scope.automaticContextGraphIds.every((id: string) => id.startsWith('static-public-'))) + .toBe(true); + } finally { + await agent.stop().catch(() => {}); + } + }, 30_000); + + it('applies the live adaptive coverage batch without capping explicit scope', async () => { + const publicIds = [ + 'store-public-a', + 'store-public-b', + 'store-public-c', + 'store-public-d', + 'store-public-e', + 'store-public-f', + ]; + const agent = await DKGAgent.create({ + name: 'BoundedCoreStoreDiscovery', + listenHost: '127.0.0.1', + nodeRole: 'core', + chainAdapter: new MockChainAdapter(), + syncContextGraphs: ['explicit-selection'], + syncCorePublicBatchSize: 8, }); try { @@ -436,15 +473,38 @@ describe('Context Graph discovery/subscription boundary', () => { }, ]))); - expect(await agent.discoverContextGraphsFromStore()).toBe(3); + expect(await agent.discoverContextGraphsFromStore()).toBe(6); expect((agent as any).config.syncContextGraphs).toEqual(['explicit-selection']); - expect(agent.getCorePublicSyncCoverageStatus().trackedContextGraphs).toBe(3); + expect(agent.getCorePublicSyncCoverageStatus().trackedContextGraphs).toBe(6); + + let cpuIdle = 0; + let cpuTotal = 0; + const constrainedRuntime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncCorePublicBatchSize: 8, + }, agent.store, { + parallelism: 16, + samplerDependencies: { + readCpuTimes: () => { + cpuIdle += 80; + cpuTotal += 100; + return { idle: cpuIdle, total: cpuTotal }; + }, + readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), + eventLoopUtilizationDelta: () => 0.2, + readHeapRatio: () => 0.82, + }, + }); + constrainedRuntime.sample(true); + expect(constrainedRuntime.getStatus().currentCoverageBatch).toBe(4); + (agent as any).syncCapacityRuntime.stopSampling(); + (agent as any).syncCapacityRuntime = constrainedRuntime; const scope = (agent as any).planCorePublicSyncPeerRound('store-peer'); const planned = scope.initialDurableContextGraphIds; expect(planned[0]).toBe('explicit-selection'); - expect(planned.slice(1)).toHaveLength(2); - expect(new Set(planned.slice(1)).size).toBe(2); + expect(planned.slice(1)).toHaveLength(4); + expect(new Set(planned.slice(1)).size).toBe(4); expect(planned.slice(1).every((id: string) => publicIds.includes(id))).toBe(true); } finally { await agent.stop().catch(() => {}); diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 5696f9829..3833bc256 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -101,6 +101,7 @@ describe('sync capacity runtime resolution', () => { heapRatio = 0.3; expect(runtime.startSampling({ hasSupplementalDemand: () => coverage.hasAutomaticCoverageBacklog( + [], runtime.getEffectiveCoverageBatch(), ), intervalMs: 5_000, @@ -214,6 +215,47 @@ describe('sync capacity runtime resolution', () => { }); }); + it('does not reapply a larger global-limit env after deriving the adaptive hard maximum', () => { + vi.stubEnv('DKG_SYNC_GLOBAL_MAX_INFLIGHT', '100'); + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncAdaptiveCapacity: { enabled: true }, + }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); + + expect(runtime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); + expect(runtime.getStatus()).toMatchObject({ + mode: 'adaptive', + currentInflight: 2, + maxInflight: 3, + }); + }); + + it('honors the config-level adaptive opt-out', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncAdaptiveCapacity: { enabled: false }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(runtime.isAdaptive()).toBe(false); + expect(runtime.getStatus()).toMatchObject({ mode: 'static', currentInflight: 2 }); + }); + + it('applies valid config-level adaptive min and max bounds', () => { + const runtime = SyncCapacityRuntime.create({ + nodeRole: 'core', + syncAdaptiveCapacity: { minInflight: 1, maxInflight: 2 }, + }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + + expect(runtime.isAdaptive()).toBe(true); + expect(runtime.policy).toMatchObject({ limit: 2, queueLimit: 4 }); + expect(runtime.getStatus()).toMatchObject({ + mode: 'adaptive', + currentInflight: 2, + minInflight: 1, + maxInflight: 2, + }); + }); + it('uses the legacy environment limit ahead of newer config for adaptive policy and status', () => { vi.stubEnv('DKG_SYNC_GLOBAL_LIMIT', '1'); const runtime = SyncCapacityRuntime.create({ From d8f1d6d30bbb6c969649ee2f03d2c7e5c4a98e89 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 15:15:57 +0200 Subject: [PATCH 10/12] test(sync): prove adaptive sampler lifecycle --- .../rfc64-agent-inventory-lifecycle.test.ts | 45 ++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts index 4a5e1d9c0..723501009 100644 --- a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts +++ b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts @@ -10,7 +10,10 @@ import { type SignedControlEnvelopeV1, type UnsignedControlEnvelopeV1, } from '@origintrail-official/dkg-core'; -import { verifyControlEnvelopeIssuerSignatureV1 } from '@origintrail-official/dkg-chain'; +import { + MockChainAdapter, + verifyControlEnvelopeIssuerSignatureV1, +} from '@origintrail-official/dkg-chain'; import { ethers } from 'ethers'; import { afterEach, describe, expect, it, vi } from 'vitest'; @@ -387,6 +390,46 @@ describe('DKGAgent RFC-64 inventory lifecycle', () => { expect(agent.started).toBe(false); }); + it('starts adaptive-capacity sampling through the real agent lifecycle', async () => { + const agent = await DKGAgent.create({ + name: 'AdaptiveCapacityLifecycle', + listenHost: '127.0.0.1', + listenPort: 0, + chainAdapter: new MockChainAdapter(), + nodeRole: 'core', + syncContextGraphs: ['explicit-cg'], + syncAdaptiveCapacity: { enabled: true }, + }); + const runtime = (agent as any).syncCapacityRuntime; + const scheduler = (agent as any).corePublicSyncCoverageScheduler; + const startSampling = vi.spyOn(runtime, 'startSampling').mockReturnValue(true); + const getEffectiveCoverageBatch = vi.spyOn( + runtime, + 'getEffectiveCoverageBatch', + ).mockReturnValue(4); + const hasAutomaticCoverageBacklog = vi.spyOn( + scheduler, + 'hasAutomaticCoverageBacklog', + ).mockReturnValue(true); + + try { + await agent.start(); + + expect(runtime.isAdaptive()).toBe(true); + expect(startSampling).toHaveBeenCalledOnce(); + const samplingOptions = startSampling.mock.calls[0]?.[0]; + expect(samplingOptions).toBeDefined(); + expect(samplingOptions.hasSupplementalDemand()).toBe(true); + expect(getEffectiveCoverageBatch).toHaveBeenCalledOnce(); + expect(hasAutomaticCoverageBacklog).toHaveBeenCalledWith( + ['explicit-cg'], + 4, + ); + } finally { + await agent.stop().catch(() => {}); + } + }, 30_000); + it.runIf(process.platform !== 'win32')( 'releases inventory ownership when control-store topology is unsafe', async () => { From df6f044d35e02684e06e615ddf5475444329b1b7 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 15:21:42 +0200 Subject: [PATCH 11/12] fix(sync): close capacity runtime boundary --- packages/agent/src/dkg-agent-base.ts | 13 ++--- packages/agent/src/dkg-agent-lifecycle.ts | 7 ++- packages/agent/src/sync/capacity-runtime.ts | 23 ++++++++- .../sync/core-public-coverage-scheduler.ts | 2 +- packages/agent/test/sync-backpressure.test.ts | 48 ++++++++++++++++++- .../agent/test/sync-capacity-runtime.test.ts | 39 ++++++++++----- 6 files changed, 102 insertions(+), 30 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index ab26c431c..ecfadccba 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1725,19 +1725,12 @@ export class DKGAgentBase { const selected = [...(this.config.syncContextGraphs ?? [])]; let automaticContextGraphIds: string[] = []; if ((this.config.nodeRole ?? 'edge') === 'core') { - // Keep the established positional boundary exact for static callers; - // only adaptive activation crosses the named options seam. - automaticContextGraphIds = this.syncCapacityRuntime.isAdaptive() - ? this.corePublicSyncCoverageScheduler.planAutomaticCoverageWithOptions(selected, { + automaticContextGraphIds = this.corePublicSyncCoverageScheduler + .planAutomaticCoverageWithOptions(selected, { priorities: this.config.syncContextGraphPriorities, planningLane: remotePeer, effectiveBatchSize: this.syncCapacityRuntime.getEffectiveCoverageBatch(), - }) - : this.corePublicSyncCoverageScheduler.planAutomaticCoverage( - selected, - this.config.syncContextGraphPriorities, - remotePeer, - ); + }); } const initialDurableContextGraphIds = [...new Set([ ...selected, diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 5816c0d06..dd67d334c 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -303,7 +303,6 @@ import { type DurableSyncAccumulator, } from './sync/durable-progress.js'; import { - getSyncBackpressureSnapshot, getSyncBackpressureBusyError, resolveBooleanSwitch, resolveNonNegativeIntegerSwitch, @@ -2021,7 +2020,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { ); }, onDecline: (details) => { - const syncPressure = getSyncBackpressureSnapshot(this.syncCapacityRuntime.policy); + const syncPressure = this.syncCapacityRuntime.getBackpressureSnapshot(); const syncPressureLabel = `syncGlobalInflight=${syncPressure.inflight} ` + `syncGlobalQueued=${syncPressure.queued} ` + @@ -2544,7 +2543,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { process.env, (message) => this.log.warn(ctx, message), ); - const syncGlobalPolicy = this.syncCapacityRuntime.policy; + const syncGlobalPolicy = this.syncCapacityRuntime.getResolvedPolicyStatus(); const syncCapacity = this.syncCapacityRuntime.getStatus(); const configuredPriorityCounts = countSyncPriorityClasses(this.config.syncContextGraphPriorities); this.log.info(ctx, `Resolved sync policy ${JSON.stringify({ @@ -2552,7 +2551,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { snapshotGlobalBytesEstimate: snapshotPolicy.budget.maxBytesEstimate, snapshotLocalRows: snapshotPolicy.budget.maxSnapshotRows, snapshotLocalBytesEstimate: snapshotPolicy.budget.maxSnapshotBytesEstimate, - syncGlobalInflightLimit: syncGlobalPolicy.limit ?? 0, + syncGlobalInflightLimit: syncGlobalPolicy.inflightLimit ?? 0, syncGlobalQueueLimit: syncGlobalPolicy.queueLimit ?? 0, syncCapacityMode: syncCapacity.mode, syncCapacityCurrentInflight: syncCapacity.currentInflight ?? 0, diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index eff5edb4e..c16fc5145 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -19,6 +19,7 @@ import { parseBooleanEnv, resolveExplicitSyncGlobalLimit, resolveSyncGlobalBackpressure, + type SyncBackpressureSnapshot, type SyncGlobalBackpressureConfig, type SyncGlobalBackpressurePolicy, } from './backpressure.js'; @@ -60,6 +61,11 @@ export interface SyncCapacitySamplingOptions { onError?: (error: unknown) => void; } +export interface SyncCapacityPolicyStatus { + inflightLimit: number | null; + queueLimit: number | null; +} + function readStorePressure(store: TripleStore) { try { return store.getPressureSnapshot?.(); @@ -96,7 +102,7 @@ export class SyncCapacityRuntime { private samplingTimer: ReturnType | undefined; private constructor( - readonly policy: SyncGlobalBackpressurePolicy, + private readonly policy: SyncGlobalBackpressurePolicy, private readonly configuredCoverageBatch: number, controller?: AdaptiveCapacityController, sampler?: AdaptiveCapacitySampler, @@ -179,6 +185,19 @@ export class SyncCapacityRuntime { return { policy: this.policy }; } + /** Runtime-owned requester-pressure view for lifecycle diagnostics and sampling. */ + getBackpressureSnapshot(): SyncBackpressureSnapshot { + return getSyncBackpressureSnapshot(this.policy); + } + + /** Stable resolved ceilings without exposing the branded admission policy. */ + getResolvedPolicyStatus(): SyncCapacityPolicyStatus { + return { + inflightLimit: this.policy.limit ?? null, + queueLimit: this.policy.queueLimit ?? null, + }; + } + getEffectiveCoverageBatch(): number { return this.controller?.getEffectiveCoverageBatch() ?? this.configuredCoverageBatch; } @@ -204,7 +223,7 @@ export class SyncCapacityRuntime { ); this.samplingTimer = setInterval(() => { try { - const pressure = getSyncBackpressureSnapshot(this.policy); + const pressure = this.getBackpressureSnapshot(); const demand = pressure.inflight > 0 || pressure.queued > 0 || (options.hasSupplementalDemand?.() ?? false); diff --git a/packages/agent/src/sync/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index bd05ebeb0..12cf1b106 100644 --- a/packages/agent/src/sync/core-public-coverage-scheduler.ts +++ b/packages/agent/src/sync/core-public-coverage-scheduler.ts @@ -135,7 +135,7 @@ export class CorePublicSyncCoverageScheduler { }); } - /** Named adaptive planning boundary; existing static callers keep their exact call shape. */ + /** Canonical named planning boundary; the positional wrapper preserves outside call shapes. */ planAutomaticCoverageWithOptions( selectedContextGraphIds: readonly string[], options: Readonly = {}, diff --git a/packages/agent/test/sync-backpressure.test.ts b/packages/agent/test/sync-backpressure.test.ts index b06ab57d5..c60676367 100644 --- a/packages/agent/test/sync-backpressure.test.ts +++ b/packages/agent/test/sync-backpressure.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { backpressureRegistry, createOperationContext, @@ -438,6 +438,52 @@ describe('sync global backpressure', () => { } }); + it('uses the runtime-owned live admission policy at the lifecycle boundary', async () => { + const config = { syncGlobalMaxInflight: 2, syncGlobalQueueLimit: 2 }; + const runtimePolicy = resolveSyncGlobalBackpressure(config, () => 1); + const getAdmissionOptions = vi.fn(() => ({ policy: runtimePolicy })); + const agentLike = { + config, + node: { stopSignal: undefined }, + log: { info: () => {}, warn: () => {}, debug: () => {} }, + syncCapacityRuntime: { getAdmissionOptions }, + }; + let releaseFirst!: () => void; + const first = LifecycleSyncMethods.prototype.runContextGraphSyncWithBackpressure.call( + agentLike as never, + createOperationContext('sync'), + 'runtime-policy-first', + 'durable' as never, + 'durable:runtime-policy-first', + () => new Promise((resolve) => { releaseFirst = resolve; }), + ); + await tick(); + + let secondRan = false; + const second = LifecycleSyncMethods.prototype.runContextGraphSyncWithBackpressure.call( + agentLike as never, + createOperationContext('sync'), + 'runtime-policy-second', + 'durable' as never, + 'durable:runtime-policy-second', + async () => { secondRan = true; }, + ); + await tick(); + + try { + expect(getAdmissionOptions).toHaveBeenCalledTimes(2); + expect(secondRan).toBe(false); + expect(getSyncBackpressureSnapshot(runtimePolicy)).toMatchObject({ + inflight: 1, + queued: 1, + limit: 1, + }); + } finally { + releaseFirst(); + await Promise.all([first, second]); + } + }); + it('removes CG and peer correlation identifiers from node-wide pressure diagnostics', async () => { const ctx = createOperationContext('sync'); const policy = resolveSyncGlobalBackpressure({ diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 3833bc256..65811bc96 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -39,8 +39,8 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); expect(runtime.isAdaptive()).toBe(false); - expect(runtime.policy.limit).toBe(4); - expect(runtime.getAdmissionOptions()).toEqual({ policy: runtime.policy }); + expect(runtime.getResolvedPolicyStatus()).toMatchObject({ inflightLimit: 4 }); + expect(runtime.getAdmissionOptions()).toHaveProperty('policy'); expect(runtime.getStatus()).toMatchObject({ mode: 'static', currentInflight: 4 }); }); @@ -51,7 +51,10 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16, now: () => 100 }); expect(runtime.isAdaptive()).toBe(true); - expect(runtime.policy.limit).toBe(3); + expect(runtime.getResolvedPolicyStatus()).toMatchObject({ + inflightLimit: 3, + queueLimit: 6, + }); expect(runtime.getStatus()).toMatchObject({ mode: 'adaptive', currentInflight: 2, @@ -60,8 +63,8 @@ describe('sync capacity runtime resolution', () => { currentCoverageBatch: 7, configuredCoverageBatch: 7, }); - expect(runtime.getAdmissionOptions()).toEqual({ policy: runtime.policy }); - expect(runtime.policy.currentLimit?.()).toBe(2); + expect(runtime.getAdmissionOptions()).toHaveProperty('policy'); + expect(runtime.getBackpressureSnapshot().limit).toBe(2); }); it('owns sampling and restores constrained coverage from supplemental Core demand', () => { @@ -194,9 +197,12 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); expect(staticRuntime.isAdaptive()).toBe(false); - expect(staticRuntime.policy.limit).toBe(6); + expect(staticRuntime.getResolvedPolicyStatus().inflightLimit).toBe(6); expect(adaptiveRuntime.isAdaptive()).toBe(true); - expect(adaptiveRuntime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); + expect(adaptiveRuntime.getResolvedPolicyStatus()).toEqual({ + inflightLimit: 3, + queueLimit: 6, + }); expect(adaptiveRuntime.getStatus().maxInflight).toBe(3); }); @@ -207,7 +213,10 @@ describe('sync capacity runtime resolution', () => { syncAdaptiveCapacity: { enabled: true }, }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); - expect(runtime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); + expect(runtime.getResolvedPolicyStatus()).toEqual({ + inflightLimit: 3, + queueLimit: 6, + }); expect(runtime.getStatus()).toMatchObject({ mode: 'adaptive', currentInflight: 2, @@ -222,7 +231,10 @@ describe('sync capacity runtime resolution', () => { syncAdaptiveCapacity: { enabled: true }, }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); - expect(runtime.policy).toMatchObject({ limit: 3, queueLimit: 6 }); + expect(runtime.getResolvedPolicyStatus()).toEqual({ + inflightLimit: 3, + queueLimit: 6, + }); expect(runtime.getStatus()).toMatchObject({ mode: 'adaptive', currentInflight: 2, @@ -247,7 +259,10 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); expect(runtime.isAdaptive()).toBe(true); - expect(runtime.policy).toMatchObject({ limit: 2, queueLimit: 4 }); + expect(runtime.getResolvedPolicyStatus()).toEqual({ + inflightLimit: 2, + queueLimit: 4, + }); expect(runtime.getStatus()).toMatchObject({ mode: 'adaptive', currentInflight: 2, @@ -265,7 +280,7 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); expect(runtime.isAdaptive()).toBe(true); - expect(runtime.policy.limit).toBe(1); + expect(runtime.getResolvedPolicyStatus().inflightLimit).toBe(1); expect(runtime.getStatus()).toMatchObject({ mode: 'adaptive', currentInflight: 1, @@ -310,7 +325,7 @@ describe('sync capacity runtime resolution', () => { }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); expect(runtime.isAdaptive()).toBe(false); - expect(runtime.policy.limit).toBeUndefined(); + expect(runtime.getResolvedPolicyStatus().inflightLimit).toBeNull(); expect(runtime.getStatus().currentInflight).toBeNull(); }); From 6992dd36ec23bf298e039ad5fd7a9b517b7a855d Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 15:58:11 +0200 Subject: [PATCH 12/12] fix(sync): resolve coverage batch once --- packages/agent/src/dkg-agent-base.ts | 9 +- packages/agent/src/sync/capacity-runtime.ts | 10 +- .../discovery-subscription-boundary.test.ts | 2 +- .../agent/test/sync-capacity-runtime.test.ts | 117 +++++++++++++----- 4 files changed, 96 insertions(+), 42 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index ecfadccba..d8bc5cf28 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1630,13 +1630,18 @@ export class DKGAgentBase { publicSnapshotStore?: WorkspacePublicSnapshotStore, ) { this.config = config; + const resolvedCoverageBatch = resolveCorePublicSyncBatchSize( + config.syncCorePublicBatchSize, + ); this.corePublicSyncCoverageScheduler = new CorePublicSyncCoverageScheduler( - resolveCorePublicSyncBatchSize(config.syncCorePublicBatchSize), + resolvedCoverageBatch, ); this.wallet = wallet; this.node = node; this.store = store; - this.syncCapacityRuntime = SyncCapacityRuntime.create(config, store); + this.syncCapacityRuntime = SyncCapacityRuntime.create(config, store, { + resolvedCoverageBatch, + }); this.contextGraphMetaProjection = new ContextGraphMetaProjection(store); this.publisher = publisher; this.queryEngine = queryEngine; diff --git a/packages/agent/src/sync/capacity-runtime.ts b/packages/agent/src/sync/capacity-runtime.ts index c16fc5145..399a68c1d 100644 --- a/packages/agent/src/sync/capacity-runtime.ts +++ b/packages/agent/src/sync/capacity-runtime.ts @@ -23,14 +23,12 @@ import { type SyncGlobalBackpressureConfig, type SyncGlobalBackpressurePolicy, } from './backpressure.js'; -import { resolveCorePublicSyncBatchSize } from './core-public-coverage-scheduler.js'; export const DEFAULT_SYNC_CAPACITY_SAMPLE_INTERVAL_MS = 5_000; export interface SyncCapacityRuntimeConfig extends SyncGlobalBackpressureConfig { nodeRole?: 'core' | 'edge'; syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig; - syncCorePublicBatchSize?: number; } export interface SyncCapacityStatus { @@ -49,6 +47,8 @@ export interface SyncCapacityStatus { } export interface SyncCapacityRuntimeOptions { + /** Agent-boundary value shared with the Core coverage scheduler. */ + resolvedCoverageBatch: number; parallelism?: number; samplerDependencies?: AdaptiveCapacitySamplerDependencies; now?: () => number; @@ -114,11 +114,9 @@ export class SyncCapacityRuntime { static create( config: SyncCapacityRuntimeConfig, store: TripleStore, - options: SyncCapacityRuntimeOptions = {}, + options: SyncCapacityRuntimeOptions, ): SyncCapacityRuntime { - const configuredCoverageBatch = resolveCorePublicSyncBatchSize( - config.syncCorePublicBatchSize, - ); + const configuredCoverageBatch = options.resolvedCoverageBatch; const staticPolicy = resolveSyncGlobalBackpressure(config); const explicitGlobalLimit = resolveExplicitSyncGlobalLimit(config); const explicitlyEnabled = parseBooleanEnv('DKG_SYNC_ADAPTIVE_CAPACITY_ENABLED') diff --git a/packages/agent/test/discovery-subscription-boundary.test.ts b/packages/agent/test/discovery-subscription-boundary.test.ts index b9ea1871f..491c4a686 100644 --- a/packages/agent/test/discovery-subscription-boundary.test.ts +++ b/packages/agent/test/discovery-subscription-boundary.test.ts @@ -481,8 +481,8 @@ describe('Context Graph discovery/subscription boundary', () => { let cpuTotal = 0; const constrainedRuntime = SyncCapacityRuntime.create({ nodeRole: 'core', - syncCorePublicBatchSize: 8, }, agent.store, { + resolvedCoverageBatch: 8, parallelism: 16, samplerDependencies: { readCpuTimes: () => { diff --git a/packages/agent/test/sync-capacity-runtime.test.ts b/packages/agent/test/sync-capacity-runtime.test.ts index 65811bc96..1dbd06f8b 100644 --- a/packages/agent/test/sync-capacity-runtime.test.ts +++ b/packages/agent/test/sync-capacity-runtime.test.ts @@ -1,8 +1,15 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { createOperationContext } from '@origintrail-official/dkg-core'; import type { StorePressureSnapshot, TripleStore } from '@origintrail-official/dkg-storage'; -import { SyncCapacityRuntime } from '../src/sync/capacity-runtime.js'; -import { CorePublicSyncCoverageScheduler } from '../src/sync/core-public-coverage-scheduler.js'; +import { + SyncCapacityRuntime, + type SyncCapacityRuntimeOptions, +} from '../src/sync/capacity-runtime.js'; +import { + CorePublicSyncCoverageScheduler, + DEFAULT_CORE_PUBLIC_SYNC_BATCH_SIZE, + resolveCorePublicSyncBatchSize, +} from '../src/sync/core-public-coverage-scheduler.js'; import { withGlobalSyncBackpressure } from '../src/sync/backpressure.js'; function fakeStore(pressure?: StorePressureSnapshot): TripleStore { @@ -11,6 +18,15 @@ function fakeStore(pressure?: StorePressureSnapshot): TripleStore { } as unknown as TripleStore; } +function capacityOptions( + overrides: Partial = {}, +): SyncCapacityRuntimeOptions { + return { + resolvedCoverageBatch: DEFAULT_CORE_PUBLIC_SYNC_BATCH_SIZE, + ...overrides, + }; +} + const STORE_PRESSURE: StorePressureSnapshot = { ackInflight: 0, healthInflight: 0, @@ -36,7 +52,7 @@ describe('sync capacity runtime resolution', () => { nodeRole: 'edge', syncGlobalMaxInflight: 4, syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(runtime.isAdaptive()).toBe(false); expect(runtime.getResolvedPolicyStatus()).toMatchObject({ inflightLimit: 4 }); @@ -47,8 +63,11 @@ describe('sync capacity runtime resolution', () => { it('defaults Core nodes without an explicit global limit to adaptive mode', () => { const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', - syncCorePublicBatchSize: 7, - }, fakeStore(STORE_PRESSURE), { parallelism: 16, now: () => 100 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ + resolvedCoverageBatch: 7, + parallelism: 16, + now: () => 100, + })); expect(runtime.isAdaptive()).toBe(true); expect(runtime.getResolvedPolicyStatus()).toMatchObject({ @@ -67,6 +86,31 @@ describe('sync capacity runtime resolution', () => { expect(runtime.getBackpressureSnapshot().limit).toBe(2); }); + it('shares one boundary-resolved coverage batch across scheduler and runtime', () => { + const config = { nodeRole: 'core' as const, syncCorePublicBatchSize: 7 }; + vi.stubEnv('DKG_SYNC_CORE_PUBLIC_BATCH_SIZE', '3'); + const resolvedCoverageBatch = resolveCorePublicSyncBatchSize( + config.syncCorePublicBatchSize, + ); + const scheduler = new CorePublicSyncCoverageScheduler(resolvedCoverageBatch); + + // A later environment change would expose any second owner that re-resolved + // the original config instead of consuming the agent-boundary value. + vi.stubEnv('DKG_SYNC_CORE_PUBLIC_BATCH_SIZE', '5'); + const runtime = SyncCapacityRuntime.create( + config, + fakeStore(STORE_PRESSURE), + capacityOptions({ resolvedCoverageBatch }), + ); + + expect(resolveCorePublicSyncBatchSize(config.syncCorePublicBatchSize)).toBe(5); + expect(scheduler.getStatus(true).batchSize).toBe(3); + expect(runtime.getStatus()).toMatchObject({ + configuredCoverageBatch: 3, + currentCoverageBatch: 3, + }); + }); + it('owns sampling and restores constrained coverage from supplemental Core demand', () => { vi.useFakeTimers(); vi.setSystemTime(0); @@ -75,8 +119,7 @@ describe('sync capacity runtime resolution', () => { let cpuTotal = 0; const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', - syncCorePublicBatchSize: 8, - }, fakeStore(STORE_PRESSURE), { + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16, samplerDependencies: { readCpuTimes: () => { @@ -88,7 +131,7 @@ describe('sync capacity runtime resolution', () => { eventLoopUtilizationDelta: () => 0.2, readHeapRatio: () => heapRatio, }, - }); + })); const coverage = new CorePublicSyncCoverageScheduler(8); for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d', 'cg:e']) { coverage.register(contextGraphId); @@ -126,20 +169,24 @@ describe('sync capacity runtime resolution', () => { let heapRatio = 0.82; let cpuIdle = 0; let cpuTotal = 0; - const runtime = SyncCapacityRuntime.create({ nodeRole: 'core' }, fakeStore(STORE_PRESSURE), { - parallelism: 16, - now: () => now, - samplerDependencies: { - readCpuTimes: () => { - cpuIdle += 80; - cpuTotal += 100; - return { idle: cpuIdle, total: cpuTotal }; + const runtime = SyncCapacityRuntime.create( + { nodeRole: 'core' }, + fakeStore(STORE_PRESSURE), + capacityOptions({ + parallelism: 16, + now: () => now, + samplerDependencies: { + readCpuTimes: () => { + cpuIdle += 80; + cpuTotal += 100; + return { idle: cpuIdle, total: cpuTotal }; + }, + readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), + eventLoopUtilizationDelta: () => 0.2, + readHeapRatio: () => heapRatio, }, - readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), - eventLoopUtilizationDelta: () => 0.2, - readHeapRatio: () => heapRatio, - }, - }); + }), + ); runtime.sample(true); expect(runtime.getStatus().currentInflight).toBe(1); @@ -189,12 +236,12 @@ describe('sync capacity runtime resolution', () => { const staticRuntime = SyncCapacityRuntime.create({ nodeRole: 'core', syncGlobalMaxInflight: 6, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); const adaptiveRuntime = SyncCapacityRuntime.create({ nodeRole: 'core', syncGlobalMaxInflight: 6, syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(staticRuntime.isAdaptive()).toBe(false); expect(staticRuntime.getResolvedPolicyStatus().inflightLimit).toBe(6); @@ -211,7 +258,7 @@ describe('sync capacity runtime resolution', () => { nodeRole: 'core', syncGlobalMaxInflight: 100, syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 64 })); expect(runtime.getResolvedPolicyStatus()).toEqual({ inflightLimit: 3, @@ -229,7 +276,7 @@ describe('sync capacity runtime resolution', () => { const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 64 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 64 })); expect(runtime.getResolvedPolicyStatus()).toEqual({ inflightLimit: 3, @@ -246,7 +293,7 @@ describe('sync capacity runtime resolution', () => { const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', syncAdaptiveCapacity: { enabled: false }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(runtime.isAdaptive()).toBe(false); expect(runtime.getStatus()).toMatchObject({ mode: 'static', currentInflight: 2 }); @@ -256,7 +303,7 @@ describe('sync capacity runtime resolution', () => { const runtime = SyncCapacityRuntime.create({ nodeRole: 'core', syncAdaptiveCapacity: { minInflight: 1, maxInflight: 2 }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(runtime.isAdaptive()).toBe(true); expect(runtime.getResolvedPolicyStatus()).toEqual({ @@ -277,7 +324,7 @@ describe('sync capacity runtime resolution', () => { nodeRole: 'core', syncGlobalMaxInflight: 6, syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(runtime.isAdaptive()).toBe(true); expect(runtime.getResolvedPolicyStatus().inflightLimit).toBe(1); @@ -293,7 +340,7 @@ describe('sync capacity runtime resolution', () => { expect(() => SyncCapacityRuntime.create({ nodeRole: 'core', syncAdaptiveCapacity: { maxInflight }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 }))).toThrow( 'syncAdaptiveCapacity.maxInflight must be a positive integer', ); }); @@ -303,7 +350,7 @@ describe('sync capacity runtime resolution', () => { expect(() => SyncCapacityRuntime.create({ nodeRole: 'core', syncAdaptiveCapacity: { maxInflight: 1 }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 }))).toThrow( 'DKG_SYNC_ADAPTIVE_MAX_INFLIGHT must be a positive integer', ); }); @@ -312,7 +359,7 @@ describe('sync capacity runtime resolution', () => { vi.stubEnv('DKG_SYNC_ADAPTIVE_MIN_INFLIGHT', '1.5'); expect(() => SyncCapacityRuntime.create({ nodeRole: 'core', - }, fakeStore(STORE_PRESSURE), { parallelism: 16 })).toThrow( + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 }))).toThrow( 'DKG_SYNC_ADAPTIVE_MIN_INFLIGHT must be a positive integer', ); }); @@ -322,7 +369,7 @@ describe('sync capacity runtime resolution', () => { nodeRole: 'core', syncGlobalMaxInflight: 0, syncAdaptiveCapacity: { enabled: true }, - }, fakeStore(STORE_PRESSURE), { parallelism: 16 }); + }, fakeStore(STORE_PRESSURE), capacityOptions({ parallelism: 16 })); expect(runtime.isAdaptive()).toBe(false); expect(runtime.getResolvedPolicyStatus().inflightLimit).toBeNull(); @@ -331,7 +378,11 @@ describe('sync capacity runtime resolution', () => { it('honors the adaptive environment disable over inferred Core defaults', () => { vi.stubEnv('DKG_SYNC_ADAPTIVE_CAPACITY_ENABLED', '0'); - const runtime = SyncCapacityRuntime.create({ nodeRole: 'core' }, fakeStore(STORE_PRESSURE)); + const runtime = SyncCapacityRuntime.create( + { nodeRole: 'core' }, + fakeStore(STORE_PRESSURE), + capacityOptions(), + ); expect(runtime.isAdaptive()).toBe(false); }); });