From 078be0795bb9237c245656dd69cb794b1ec97f2a Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 13:57:01 +0200 Subject: [PATCH 1/9] feat(sync): add policy-owned adaptive admission --- packages/agent/src/sync/backpressure.ts | 163 ++++++++++++++--- .../src/sync/priority-admission-queue.ts | 6 + packages/agent/test/sync-backpressure.test.ts | 170 ++++++++++++++++++ 3 files changed, 312 insertions(+), 27 deletions(-) diff --git a/packages/agent/src/sync/backpressure.ts b/packages/agent/src/sync/backpressure.ts index 7b4eb8d7aa..9186f8a53f 100644 --- a/packages/agent/src/sync/backpressure.ts +++ b/packages/agent/src/sync/backpressure.ts @@ -28,17 +28,23 @@ export interface SyncGlobalBackpressureConfig { declare const syncGlobalBackpressurePolicyBrand: unique symbol; export type SyncGlobalBackpressurePolicy = Readonly<( - | { limit: number; queueLimit: number } + | { + limit: number; + queueLimit: number; + currentLimit?: SyncBackpressureCurrentLimit; + } | { limit: undefined; queueLimit: undefined } ) & { [syncGlobalBackpressurePolicyBrand]: true }>; interface GlobalQueuePayload { - limit: number; label: string; contextGraphId?: string; source: SyncAdmissionSource; } +/** Resolve the current requester-sync capacity. The static policy remains the hard ceiling. */ +export type SyncBackpressureCurrentLimit = () => number; + export const DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT = 2; export const DEFAULT_SYNC_GLOBAL_QUEUE_LIMIT_MULTIPLIER = 2; export const DEFAULT_SYNC_PRIORITY_AGING_MS = 30_000; @@ -70,13 +76,91 @@ function syncAdmissionOperation(payload: GlobalQueuePayload): string { } let inflight = 0; -let lastLimit: number | null = null; -let lastQueueLimit: number | null = null; +const globalCapacity: { + hardLimit: number | null; + queueLimit: number | null; + currentLimit?: SyncBackpressureCurrentLimit; + effectiveLimit: number | null; +} = { + hardLimit: null, + queueLimit: null, + effectiveLimit: null, +}; + +function clampEffectiveLimit(hardLimit: number, currentLimit: number): number { + return Math.min(hardLimit, currentLimit); +} + +function refreshGlobalCapacity(): number | null { + const { hardLimit, currentLimit } = globalCapacity; + if (hardLimit === null) return null; + if (!currentLimit) { + globalCapacity.effectiveLimit = hardLimit; + return hardLimit; + } + try { + const current = positiveInteger(currentLimit()); + if (current !== undefined) { + globalCapacity.effectiveLimit = clampEffectiveLimit(hardLimit, current); + } + } catch { + // Runtime capacity sampling is advisory. Retain the last valid shared value. + } + return globalCapacity.effectiveLimit ?? hardLimit; +} + +function activateGlobalCapacity(policy: SyncGlobalBackpressurePolicy): number | null { + if (policy.limit === undefined) { + globalCapacity.hardLimit = null; + globalCapacity.queueLimit = null; + globalCapacity.currentLimit = undefined; + globalCapacity.effectiveLimit = null; + return null; + } + + const providerChanged = globalCapacity.currentLimit !== policy.currentLimit; + const hardLimitChanged = globalCapacity.hardLimit !== policy.limit; + globalCapacity.hardLimit = policy.limit; + globalCapacity.queueLimit = policy.queueLimit; + globalCapacity.currentLimit = policy.currentLimit; + if (!policy.currentLimit) { + globalCapacity.effectiveLimit = policy.limit; + } else if (providerChanged || hardLimitChanged) { + // A new provider starts from the safest valid value already in force. If + // its first sample fails, never increase concurrency because of the error. + globalCapacity.effectiveLimit = Math.min( + policy.limit, + globalCapacity.effectiveLimit ?? policy.limit, + ); + } + return refreshGlobalCapacity(); +} + +function snapshotPolicyLimit(policy: SyncGlobalBackpressurePolicy): number | null { + if (policy.limit === undefined) return null; + if ( + globalCapacity.hardLimit === policy.limit + && globalCapacity.queueLimit === policy.queueLimit + && globalCapacity.currentLimit === policy.currentLimit + ) { + return refreshGlobalCapacity(); + } + if (!policy.currentLimit) return policy.limit; + try { + const current = positiveInteger(policy.currentLimit()); + return current === undefined + ? policy.limit + : clampEffectiveLimit(policy.limit, current); + } catch { + return policy.limit; + } +} + const queue = new PriorityAdmissionQueue({ - canRun: (entry) => inflight < entry.payload.limit, - onStart: (entry) => { + canRun: () => inflight < (refreshGlobalCapacity() ?? 0), + onStart: () => { inflight += 1; - lastLimit = entry.payload.limit; + refreshGlobalCapacity(); getMetrics().syncGlobalInflight.record(inflight); return () => { inflight = Math.max(0, inflight - 1); @@ -90,7 +174,7 @@ const queue = new PriorityAdmissionQueue({ // them to a fixed operation class, paired with the bounded admission // source, before node-wide diagnostics/logging. operation: (entry) => syncAdmissionOperation(entry.payload), - inflightLimit: (entry) => entry.payload.limit, + inflightLimit: () => refreshGlobalCapacity(), thresholds: { degradedQueueAgeMs: DEFAULT_SYNC_PRIORITY_AGING_MS / 2, stalledActiveAgeMs: 120_000, @@ -142,16 +226,15 @@ function acquire( const { limit } = policy; if (limit === undefined) throw new Error('disabled sync backpressure policy cannot acquire'); const { queueLimit } = policy; - lastLimit = limit; - lastQueueLimit = queueLimit; + const payload: GlobalQueuePayload = { + label: options.label, + contextGraphId: options.contextGraphId, + source: options.source, + }; + const currentLimit = activateGlobalCapacity(policy) ?? limit; const queuedBefore = queue.length; return queue.acquire({ - payload: { - limit, - label: options.label, - contextGraphId: options.contextGraphId, - source: options.source, - }, + payload, ownerKey: 'global', lane: options.lane, priority: options.priority, @@ -162,7 +245,7 @@ function acquire( queueLimit, createBusyError: () => new SyncBackpressureBusyError( `Sync backpressure rejected ${options.label} ` - + `(global inflight=${inflight}/${limit}, queued=${queuedBefore}/${queueLimit})`, + + `(global inflight=${inflight}/${currentLimit}, queued=${queuedBefore}/${queueLimit})`, ), createDisplacedError: (victim) => new SyncBackpressureBusyError( `Sync backpressure displaced ${victim.payload.contextGraphId ?? 'queued work'} for higher-priority ${options.label}`, @@ -220,13 +303,22 @@ export function resolveNonNegativeIntegerSwitch( return nonNegativeInteger(value); } -export function resolveSyncGlobalBackpressure( +/** Resolve only an explicitly configured global limit, using runtime policy precedence. */ +export function resolveExplicitSyncGlobalLimit( config: SyncGlobalBackpressureConfig, -): SyncGlobalBackpressurePolicy { - const limit = nonNegativeInteger(parseIntegerEnv('DKG_SYNC_GLOBAL_MAX_INFLIGHT')) +): number | undefined { + return nonNegativeInteger(parseIntegerEnv('DKG_SYNC_GLOBAL_MAX_INFLIGHT')) ?? nonNegativeInteger(parseIntegerEnv('DKG_SYNC_GLOBAL_LIMIT')) ?? nonNegativeInteger(config.syncGlobalMaxInflight) - ?? nonNegativeInteger(config.syncGlobalLimit) + ?? nonNegativeInteger(config.syncGlobalLimit); +} + +/** 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; if (limit === 0) { return Object.freeze({ @@ -241,6 +333,7 @@ export function resolveSyncGlobalBackpressure( return Object.freeze({ limit, queueLimit, + ...(currentLimit ? { currentLimit } : {}), }) as SyncGlobalBackpressurePolicy; } @@ -257,13 +350,28 @@ export function getSyncBackpressureSnapshot( return { inflight, queued: queue.length, - limit: policy ? policy.limit ?? null : lastLimit, - queueLimit: policy ? policy.queueLimit ?? null : lastQueueLimit, + limit: policy ? snapshotPolicyLimit(policy) : globalCapacity.effectiveLimit, + queueLimit: policy ? policy.queueLimit ?? null : globalCapacity.queueLimit, queuedByPriorityClass, oldestQueuedAgeMs: queue.oldestAgeMs(now), }; } +/** + * Re-evaluate dynamic requester capacity and start newly admissible queued work. + * Downshifts are drain-only because the queue never revokes running admissions. + */ +export function notifyGlobalSyncBackpressureCapacityChanged(): void { + const effectiveLimit = refreshGlobalCapacity(); + if (effectiveLimit !== null && globalCapacity.queueLimit !== null) { + queue.refreshPressureCapacity({ + inflightLimit: effectiveLimit, + queueLimit: globalCapacity.queueLimit, + }); + } + queue.pump(); +} + export async function withGlobalSyncBackpressure( options: { policy: SyncGlobalBackpressurePolicy; @@ -291,8 +399,7 @@ export async function withGlobalSyncBackpressure( ): Promise { const { limit, queueLimit } = options.policy; if (limit === undefined) { - lastLimit = null; - lastQueueLimit = null; + if (inflight === 0 && queue.length === 0) activateGlobalCapacity(options.policy); if (options.signal?.aborted) { throw options.signal.reason instanceof Error ? options.signal.reason @@ -324,18 +431,20 @@ export async function withGlobalSyncBackpressure( } if (admission.status === 'queued') { + const currentLimit = refreshGlobalCapacity() ?? limit; options.logInfo?.( options.ctx, `Sync backpressure queued ${options.label} ` - + `(global inflight=${inflight}/${limit}, queued=${admission.queuedBefore}/${queueLimit})`, + + `(global inflight=${inflight}/${currentLimit}, queued=${admission.queuedBefore}/${queueLimit})`, ); } const release = await admission.release; try { + const currentLimit = refreshGlobalCapacity() ?? limit; options.logInfo?.( options.ctx, `Sync backpressure running ${options.label} ` - + `(global inflight=${inflight}/${limit}, queued=${queue.length}/${queueLimit})`, + + `(global inflight=${inflight}/${currentLimit}, queued=${queue.length}/${queueLimit})`, ); return await work(); } finally { diff --git a/packages/agent/src/sync/priority-admission-queue.ts b/packages/agent/src/sync/priority-admission-queue.ts index 74d802f35a..754119cbc8 100644 --- a/packages/agent/src/sync/priority-admission-queue.ts +++ b/packages/agent/src/sync/priority-admission-queue.ts @@ -2,6 +2,7 @@ import { backpressureRegistry, getMetrics, ObservableScheduler, + type SchedulerPressureCapacity, type SchedulerPressureThresholds, type SchedulerPressureTicket, } from '@origintrail-official/dkg-core'; @@ -141,6 +142,11 @@ export class PriorityAdmissionQueue extends ObservableScheduler { return Math.max(0, now - Math.min(...this.queue.map((entry) => entry.enqueuedAt))); } + /** Refresh reported capacity after a caller-owned dynamic limit changes. */ + refreshPressureCapacity(capacity: SchedulerPressureCapacity): void { + this.updatePressureCapacity(capacity); + } + acquire(options: PriorityAdmissionAcquireOptions): PriorityAdmission { return this.acquireInternal(options); } diff --git a/packages/agent/test/sync-backpressure.test.ts b/packages/agent/test/sync-backpressure.test.ts index 382fe6b2a0..b590c534ec 100644 --- a/packages/agent/test/sync-backpressure.test.ts +++ b/packages/agent/test/sync-backpressure.test.ts @@ -5,6 +5,7 @@ import { } from '@origintrail-official/dkg-core'; import { getSyncBackpressureSnapshot, + notifyGlobalSyncBackpressureCapacityChanged, resolveBooleanSwitch, resolveNonNegativeIntegerSwitch, resolveSyncGlobalBackpressure, @@ -217,6 +218,175 @@ describe('sync global backpressure', () => { ]); }); + it('preserves the static admission API and configured concurrency when no resolver is supplied', async () => { + const ctx = createOperationContext('sync'); + const policy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: 2, + syncGlobalQueueLimit: 1, + }); + const events: string[] = []; + let releaseFirst!: () => void; + let releaseSecond!: () => void; + + const first = withGlobalSyncBackpressure({ policy, ctx, label: 'static-first' }, async () => { + events.push('first-start'); + await new Promise((resolve) => { releaseFirst = resolve; }); + }); + const second = withGlobalSyncBackpressure({ policy, ctx, label: 'static-second' }, async () => { + events.push('second-start'); + await new Promise((resolve) => { releaseSecond = resolve; }); + }); + await tick(); + + expect(events).toEqual(['first-start', 'second-start']); + expect(getSyncBackpressureSnapshot(policy)).toMatchObject({ + inflight: 2, + queued: 0, + limit: 2, + queueLimit: 1, + }); + + releaseFirst(); + releaseSecond(); + await Promise.all([first, second]); + }); + + it('lets a lower effective limit drain without cancelling already-running work', async () => { + const ctx = createOperationContext('sync'); + let currentLimit = 2; + const policy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: 2, + syncGlobalQueueLimit: 1, + }, () => currentLimit); + const events: string[] = []; + let releaseFirst!: () => void; + let releaseSecond!: () => void; + const options = (label: string) => ({ policy, ctx, label }); + + const first = withGlobalSyncBackpressure(options('dynamic-first'), async () => { + events.push('first-start'); + await new Promise((resolve) => { releaseFirst = resolve; }); + events.push('first-end'); + }); + const second = withGlobalSyncBackpressure(options('dynamic-second'), async () => { + events.push('second-start'); + await new Promise((resolve) => { releaseSecond = resolve; }); + events.push('second-end'); + }); + await tick(); + expect(events).toEqual(['first-start', 'second-start']); + + currentLimit = 1; + notifyGlobalSyncBackpressureCapacityChanged(); + const third = withGlobalSyncBackpressure(options('dynamic-third'), async () => { + events.push('third-start'); + }); + await tick(); + expect(events).toEqual(['first-start', 'second-start']); + + releaseFirst(); + await first; + await tick(); + expect(events).toEqual(['first-start', 'second-start', 'first-end']); + + releaseSecond(); + await Promise.all([second, third]); + expect(events).toEqual([ + 'first-start', + 'second-start', + 'first-end', + 'second-end', + 'third-start', + ]); + }); + + it('pumps already-queued work immediately when the effective limit increases', async () => { + const ctx = createOperationContext('sync'); + let currentLimit = 1; + const policy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: 2, + syncGlobalQueueLimit: 1, + }, () => currentLimit); + const events: string[] = []; + let releaseFirst!: () => void; + let releaseSecond!: () => void; + const options = (label: string) => ({ policy, ctx, label }); + + const first = withGlobalSyncBackpressure(options('dynamic-first'), async () => { + events.push('first-start'); + await new Promise((resolve) => { releaseFirst = resolve; }); + }); + await tick(); + const second = withGlobalSyncBackpressure(options('dynamic-second'), async () => { + events.push('second-start'); + await new Promise((resolve) => { releaseSecond = resolve; }); + }); + await tick(); + expect(events).toEqual(['first-start']); + expect(getSyncBackpressureSnapshot()).toMatchObject({ inflight: 1, queued: 1, limit: 1 }); + + currentLimit = 2; + notifyGlobalSyncBackpressureCapacityChanged(); + await tick(); + expect(events).toEqual(['first-start', 'second-start']); + expect(getSyncBackpressureSnapshot()).toMatchObject({ inflight: 2, queued: 0, limit: 2 }); + + releaseFirst(); + releaseSecond(); + await Promise.all([first, second]); + }); + + it('retains the last valid shared capacity when the policy resolver fails', async () => { + const ctx = createOperationContext('sync'); + let currentLimit = 1; + let resolverFails = false; + const policy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: 2, + syncGlobalQueueLimit: 1, + }, () => { + if (resolverFails) throw new Error('capacity sample unavailable'); + return currentLimit; + }); + const events: string[] = []; + let releaseFirst!: () => void; + let releaseSecond!: () => void; + + const first = withGlobalSyncBackpressure({ policy, ctx, label: 'dynamic-first' }, async () => { + events.push('first-start'); + await new Promise((resolve) => { releaseFirst = resolve; }); + }); + await tick(); + const second = withGlobalSyncBackpressure({ policy, ctx, label: 'dynamic-second' }, async () => { + events.push('second-start'); + await new Promise((resolve) => { releaseSecond = resolve; }); + }); + await tick(); + expect(events).toEqual(['first-start']); + + currentLimit = 2; + resolverFails = true; + notifyGlobalSyncBackpressureCapacityChanged(); + await tick(); + expect(events).toEqual(['first-start']); + expect(getSyncBackpressureSnapshot()).toMatchObject({ inflight: 1, queued: 1, limit: 1 }); + + resolverFails = false; + currentLimit = 0; + notifyGlobalSyncBackpressureCapacityChanged(); + await tick(); + expect(events).toEqual(['first-start']); + expect(getSyncBackpressureSnapshot()).toMatchObject({ inflight: 1, queued: 1, limit: 1 }); + + currentLimit = 2; + notifyGlobalSyncBackpressureCapacityChanged(); + await tick(); + expect(events).toEqual(['first-start', 'second-start']); + + releaseFirst(); + releaseSecond(); + await Promise.all([first, second]); + }); + it('carries the admission source from the production helper through to the scheduler', async () => { // The two halves of this contract were covered separately: the call sites // were proven to SUPPLY a source, and `withGlobalSyncBackpressure` was proven From 1c0e33f13653a6dc0d9d3fdf680715aa8f1cdef7 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 13:57:38 +0200 Subject: [PATCH 2/9] feat(sync): adapt Core coverage breadth --- .../sync/core-public-coverage-scheduler.ts | 17 ++++++- .../core-public-coverage-scheduler.test.ts | 51 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/packages/agent/src/sync/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index 99a0dba780..6f6cfc6de9 100644 --- a/packages/agent/src/sync/core-public-coverage-scheduler.ts +++ b/packages/agent/src/sync/core-public-coverage-scheduler.ts @@ -27,6 +27,17 @@ function normalizeBatchSize(value: number | undefined): number { return value; } +function resolveEffectiveBatchSize( + configuredBatchSize: number, + effectiveBatchSize: number | undefined, +): number { + if (effectiveBatchSize === undefined) return configuredBatchSize; + if (!Number.isInteger(effectiveBatchSize) || effectiveBatchSize < 0) { + throw new TypeError('effective Core public sync batch size must be a non-negative integer'); + } + return Math.min(configuredBatchSize, effectiveBatchSize); +} + function greatestCommonDivisor(a: number, b: number): number { let left = a; let right = b; @@ -111,6 +122,7 @@ export class CorePublicSyncCoverageScheduler { selectedContextGraphIds: readonly string[], priorities?: Readonly, planningLane = 'default', + effectiveBatchSize?: number, ): string[] { const selected = [...new Set( selectedContextGraphIds.map((id) => id.trim()).filter(Boolean), @@ -127,8 +139,9 @@ export class CorePublicSyncCoverageScheduler { .map(({ contextGraphId }) => contextGraphId); const scheduledCoverage: string[] = []; - if (this.batchSize > 0 && coverage.length > 0) { - const count = Math.min(this.batchSize, coverage.length); + const batchSize = resolveEffectiveBatchSize(this.batchSize, effectiveBatchSize); + if (batchSize > 0 && coverage.length > 0) { + const count = Math.min(batchSize, coverage.length); const previousAnchor = this.laneAnchors.get(planningLane); const previousAnchorIndex = previousAnchor === undefined ? -1 diff --git a/packages/agent/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index dee5a37615..07e0b0fd9c 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -145,6 +145,57 @@ describe('Core public Context Graph coverage scheduler', () => { }); }); + it('clamps a live automatic-coverage batch without counting selected CGs', () => { + const scheduler = new CorePublicSyncCoverageScheduler(4); + for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d']) { + scheduler.register(contextGraphId); + } + + const constrained = scheduler.planAutomaticCoverage( + ['cg:selected'], + undefined, + 'peer-a', + 1, + ); + expect(constrained).toHaveLength(1); + + const recovered = scheduler.planAutomaticCoverage( + ['cg:selected'], + undefined, + 'peer-a', + 20, + ); + expect(recovered).toHaveLength(4); + expect(recovered).not.toContain('cg:selected'); + }); + + it('preserves first-slot rotation while the effective batch changes', () => { + const scheduler = new CorePublicSyncCoverageScheduler(3); + for (const contextGraphId of ['cg:a', 'cg:b', 'cg:c', 'cg:d', 'cg:e']) { + scheduler.register(contextGraphId); + } + + const first = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1); + const second = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 2); + const third = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1); + + expect(first).toEqual(['cg:a']); + expect(second).toEqual(['cg:c', 'cg:d']); + expect(third).toEqual(['cg:d']); + }); + + it('rejects invalid live automatic-coverage batches', () => { + const scheduler = new CorePublicSyncCoverageScheduler(3); + scheduler.register('cg:a'); + + expect(() => scheduler.planAutomaticCoverage([], undefined, 'peer-a', -1)).toThrow( + /effective Core public sync batch size/, + ); + expect(() => scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1.5)).toThrow( + /effective Core public sync batch size/, + ); + }); + it('resolves the env override and rejects unsafe batch sizes', () => { expect(resolveCorePublicSyncBatchSize(undefined, undefined)) .toBe(DEFAULT_CORE_PUBLIC_SYNC_BATCH_SIZE); From 1a1f00d09f964253bbf35636b3ef1b06a3629b1b Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 03:33:51 +0200 Subject: [PATCH 3/9] feat(sync): add adaptive capacity controller --- packages/agent/src/sync/adaptive-capacity.ts | 502 ++++++++++++++++++ packages/agent/test/adaptive-capacity.test.ts | 335 ++++++++++++ packages/agent/vitest.unit.config.ts | 1 + 3 files changed, 838 insertions(+) create mode 100644 packages/agent/src/sync/adaptive-capacity.ts create mode 100644 packages/agent/test/adaptive-capacity.test.ts diff --git a/packages/agent/src/sync/adaptive-capacity.ts b/packages/agent/src/sync/adaptive-capacity.ts new file mode 100644 index 0000000000..8f71753340 --- /dev/null +++ b/packages/agent/src/sync/adaptive-capacity.ts @@ -0,0 +1,502 @@ +export const DEFAULT_SYNC_ADAPTIVE_INITIAL_INFLIGHT = 2; +export const DEFAULT_SYNC_ADAPTIVE_MIN_INFLIGHT = 1; +export const MAX_SYNC_ADAPTIVE_INFLIGHT = 8; +export const DEFAULT_SYNC_ADAPTIVE_COVERAGE_BATCH = 8; +export const DEFAULT_SYNC_ADAPTIVE_COOLDOWN_MS = 30_000; +export const DEFAULT_SYNC_ADAPTIVE_HEALTHY_SAMPLES = 6; +export const DEFAULT_SYNC_ADAPTIVE_STRAINED_SAMPLES = 2; + +export const SYNC_ADAPTIVE_CAPACITY_THRESHOLDS = Object.freeze({ + criticalHeapRatio: 0.82, + criticalEventLoopUtilization: 0.92, + strainedCpuUtilization: 0.85, + strainedHeapRatio: 0.72, + strainedEventLoopUtilization: 0.8, + healthyCpuUtilization: 0.65, + healthyHeapRatio: 0.6, + healthyEventLoopUtilization: 0.65, +}); + +export type AdaptiveCapacityState = + | 'warming' + | 'healthy' + | 'constrained' + | 'cooldown'; + +export type AdaptiveCapacityAction = + | 'hold' + | 'increase' + | 'decrease' + | 'halve'; + +export type AdaptiveCapacityReason = + | 'warming' + | 'ambiguous_signals' + | 'no_demand' + | 'critical_ack_queue' + | 'critical_health_queue' + | 'critical_store_saturated' + | 'critical_store_stalled' + | 'critical_heap' + | 'critical_event_loop' + | 'strained_store_queue' + | 'strained_cpu' + | 'strained_heap' + | 'strained_event_loop' + | 'strained_hysteresis' + | 'healthy_hysteresis' + | 'cooldown' + | 'at_maximum' + | 'store_telemetry_growth_cap'; + +export interface AdaptiveCapacityBoundsInput { + /** Existing requester cap. Zero retains its established unbounded meaning. */ + initialInflight?: number; + minInflight?: number; + maxInflight?: number; + /** Zero retains its established meaning: automatic Core coverage is disabled. */ + configuredCoverageBatch?: number; +} + +export type AdaptiveCapacityBounds = Readonly< + | { + mode: 'unbounded'; + configuredCoverageBatch: number; + } + | { + mode: 'bounded'; + initialInflight: number; + minInflight: number; + maxInflight: number; + configuredCoverageBatch: number; + } +>; + +function requireInteger( + name: string, + value: number, + minimum: number, +): number { + if (!Number.isInteger(value) || value < minimum) { + throw new TypeError(`${name} must be an integer greater than or equal to ${minimum}`); + } + return value; +} + +/** + * Resolve only controller-local bounds. Role/config/env precedence and the + * hardware/store-derived maximum remain integration concerns. + */ +export function resolveAdaptiveCapacityBounds( + input: AdaptiveCapacityBoundsInput, +): AdaptiveCapacityBounds { + const configuredCoverageBatch = requireInteger( + 'configuredCoverageBatch', + input.configuredCoverageBatch ?? DEFAULT_SYNC_ADAPTIVE_COVERAGE_BATCH, + 0, + ); + const requestedInitial = requireInteger( + 'initialInflight', + input.initialInflight ?? DEFAULT_SYNC_ADAPTIVE_INITIAL_INFLIGHT, + 0, + ); + + if (requestedInitial === 0) { + return Object.freeze({ + mode: 'unbounded', + configuredCoverageBatch, + }); + } + + const minInflight = requireInteger( + 'minInflight', + input.minInflight ?? DEFAULT_SYNC_ADAPTIVE_MIN_INFLIGHT, + 1, + ); + if (minInflight > MAX_SYNC_ADAPTIVE_INFLIGHT) { + throw new RangeError( + `minInflight must not exceed the absolute adaptive cap ${MAX_SYNC_ADAPTIVE_INFLIGHT}`, + ); + } + const requestedMax = requireInteger( + 'maxInflight', + input.maxInflight ?? MAX_SYNC_ADAPTIVE_INFLIGHT, + 1, + ); + const maxInflight = Math.min(requestedMax, MAX_SYNC_ADAPTIVE_INFLIGHT); + if (maxInflight < minInflight) { + throw new RangeError('maxInflight must be greater than or equal to minInflight'); + } + + return Object.freeze({ + mode: 'bounded', + initialInflight: Math.max(minInflight, Math.min(requestedInitial, maxInflight)), + minInflight, + maxInflight, + configuredCoverageBatch, + }); +} + +export type AdaptiveCapacityStoreSample = Readonly< + | { + telemetryAvailable: false; + } + | { + telemetryAvailable: true; + ackQueued: number; + healthQueued: number; + normalQueued: number; + backgroundQueued: number; + saturated?: boolean; + stalled?: boolean; + } +>; + +export interface AdaptiveCapacitySample { + /** Real requester or automatic-coverage backlog; it is not a pressure signal. */ + demand: boolean; + /** Ratios in the inclusive range 0..1. Missing host signals are ambiguous. */ + cpuUtilization?: number; + heapRatio?: number; + eventLoopUtilization?: number; + store: AdaptiveCapacityStoreSample; +} + +export interface AdaptiveCapacityDecision { + readonly action: AdaptiveCapacityAction; + readonly reason: AdaptiveCapacityReason; + readonly atMs: number; + readonly previousInflight: number; + readonly currentInflight: number; + readonly previousCoverageBatch: number; + readonly currentCoverageBatch: number; +} + +export interface AdaptiveCapacityStatus { + readonly state: AdaptiveCapacityState; + readonly currentInflight: number; + readonly minInflight: number; + readonly maxInflight: number; + readonly effectiveCoverageBatch: number; + readonly configuredCoverageBatch: number; + readonly storePressureTelemetryAvailable: boolean; + readonly consecutiveStrainedSamples: number; + readonly consecutiveHealthyDemandSamples: number; + readonly cooldownUntilMs: number; + readonly lastDecision: Readonly; +} + +export interface AdaptiveCapacityControllerOptions { + now?: () => number; + cooldownMs?: number; + healthySamplesToGrow?: number; + strainedSamplesToShrink?: number; +} + +interface ClassifiedSample { + kind: 'critical' | 'strained' | 'healthy' | 'ambiguous'; + reason: AdaptiveCapacityReason; +} + +function validateRatio(name: string, value: number | undefined): void { + if (value === undefined) return; + if (!Number.isFinite(value) || value < 0 || value > 1) { + throw new RangeError(`${name} must be a finite ratio between 0 and 1`); + } +} + +function validateQueueDepth(name: string, value: number): void { + requireInteger(name, value, 0); +} + +function classifySample(sample: AdaptiveCapacitySample): ClassifiedSample { + const thresholds = SYNC_ADAPTIVE_CAPACITY_THRESHOLDS; + const store = sample.store; + if (store.telemetryAvailable) { + if (store.ackQueued > 0) return { kind: 'critical', reason: 'critical_ack_queue' }; + if (store.healthQueued > 0) return { kind: 'critical', reason: 'critical_health_queue' }; + if (store.saturated === true) { + return { kind: 'critical', reason: 'critical_store_saturated' }; + } + if (store.stalled === true) return { kind: 'critical', reason: 'critical_store_stalled' }; + } + if ( + sample.heapRatio !== undefined + && sample.heapRatio >= thresholds.criticalHeapRatio + ) { + return { kind: 'critical', reason: 'critical_heap' }; + } + if ( + sample.eventLoopUtilization !== undefined + && sample.eventLoopUtilization >= thresholds.criticalEventLoopUtilization + ) { + return { kind: 'critical', reason: 'critical_event_loop' }; + } + + if ( + store.telemetryAvailable + && (store.normalQueued > 0 || store.backgroundQueued > 0) + ) { + return { kind: 'strained', reason: 'strained_store_queue' }; + } + if ( + sample.cpuUtilization !== undefined + && sample.cpuUtilization >= thresholds.strainedCpuUtilization + ) { + return { kind: 'strained', reason: 'strained_cpu' }; + } + if ( + sample.heapRatio !== undefined + && sample.heapRatio >= thresholds.strainedHeapRatio + ) { + return { kind: 'strained', reason: 'strained_heap' }; + } + if ( + sample.eventLoopUtilization !== undefined + && sample.eventLoopUtilization >= thresholds.strainedEventLoopUtilization + ) { + return { kind: 'strained', reason: 'strained_event_loop' }; + } + + const completeHostSample = sample.cpuUtilization !== undefined + && sample.heapRatio !== undefined + && sample.eventLoopUtilization !== undefined; + const storeQueuesHealthy = !store.telemetryAvailable + || ( + store.ackQueued === 0 + && store.healthQueued === 0 + && store.normalQueued === 0 + && store.backgroundQueued === 0 + && store.saturated !== true + && store.stalled !== true + ); + if ( + completeHostSample + && sample.cpuUtilization! <= thresholds.healthyCpuUtilization + && sample.heapRatio! <= thresholds.healthyHeapRatio + && sample.eventLoopUtilization! <= thresholds.healthyEventLoopUtilization + && storeQueuesHealthy + ) { + return { kind: 'healthy', reason: 'healthy_hysteresis' }; + } + return { kind: 'ambiguous', reason: 'ambiguous_signals' }; +} + +/** + * Pure fast-down/slow-up AIMD controller. Sampling and application of the + * returned limits are deliberately owned by the DKGAgent integration layer. + */ +export class AdaptiveCapacityController { + private readonly now: () => number; + private readonly cooldownMs: number; + private readonly healthySamplesToGrow: number; + private readonly strainedSamplesToShrink: number; + private currentInflight: number; + private currentCoverageBatch: number; + private state: AdaptiveCapacityState = 'warming'; + private storePressureTelemetryAvailable = false; + private consecutiveStrainedSamples = 0; + private consecutiveHealthyDemandSamples = 0; + private cooldownUntilMs: number; + private lastDecision: Readonly; + + constructor( + private readonly bounds: Extract, + options: AdaptiveCapacityControllerOptions = {}, + ) { + this.now = options.now ?? Date.now; + this.cooldownMs = requireInteger( + 'cooldownMs', + options.cooldownMs ?? DEFAULT_SYNC_ADAPTIVE_COOLDOWN_MS, + 0, + ); + this.healthySamplesToGrow = requireInteger( + 'healthySamplesToGrow', + options.healthySamplesToGrow ?? DEFAULT_SYNC_ADAPTIVE_HEALTHY_SAMPLES, + 1, + ); + this.strainedSamplesToShrink = requireInteger( + 'strainedSamplesToShrink', + options.strainedSamplesToShrink ?? DEFAULT_SYNC_ADAPTIVE_STRAINED_SAMPLES, + 1, + ); + this.currentInflight = bounds.initialInflight; + this.currentCoverageBatch = bounds.configuredCoverageBatch; + const createdAt = this.now(); + if (!Number.isFinite(createdAt)) throw new RangeError('now must return a finite timestamp'); + this.cooldownUntilMs = createdAt + this.cooldownMs; + this.lastDecision = Object.freeze({ + action: 'hold', + reason: 'warming', + atMs: createdAt, + previousInflight: this.currentInflight, + currentInflight: this.currentInflight, + previousCoverageBatch: this.currentCoverageBatch, + currentCoverageBatch: this.currentCoverageBatch, + }); + } + + observe(sample: AdaptiveCapacitySample, atMs = this.now()): AdaptiveCapacityStatus { + this.validateSample(sample, atMs); + this.storePressureTelemetryAvailable = sample.store.telemetryAvailable; + const classified = classifySample(sample); + + if (classified.kind === 'critical') { + this.consecutiveStrainedSamples = 0; + this.consecutiveHealthyDemandSamples = 0; + this.applyDecision('halve', classified.reason, atMs, () => { + this.currentInflight = Math.max( + this.bounds.minInflight, + Math.floor(this.currentInflight / 2), + ); + if (this.currentCoverageBatch > 0) { + this.currentCoverageBatch = Math.max( + 1, + Math.floor(this.currentCoverageBatch / 2), + ); + } + }); + this.state = 'constrained'; + this.cooldownUntilMs = atMs + this.cooldownMs; + return this.getStatus(); + } + + if (classified.kind === 'strained') { + this.consecutiveHealthyDemandSamples = 0; + this.consecutiveStrainedSamples += 1; + if (this.consecutiveStrainedSamples < this.strainedSamplesToShrink) { + this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; + this.applyDecision('hold', 'strained_hysteresis', atMs); + return this.getStatus(); + } + + this.consecutiveStrainedSamples = 0; + this.applyDecision('decrease', classified.reason, atMs, () => { + this.currentInflight = Math.max( + this.bounds.minInflight, + this.currentInflight - 1, + ); + if (this.currentCoverageBatch > 0) { + this.currentCoverageBatch = Math.max(1, this.currentCoverageBatch - 1); + } + }); + this.state = 'constrained'; + this.cooldownUntilMs = atMs + this.cooldownMs; + return this.getStatus(); + } + + this.consecutiveStrainedSamples = 0; + if (classified.kind === 'ambiguous') { + this.consecutiveHealthyDemandSamples = 0; + this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; + this.applyDecision('hold', classified.reason, atMs); + return this.getStatus(); + } + + if (!sample.demand) { + this.consecutiveHealthyDemandSamples = 0; + this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'healthy'; + this.applyDecision('hold', 'no_demand', atMs); + return this.getStatus(); + } + + this.consecutiveHealthyDemandSamples = Math.min( + this.healthySamplesToGrow, + this.consecutiveHealthyDemandSamples + 1, + ); + if (this.consecutiveHealthyDemandSamples < this.healthySamplesToGrow) { + this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; + this.applyDecision('hold', 'healthy_hysteresis', atMs); + return this.getStatus(); + } + if (atMs < this.cooldownUntilMs) { + this.state = 'cooldown'; + this.applyDecision('hold', 'cooldown', atMs); + return this.getStatus(); + } + + const growthInflightCeiling = sample.store.telemetryAvailable + ? this.bounds.maxInflight + : Math.min(this.bounds.maxInflight, DEFAULT_SYNC_ADAPTIVE_INITIAL_INFLIGHT); + const inflightCanGrow = this.currentInflight < growthInflightCeiling; + const coverageCanGrow = this.currentCoverageBatch > 0 + && this.currentCoverageBatch < this.bounds.configuredCoverageBatch; + if (!inflightCanGrow && !coverageCanGrow) { + this.state = sample.store.telemetryAvailable ? 'healthy' : 'constrained'; + this.applyDecision( + 'hold', + !sample.store.telemetryAvailable && this.currentInflight < this.bounds.maxInflight + ? 'store_telemetry_growth_cap' + : 'at_maximum', + atMs, + ); + return this.getStatus(); + } + + this.applyDecision('increase', 'healthy_hysteresis', atMs, () => { + if (inflightCanGrow) this.currentInflight += 1; + if (coverageCanGrow) this.currentCoverageBatch += 1; + }); + this.consecutiveHealthyDemandSamples = 0; + this.cooldownUntilMs = atMs + this.cooldownMs; + this.state = 'cooldown'; + return this.getStatus(); + } + + getCurrentInflight(): number { + return this.currentInflight; + } + + getEffectiveCoverageBatch(): number { + return this.currentCoverageBatch; + } + + getStatus(): AdaptiveCapacityStatus { + return Object.freeze({ + state: this.state, + currentInflight: this.currentInflight, + minInflight: this.bounds.minInflight, + maxInflight: this.bounds.maxInflight, + effectiveCoverageBatch: this.currentCoverageBatch, + configuredCoverageBatch: this.bounds.configuredCoverageBatch, + storePressureTelemetryAvailable: this.storePressureTelemetryAvailable, + consecutiveStrainedSamples: this.consecutiveStrainedSamples, + consecutiveHealthyDemandSamples: this.consecutiveHealthyDemandSamples, + cooldownUntilMs: this.cooldownUntilMs, + lastDecision: this.lastDecision, + }); + } + + private validateSample(sample: AdaptiveCapacitySample, atMs: number): void { + if (!Number.isFinite(atMs)) throw new RangeError('atMs must be a finite timestamp'); + validateRatio('cpuUtilization', sample.cpuUtilization); + validateRatio('heapRatio', sample.heapRatio); + validateRatio('eventLoopUtilization', sample.eventLoopUtilization); + if (sample.store.telemetryAvailable) { + validateQueueDepth('store.ackQueued', sample.store.ackQueued); + validateQueueDepth('store.healthQueued', sample.store.healthQueued); + validateQueueDepth('store.normalQueued', sample.store.normalQueued); + validateQueueDepth('store.backgroundQueued', sample.store.backgroundQueued); + } + } + + private applyDecision( + action: AdaptiveCapacityAction, + reason: AdaptiveCapacityReason, + atMs: number, + mutate?: () => void, + ): void { + const previousInflight = this.currentInflight; + const previousCoverageBatch = this.currentCoverageBatch; + mutate?.(); + this.lastDecision = Object.freeze({ + action, + reason, + atMs, + previousInflight, + currentInflight: this.currentInflight, + previousCoverageBatch, + currentCoverageBatch: this.currentCoverageBatch, + }); + } +} diff --git a/packages/agent/test/adaptive-capacity.test.ts b/packages/agent/test/adaptive-capacity.test.ts new file mode 100644 index 0000000000..76212607d3 --- /dev/null +++ b/packages/agent/test/adaptive-capacity.test.ts @@ -0,0 +1,335 @@ +import { describe, expect, it } from 'vitest'; +import { + AdaptiveCapacityController, + MAX_SYNC_ADAPTIVE_INFLIGHT, + resolveAdaptiveCapacityBounds, + type AdaptiveCapacitySample, +} from '../src/sync/adaptive-capacity.js'; + +const healthyStore = { + telemetryAvailable: true, + ackQueued: 0, + healthQueued: 0, + normalQueued: 0, + backgroundQueued: 0, +} as const; + +function healthySample( + overrides: Partial = {}, +): AdaptiveCapacitySample { + return { + demand: true, + cpuUtilization: 0.3, + heapRatio: 0.4, + eventLoopUtilization: 0.35, + store: healthyStore, + ...overrides, + }; +} + +function controller( + input: Parameters[0] = {}, + now = 0, +): AdaptiveCapacityController { + const bounds = resolveAdaptiveCapacityBounds(input); + if (bounds.mode !== 'bounded') throw new Error('test requires bounded capacity'); + return new AdaptiveCapacityController(bounds, { now: () => now }); +} + +describe('adaptive sync capacity bounds', () => { + it('resolves safe defaults and clamps the initial and hard maximum', () => { + expect(resolveAdaptiveCapacityBounds({})).toEqual({ + mode: 'bounded', + initialInflight: 2, + minInflight: 1, + maxInflight: MAX_SYNC_ADAPTIVE_INFLIGHT, + configuredCoverageBatch: 8, + }); + expect(resolveAdaptiveCapacityBounds({ + initialInflight: 12, + maxInflight: 20, + configuredCoverageBatch: 3, + })).toEqual({ + mode: 'bounded', + initialInflight: 8, + minInflight: 1, + maxInflight: 8, + configuredCoverageBatch: 3, + }); + expect(resolveAdaptiveCapacityBounds({ + initialInflight: 1, + minInflight: 3, + maxInflight: 5, + })).toMatchObject({ initialInflight: 3, minInflight: 3, maxInflight: 5 }); + }); + + it('preserves zero as unbounded inflight and disabled automatic coverage', () => { + expect(resolveAdaptiveCapacityBounds({ + initialInflight: 0, + configuredCoverageBatch: 0, + })).toEqual({ + mode: 'unbounded', + configuredCoverageBatch: 0, + }); + }); + + it('fails clearly on invalid or contradictory bounded input', () => { + expect(() => resolveAdaptiveCapacityBounds({ initialInflight: -1 })) + .toThrow(/initialInflight/); + expect(() => resolveAdaptiveCapacityBounds({ initialInflight: 1.5 })) + .toThrow(/initialInflight/); + expect(() => resolveAdaptiveCapacityBounds({ minInflight: 0 })) + .toThrow(/minInflight/); + expect(() => resolveAdaptiveCapacityBounds({ minInflight: 4, maxInflight: 3 })) + .toThrow(/maxInflight/); + expect(() => resolveAdaptiveCapacityBounds({ minInflight: 9, maxInflight: 10 })) + .toThrow(/absolute adaptive cap/); + expect(() => resolveAdaptiveCapacityBounds({ configuredCoverageBatch: -1 })) + .toThrow(/configuredCoverageBatch/); + expect(() => resolveAdaptiveCapacityBounds({ configuredCoverageBatch: 1.5 })) + .toThrow(/configuredCoverageBatch/); + }); +}); + +describe('adaptive sync capacity controller', () => { + it('exposes a cheap immutable status before the first sample', () => { + const capacity = controller({ initialInflight: 2, maxInflight: 4 }); + + const first = capacity.getStatus(); + const second = capacity.getStatus(); + + expect(first).not.toBe(second); + expect(capacity.getCurrentInflight()).toBe(2); + expect(capacity.getEffectiveCoverageBatch()).toBe(8); + expect(first).toMatchObject({ + state: 'warming', + currentInflight: 2, + minInflight: 1, + maxInflight: 4, + effectiveCoverageBatch: 8, + storePressureTelemetryAvailable: false, + cooldownUntilMs: 30_000, + lastDecision: { action: 'hold', reason: 'warming', atMs: 0 }, + }); + expect(Object.isFrozen(first)).toBe(true); + expect(Object.isFrozen(first.lastDecision)).toBe(true); + }); + + it('increases by exactly one after six healthy demand samples and the cooldown', () => { + const capacity = controller({ initialInflight: 2, maxInflight: 5 }); + + for (let sample = 1; sample <= 5; sample += 1) { + expect(capacity.observe(healthySample(), sample * 5_000).currentInflight).toBe(2); + } + const grown = capacity.observe(healthySample(), 30_000); + + expect(grown).toMatchObject({ + state: 'cooldown', + currentInflight: 3, + consecutiveHealthyDemandSamples: 0, + cooldownUntilMs: 60_000, + lastDecision: { + action: 'increase', + previousInflight: 2, + currentInflight: 3, + }, + }); + }); + + it('does not treat an idle sync queue as growth demand', () => { + const capacity = controller({ initialInflight: 2, maxInflight: 5 }); + + for (let sample = 1; sample <= 12; sample += 1) { + capacity.observe(healthySample({ demand: false }), sample * 5_000); + } + + expect(capacity.getStatus()).toMatchObject({ + state: 'healthy', + currentInflight: 2, + consecutiveHealthyDemandSamples: 0, + lastDecision: { action: 'hold', reason: 'no_demand' }, + }); + }); + + it.each([ + ['ACK work', healthySample({ store: { ...healthyStore, ackQueued: 1 } }), 'critical_ack_queue'], + ['health work', healthySample({ store: { ...healthyStore, healthQueued: 1 } }), 'critical_health_queue'], + ['store saturation', healthySample({ store: { ...healthyStore, saturated: true } }), 'critical_store_saturated'], + ['store stall', healthySample({ store: { ...healthyStore, stalled: true } }), 'critical_store_stalled'], + ['heap pressure', healthySample({ heapRatio: 0.82 }), 'critical_heap'], + ['event-loop pressure', healthySample({ eventLoopUtilization: 0.92 }), 'critical_event_loop'], + ] as const)('immediately halves on critical %s', (_label, sample, reason) => { + const capacity = controller({ + initialInflight: 7, + minInflight: 1, + maxInflight: 8, + configuredCoverageBatch: 7, + }); + + const status = capacity.observe(sample, 1_000); + + expect(status).toMatchObject({ + state: 'constrained', + currentInflight: 3, + effectiveCoverageBatch: 3, + cooldownUntilMs: 31_000, + lastDecision: { + action: 'halve', + reason, + previousInflight: 7, + currentInflight: 3, + previousCoverageBatch: 7, + currentCoverageBatch: 3, + }, + }); + }); + + it.each([ + ['normal store queue', healthySample({ store: { ...healthyStore, normalQueued: 1 } }), 'strained_store_queue'], + ['background store queue', healthySample({ store: { ...healthyStore, backgroundQueued: 1 } }), 'strained_store_queue'], + ['CPU', healthySample({ cpuUtilization: 0.85 }), 'strained_cpu'], + ['heap', healthySample({ heapRatio: 0.72 }), 'strained_heap'], + ['event loop', healthySample({ eventLoopUtilization: 0.8 }), 'strained_event_loop'], + ] as const)('requires two consecutive strained %s samples', (_label, sample, reason) => { + const capacity = controller({ + initialInflight: 4, + maxInflight: 8, + configuredCoverageBatch: 4, + }); + + expect(capacity.observe(sample, 5_000)).toMatchObject({ + currentInflight: 4, + effectiveCoverageBatch: 4, + consecutiveStrainedSamples: 1, + lastDecision: { action: 'hold', reason: 'strained_hysteresis' }, + }); + expect(capacity.observe(sample, 10_000)).toMatchObject({ + state: 'constrained', + currentInflight: 3, + effectiveCoverageBatch: 3, + consecutiveStrainedSamples: 0, + lastDecision: { action: 'decrease', reason }, + }); + }); + + it('requires consecutive strain and holds when signals are ambiguous', () => { + const capacity = controller({ initialInflight: 4, maxInflight: 8 }); + const strained = healthySample({ cpuUtilization: 0.85 }); + + capacity.observe(strained, 5_000); + const ambiguous = capacity.observe({ + demand: true, + heapRatio: 0.65, + store: { telemetryAvailable: false }, + }, 10_000); + const nextStrained = capacity.observe(strained, 15_000); + + expect(ambiguous.lastDecision.reason).toBe('ambiguous_signals'); + expect(nextStrained).toMatchObject({ + currentInflight: 4, + consecutiveStrainedSamples: 1, + }); + }); + + it('uses the cooldown to prevent an immediate healthy rebound', () => { + const capacity = controller({ initialInflight: 4, maxInflight: 8 }); + capacity.observe(healthySample({ heapRatio: 0.82 }), 1_000); + + for (let sample = 1; sample <= 6; sample += 1) { + capacity.observe(healthySample(), 1_000 + sample * 4_000); + } + expect(capacity.getStatus()).toMatchObject({ + state: 'cooldown', + currentInflight: 2, + consecutiveHealthyDemandSamples: 6, + lastDecision: { action: 'hold', reason: 'cooldown' }, + }); + + expect(capacity.observe(healthySample(), 31_000)).toMatchObject({ + currentInflight: 3, + consecutiveHealthyDemandSamples: 0, + lastDecision: { action: 'increase' }, + }); + }); + + it('allows host-healthy growth to two without store telemetry but never above it', () => { + const capacity = controller({ + initialInflight: 1, + minInflight: 1, + maxInflight: 6, + configuredCoverageBatch: 0, + }); + const noStoreTelemetry = healthySample({ + store: { telemetryAvailable: false }, + }); + + for (let sample = 1; sample <= 6; sample += 1) { + capacity.observe(noStoreTelemetry, sample * 5_000); + } + expect(capacity.getStatus()).toMatchObject({ + currentInflight: 2, + effectiveCoverageBatch: 0, + storePressureTelemetryAvailable: false, + }); + + for (let sample = 7; sample <= 18; sample += 1) { + capacity.observe(noStoreTelemetry, sample * 5_000); + } + expect(capacity.getStatus()).toMatchObject({ + state: 'constrained', + currentInflight: 2, + effectiveCoverageBatch: 0, + lastDecision: { action: 'hold', reason: 'store_telemetry_growth_cap' }, + }); + }); + + it('restores a reduced coverage batch without exceeding its configured maximum', () => { + const capacity = controller({ + initialInflight: 2, + maxInflight: 2, + configuredCoverageBatch: 4, + }); + capacity.observe(healthySample({ heapRatio: 0.82 }), 0); + expect(capacity.getStatus().effectiveCoverageBatch).toBe(2); + + for (let sample = 1; sample <= 6; sample += 1) { + capacity.observe(healthySample(), 30_000 + sample * 5_000); + } + expect(capacity.getStatus()).toMatchObject({ + currentInflight: 2, + effectiveCoverageBatch: 3, + lastDecision: { action: 'increase' }, + }); + + for (let sample = 1; sample <= 6; sample += 1) { + capacity.observe(healthySample(), 60_000 + sample * 5_000); + } + expect(capacity.getStatus().effectiveCoverageBatch).toBe(4); + }); + + it('never enables a configured zero coverage batch', () => { + const capacity = controller({ + initialInflight: 4, + maxInflight: 8, + configuredCoverageBatch: 0, + }); + capacity.observe(healthySample({ heapRatio: 0.82 }), 1_000); + for (let sample = 1; sample <= 12; sample += 1) { + capacity.observe(healthySample(), 31_000 + sample * 5_000); + } + + expect(capacity.getStatus().effectiveCoverageBatch).toBe(0); + }); + + it('rejects malformed samples instead of making unsafe decisions', () => { + const capacity = controller(); + expect(() => capacity.observe(healthySample({ cpuUtilization: 1.1 }), 1)) + .toThrow(/cpuUtilization/); + expect(() => capacity.observe(healthySample({ heapRatio: Number.NaN }), 1)) + .toThrow(/heapRatio/); + expect(() => capacity.observe(healthySample({ + store: { ...healthyStore, normalQueued: -1 }, + }), 1)).toThrow(/store.normalQueued/); + expect(() => capacity.observe(healthySample(), Number.NaN)).toThrow(/atMs/); + }); +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 5e71c1fc41..7e946ca6c2 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -86,6 +86,7 @@ export default defineConfig({ "test/peer-selection.test.ts", "test/sync-requester-priority.test.ts", "test/core-public-coverage-scheduler.test.ts", + "test/adaptive-capacity.test.ts", "test/sync-requester-progress.test.ts", "test/rootless-durable-bounded-progress.test.ts", "test/rootless-durable-skips-legacy-partition.test.ts", From fb66e5a0cd9a3839da63e6d58cc4c89b1db92d8e Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 03:36:06 +0200 Subject: [PATCH 4/9] feat(sync): sample host and store capacity --- .../src/sync/adaptive-capacity-sampler.ts | 152 ++++++++++++++++++ .../test/adaptive-capacity-sampler.test.ts | 89 ++++++++++ packages/agent/vitest.unit.config.ts | 1 + 3 files changed, 242 insertions(+) create mode 100644 packages/agent/src/sync/adaptive-capacity-sampler.ts create mode 100644 packages/agent/test/adaptive-capacity-sampler.test.ts diff --git a/packages/agent/src/sync/adaptive-capacity-sampler.ts b/packages/agent/src/sync/adaptive-capacity-sampler.ts new file mode 100644 index 0000000000..b7811c5912 --- /dev/null +++ b/packages/agent/src/sync/adaptive-capacity-sampler.ts @@ -0,0 +1,152 @@ +import { availableParallelism, cpus } from 'node:os'; +import { performance, type EventLoopUtilization } from 'node:perf_hooks'; +import { memoryUsage } from 'node:process'; +import { getHeapStatistics } from 'node:v8'; +import type { StorePressureSnapshot, TripleStore } from '@origintrail-official/dkg-storage'; +import { + MAX_SYNC_ADAPTIVE_INFLIGHT, + type AdaptiveCapacitySample, +} from './adaptive-capacity.js'; + +export interface CpuTimeSnapshot { + idle: number; + total: number; +} + +export interface AdaptiveCapacitySamplerDependencies { + readCpuTimes?: () => CpuTimeSnapshot; + readEventLoopUtilization?: () => EventLoopUtilization; + eventLoopUtilizationDelta?: ( + current: EventLoopUtilization, + previous: EventLoopUtilization, + ) => number; + readHeapRatio?: () => number | undefined; +} + +function defaultCpuTimes(): CpuTimeSnapshot { + let idle = 0; + let total = 0; + for (const cpu of cpus()) { + idle += cpu.times.idle; + total += Object.values(cpu.times).reduce((sum, value) => sum + value, 0); + } + return { idle, total }; +} + +function intervalCpuUtilization( + current: CpuTimeSnapshot, + previous: CpuTimeSnapshot, +): number | undefined { + const totalDelta = current.total - previous.total; + const idleDelta = current.idle - previous.idle; + if (totalDelta <= 0 || idleDelta < 0) return undefined; + return Math.max(0, Math.min(1, 1 - (idleDelta / totalDelta))); +} + +function defaultHeapRatio(): number | undefined { + const heapLimit = getHeapStatistics().heap_size_limit; + if (!Number.isFinite(heapLimit) || heapLimit <= 0) return undefined; + return Math.max(0, Math.min(1, memoryUsage().heapUsed / heapLimit)); +} + +function storeSample(store: TripleStore): AdaptiveCapacitySample['store'] { + let pressure: StorePressureSnapshot | undefined; + try { + pressure = store.getPressureSnapshot?.(); + } catch { + return { telemetryAvailable: false }; + } + if (!pressure) return { telemetryAvailable: false }; + return { + telemetryAvailable: true, + ackQueued: pressure.ackQueued, + healthQueued: pressure.healthQueued ?? 0, + normalQueued: pressure.normalQueued, + backgroundQueued: pressure.backgroundQueued, + }; +} + +/** + * Node-local interval sampler. Ordinary full store utilization is deliberately + * not classified as saturation: critical store state requires queue-age or + * no-progress evidence that the current pressure API does not expose. + */ +export class AdaptiveCapacitySampler { + private readonly readCpuTimes: () => CpuTimeSnapshot; + private readonly readEventLoopUtilization: () => EventLoopUtilization; + private readonly eventLoopUtilizationDelta: ( + current: EventLoopUtilization, + previous: EventLoopUtilization, + ) => number; + private readonly readHeapRatio: () => number | undefined; + private previousCpuTimes: CpuTimeSnapshot; + private previousEventLoopUtilization: EventLoopUtilization; + + constructor( + private readonly store: TripleStore, + dependencies: AdaptiveCapacitySamplerDependencies = {}, + ) { + this.readCpuTimes = dependencies.readCpuTimes ?? defaultCpuTimes; + this.readEventLoopUtilization = dependencies.readEventLoopUtilization + ?? (() => performance.eventLoopUtilization()); + this.eventLoopUtilizationDelta = dependencies.eventLoopUtilizationDelta + ?? ((current, previous) => performance.eventLoopUtilization(current, previous).utilization); + this.readHeapRatio = dependencies.readHeapRatio ?? defaultHeapRatio; + this.previousCpuTimes = this.readCpuTimes(); + this.previousEventLoopUtilization = this.readEventLoopUtilization(); + } + + sample(demand: boolean): AdaptiveCapacitySample { + const currentCpuTimes = this.readCpuTimes(); + const cpuUtilization = intervalCpuUtilization(currentCpuTimes, this.previousCpuTimes); + this.previousCpuTimes = currentCpuTimes; + + const currentEventLoopUtilization = this.readEventLoopUtilization(); + const eventLoopUtilization = this.eventLoopUtilizationDelta( + currentEventLoopUtilization, + this.previousEventLoopUtilization, + ); + this.previousEventLoopUtilization = currentEventLoopUtilization; + + const heapRatio = this.readHeapRatio(); + return { + demand, + ...(cpuUtilization !== undefined ? { cpuUtilization } : {}), + ...(Number.isFinite(eventLoopUtilization) + ? { eventLoopUtilization: Math.max(0, Math.min(1, eventLoopUtilization)) } + : {}), + ...(heapRatio !== undefined ? { heapRatio } : {}), + store: storeSample(this.store), + }; + } +} + +export interface AdaptiveInflightHardMaxInput { + operatorMax?: number; + parallelism?: number; + storePressure?: StorePressureSnapshot; +} + +/** Resolve the largest requester cap the controller may ever reach. */ +export function deriveAdaptiveInflightHardMax( + input: AdaptiveInflightHardMaxInput = {}, +): number { + const operatorMax = input.operatorMax ?? MAX_SYNC_ADAPTIVE_INFLIGHT; + const parallelism = input.parallelism ?? availableParallelism(); + if (!Number.isInteger(operatorMax) || operatorMax < 1) { + throw new TypeError('adaptive operator maximum must be a positive integer'); + } + if (!Number.isInteger(parallelism) || parallelism < 1) { + throw new TypeError('available parallelism must be a positive integer'); + } + const hardwareMax = Math.max(1, Math.floor(parallelism / 2)); + const storeMax = input.storePressure + ? Math.max( + 1, + input.storePressure.maxConcurrent + - input.storePressure.ackReservedSlots + - (input.storePressure.healthReservedSlots ?? 0), + ) + : MAX_SYNC_ADAPTIVE_INFLIGHT; + return Math.min(MAX_SYNC_ADAPTIVE_INFLIGHT, operatorMax, hardwareMax, storeMax); +} diff --git a/packages/agent/test/adaptive-capacity-sampler.test.ts b/packages/agent/test/adaptive-capacity-sampler.test.ts new file mode 100644 index 0000000000..1734cae493 --- /dev/null +++ b/packages/agent/test/adaptive-capacity-sampler.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from 'vitest'; +import type { TripleStore } from '@origintrail-official/dkg-storage'; +import { + AdaptiveCapacitySampler, + deriveAdaptiveInflightHardMax, +} from '../src/sync/adaptive-capacity-sampler.js'; + +function fakeStore(pressure?: ReturnType>): TripleStore { + return { + getPressureSnapshot: () => pressure, + } as unknown as TripleStore; +} + +describe('adaptive capacity sampler', () => { + it('reports interval host utilization and live store queues', () => { + const cpu = [ + { idle: 40, total: 100 }, + { idle: 50, total: 200 }, + ]; + const eventLoop = [ + { idle: 1, active: 1, utilization: 0.5 }, + { idle: 2, active: 2, utilization: 0.5 }, + ]; + const sampler = new AdaptiveCapacitySampler(fakeStore({ + ackInflight: 0, + normalInflight: 1, + backgroundInflight: 0, + ackQueued: 0, + normalQueued: 2, + backgroundQueued: 0, + maxConcurrent: 4, + ackReservedSlots: 1, + }), { + readCpuTimes: () => cpu.shift()!, + readEventLoopUtilization: () => eventLoop.shift()!, + eventLoopUtilizationDelta: () => 0.25, + readHeapRatio: () => 0.4, + }); + + expect(sampler.sample(true)).toEqual({ + demand: true, + cpuUtilization: 0.9, + eventLoopUtilization: 0.25, + heapRatio: 0.4, + store: { + telemetryAvailable: true, + ackQueued: 0, + healthQueued: 0, + normalQueued: 2, + backgroundQueued: 0, + }, + }); + }); + + it('fails closed to unavailable store telemetry', () => { + const sampler = new AdaptiveCapacitySampler(fakeStore(), { + readCpuTimes: () => ({ idle: 1, total: 2 }), + readEventLoopUtilization: () => ({ idle: 1, active: 1, utilization: 0.5 }), + eventLoopUtilizationDelta: () => Number.NaN, + readHeapRatio: () => undefined, + }); + + expect(sampler.sample(false)).toMatchObject({ + demand: false, + store: { telemetryAvailable: false }, + }); + }); + + it('caps the hard maximum by operator, hardware, and store capacity', () => { + expect(deriveAdaptiveInflightHardMax({ + operatorMax: 7, + parallelism: 12, + storePressure: { + ackInflight: 0, + healthInflight: 0, + normalInflight: 0, + backgroundInflight: 0, + ackQueued: 0, + healthQueued: 0, + normalQueued: 0, + backgroundQueued: 0, + maxConcurrent: 5, + ackReservedSlots: 1, + healthReservedSlots: 1, + }, + })).toBe(3); + expect(deriveAdaptiveInflightHardMax({ operatorMax: 8, parallelism: 2 })).toBe(1); + }); +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 7e946ca6c2..e1e677ed50 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -87,6 +87,7 @@ export default defineConfig({ "test/sync-requester-priority.test.ts", "test/core-public-coverage-scheduler.test.ts", "test/adaptive-capacity.test.ts", + "test/adaptive-capacity-sampler.test.ts", "test/sync-requester-progress.test.ts", "test/rootless-durable-bounded-progress.test.ts", "test/rootless-durable-skips-legacy-partition.test.ts", From e44e6d0c9cbde04fb2bac8ea94f6287aa993afe5 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 04:39:36 +0200 Subject: [PATCH 5/9] refactor(sync): reduce adaptive capacity transitions --- packages/agent/src/sync/adaptive-capacity.ts | 389 ++++++++++++------ packages/agent/test/adaptive-capacity.test.ts | 75 ++++ 2 files changed, 327 insertions(+), 137 deletions(-) diff --git a/packages/agent/src/sync/adaptive-capacity.ts b/packages/agent/src/sync/adaptive-capacity.ts index 8f71753340..ee3c039451 100644 --- a/packages/agent/src/sync/adaptive-capacity.ts +++ b/packages/agent/src/sync/adaptive-capacity.ts @@ -198,6 +198,41 @@ interface ClassifiedSample { reason: AdaptiveCapacityReason; } +interface CapacityWindow { + readonly inflight: number; + readonly coverageBatch: number; +} + +interface CapacityCounters { + readonly strainedSamples: number; + readonly healthyDemandSamples: number; +} + +interface CapacityControllerState { + readonly window: Readonly; + readonly counters: Readonly; + readonly state: AdaptiveCapacityState; + readonly cooldownUntilMs: number; + readonly storePressureTelemetryAvailable: boolean; +} + +interface CapacityTransitionContext { + readonly current: Readonly; + readonly classified: Readonly; + readonly sample: Readonly; + readonly bounds: Extract; + readonly atMs: number; + readonly cooldownMs: number; + readonly healthySamplesToGrow: number; + readonly strainedSamplesToShrink: number; +} + +interface CapacityTransition { + readonly action: AdaptiveCapacityAction; + readonly reason: AdaptiveCapacityReason; + readonly next: Readonly; +} + function validateRatio(name: string, value: number | undefined): void { if (value === undefined) return; if (!Number.isFinite(value) || value < 0 || value > 1) { @@ -282,6 +317,178 @@ function classifySample(sample: AdaptiveCapacitySample): ClassifiedSample { return { kind: 'ambiguous', reason: 'ambiguous_signals' }; } +function buildTransition( + context: CapacityTransitionContext, + action: AdaptiveCapacityAction, + reason: AdaptiveCapacityReason, + next: { + window?: Readonly; + counters?: Readonly; + state: AdaptiveCapacityState; + cooldownUntilMs?: number; + }, +): CapacityTransition { + return { + action, + reason, + next: { + window: next.window ?? context.current.window, + counters: next.counters ?? context.current.counters, + state: next.state, + cooldownUntilMs: next.cooldownUntilMs ?? context.current.cooldownUntilMs, + storePressureTelemetryAvailable: context.sample.store.telemetryAvailable, + }, + }; +} + +function holdCapacity( + context: CapacityTransitionContext, + reason: AdaptiveCapacityReason, + state: AdaptiveCapacityState, + counters: Readonly = context.current.counters, +): CapacityTransition { + return buildTransition(context, 'hold', reason, { state, counters }); +} + +function halveCapacity( + context: CapacityTransitionContext, +): CapacityTransition { + const { current, bounds, classified, atMs, cooldownMs } = context; + return buildTransition(context, 'halve', classified.reason, { + window: { + inflight: Math.max(bounds.minInflight, Math.floor(current.window.inflight / 2)), + coverageBatch: current.window.coverageBatch > 0 + ? Math.max(1, Math.floor(current.window.coverageBatch / 2)) + : 0, + }, + counters: { strainedSamples: 0, healthyDemandSamples: 0 }, + state: 'constrained', + cooldownUntilMs: atMs + cooldownMs, + }); +} + +function decreaseCapacity( + context: CapacityTransitionContext, +): CapacityTransition { + const { current, bounds, classified, atMs, cooldownMs } = context; + return buildTransition(context, 'decrease', classified.reason, { + window: { + inflight: Math.max(bounds.minInflight, current.window.inflight - 1), + coverageBatch: current.window.coverageBatch > 0 + ? Math.max(1, current.window.coverageBatch - 1) + : 0, + }, + counters: { strainedSamples: 0, healthyDemandSamples: 0 }, + state: 'constrained', + cooldownUntilMs: atMs + cooldownMs, + }); +} + +function increaseCapacity( + context: CapacityTransitionContext, + inflightCanGrow: boolean, + coverageCanGrow: boolean, +): CapacityTransition { + const { current, atMs, cooldownMs } = context; + return buildTransition(context, 'increase', 'healthy_hysteresis', { + window: { + inflight: current.window.inflight + (inflightCanGrow ? 1 : 0), + coverageBatch: current.window.coverageBatch + (coverageCanGrow ? 1 : 0), + }, + counters: { strainedSamples: 0, healthyDemandSamples: 0 }, + state: 'cooldown', + cooldownUntilMs: atMs + cooldownMs, + }); +} + +function calculateCapacityTransition( + context: CapacityTransitionContext, +): CapacityTransition { + const { + current, + classified, + sample, + atMs, + healthySamplesToGrow, + strainedSamplesToShrink, + bounds, + } = context; + const waitingForCooldown = atMs < current.cooldownUntilMs; + + if (classified.kind === 'critical') return halveCapacity(context); + + if (classified.kind === 'strained') { + const strainedSamples = current.counters.strainedSamples + 1; + if (strainedSamples < strainedSamplesToShrink) { + return holdCapacity( + context, + 'strained_hysteresis', + waitingForCooldown ? 'cooldown' : 'warming', + { strainedSamples, healthyDemandSamples: 0 }, + ); + } + return decreaseCapacity(context); + } + + const resetStrainedCounters = { + strainedSamples: 0, + healthyDemandSamples: current.counters.healthyDemandSamples, + }; + if (classified.kind === 'ambiguous') { + return holdCapacity( + context, + classified.reason, + waitingForCooldown ? 'cooldown' : 'warming', + { strainedSamples: 0, healthyDemandSamples: 0 }, + ); + } + + if (!sample.demand) { + return holdCapacity( + context, + 'no_demand', + waitingForCooldown ? 'cooldown' : 'healthy', + { strainedSamples: 0, healthyDemandSamples: 0 }, + ); + } + + const healthyDemandSamples = Math.min( + healthySamplesToGrow, + resetStrainedCounters.healthyDemandSamples + 1, + ); + const healthyCounters = { strainedSamples: 0, healthyDemandSamples }; + if (healthyDemandSamples < healthySamplesToGrow) { + return holdCapacity( + context, + 'healthy_hysteresis', + waitingForCooldown ? 'cooldown' : 'warming', + healthyCounters, + ); + } + if (waitingForCooldown) { + return holdCapacity(context, 'cooldown', 'cooldown', healthyCounters); + } + + const growthInflightCeiling = sample.store.telemetryAvailable + ? bounds.maxInflight + : Math.min(bounds.maxInflight, DEFAULT_SYNC_ADAPTIVE_INITIAL_INFLIGHT); + const inflightCanGrow = current.window.inflight < growthInflightCeiling; + const coverageCanGrow = current.window.coverageBatch > 0 + && current.window.coverageBatch < bounds.configuredCoverageBatch; + if (!inflightCanGrow && !coverageCanGrow) { + return holdCapacity( + context, + !sample.store.telemetryAvailable && current.window.inflight < bounds.maxInflight + ? 'store_telemetry_growth_cap' + : 'at_maximum', + sample.store.telemetryAvailable ? 'healthy' : 'constrained', + healthyCounters, + ); + } + + return increaseCapacity(context, inflightCanGrow, coverageCanGrow); +} + /** * Pure fast-down/slow-up AIMD controller. Sampling and application of the * returned limits are deliberately owned by the DKGAgent integration layer. @@ -291,13 +498,7 @@ export class AdaptiveCapacityController { private readonly cooldownMs: number; private readonly healthySamplesToGrow: number; private readonly strainedSamplesToShrink: number; - private currentInflight: number; - private currentCoverageBatch: number; - private state: AdaptiveCapacityState = 'warming'; - private storePressureTelemetryAvailable = false; - private consecutiveStrainedSamples = 0; - private consecutiveHealthyDemandSamples = 0; - private cooldownUntilMs: number; + private controllerState: Readonly; private lastDecision: Readonly; constructor( @@ -320,149 +521,66 @@ export class AdaptiveCapacityController { options.strainedSamplesToShrink ?? DEFAULT_SYNC_ADAPTIVE_STRAINED_SAMPLES, 1, ); - this.currentInflight = bounds.initialInflight; - this.currentCoverageBatch = bounds.configuredCoverageBatch; const createdAt = this.now(); if (!Number.isFinite(createdAt)) throw new RangeError('now must return a finite timestamp'); - this.cooldownUntilMs = createdAt + this.cooldownMs; + this.controllerState = { + window: { + inflight: bounds.initialInflight, + coverageBatch: bounds.configuredCoverageBatch, + }, + counters: { strainedSamples: 0, healthyDemandSamples: 0 }, + state: 'warming', + cooldownUntilMs: createdAt + this.cooldownMs, + storePressureTelemetryAvailable: false, + }; this.lastDecision = Object.freeze({ action: 'hold', reason: 'warming', atMs: createdAt, - previousInflight: this.currentInflight, - currentInflight: this.currentInflight, - previousCoverageBatch: this.currentCoverageBatch, - currentCoverageBatch: this.currentCoverageBatch, + previousInflight: bounds.initialInflight, + currentInflight: bounds.initialInflight, + previousCoverageBatch: bounds.configuredCoverageBatch, + currentCoverageBatch: bounds.configuredCoverageBatch, }); } observe(sample: AdaptiveCapacitySample, atMs = this.now()): AdaptiveCapacityStatus { this.validateSample(sample, atMs); - this.storePressureTelemetryAvailable = sample.store.telemetryAvailable; - const classified = classifySample(sample); - - if (classified.kind === 'critical') { - this.consecutiveStrainedSamples = 0; - this.consecutiveHealthyDemandSamples = 0; - this.applyDecision('halve', classified.reason, atMs, () => { - this.currentInflight = Math.max( - this.bounds.minInflight, - Math.floor(this.currentInflight / 2), - ); - if (this.currentCoverageBatch > 0) { - this.currentCoverageBatch = Math.max( - 1, - Math.floor(this.currentCoverageBatch / 2), - ); - } - }); - this.state = 'constrained'; - this.cooldownUntilMs = atMs + this.cooldownMs; - return this.getStatus(); - } - - if (classified.kind === 'strained') { - this.consecutiveHealthyDemandSamples = 0; - this.consecutiveStrainedSamples += 1; - if (this.consecutiveStrainedSamples < this.strainedSamplesToShrink) { - this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; - this.applyDecision('hold', 'strained_hysteresis', atMs); - return this.getStatus(); - } - - this.consecutiveStrainedSamples = 0; - this.applyDecision('decrease', classified.reason, atMs, () => { - this.currentInflight = Math.max( - this.bounds.minInflight, - this.currentInflight - 1, - ); - if (this.currentCoverageBatch > 0) { - this.currentCoverageBatch = Math.max(1, this.currentCoverageBatch - 1); - } - }); - this.state = 'constrained'; - this.cooldownUntilMs = atMs + this.cooldownMs; - return this.getStatus(); - } - - this.consecutiveStrainedSamples = 0; - if (classified.kind === 'ambiguous') { - this.consecutiveHealthyDemandSamples = 0; - this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; - this.applyDecision('hold', classified.reason, atMs); - return this.getStatus(); - } - - if (!sample.demand) { - this.consecutiveHealthyDemandSamples = 0; - this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'healthy'; - this.applyDecision('hold', 'no_demand', atMs); - return this.getStatus(); - } - - this.consecutiveHealthyDemandSamples = Math.min( - this.healthySamplesToGrow, - this.consecutiveHealthyDemandSamples + 1, - ); - if (this.consecutiveHealthyDemandSamples < this.healthySamplesToGrow) { - this.state = atMs < this.cooldownUntilMs ? 'cooldown' : 'warming'; - this.applyDecision('hold', 'healthy_hysteresis', atMs); - return this.getStatus(); - } - if (atMs < this.cooldownUntilMs) { - this.state = 'cooldown'; - this.applyDecision('hold', 'cooldown', atMs); - return this.getStatus(); - } - - const growthInflightCeiling = sample.store.telemetryAvailable - ? this.bounds.maxInflight - : Math.min(this.bounds.maxInflight, DEFAULT_SYNC_ADAPTIVE_INITIAL_INFLIGHT); - const inflightCanGrow = this.currentInflight < growthInflightCeiling; - const coverageCanGrow = this.currentCoverageBatch > 0 - && this.currentCoverageBatch < this.bounds.configuredCoverageBatch; - if (!inflightCanGrow && !coverageCanGrow) { - this.state = sample.store.telemetryAvailable ? 'healthy' : 'constrained'; - this.applyDecision( - 'hold', - !sample.store.telemetryAvailable && this.currentInflight < this.bounds.maxInflight - ? 'store_telemetry_growth_cap' - : 'at_maximum', - atMs, - ); - return this.getStatus(); - } - - this.applyDecision('increase', 'healthy_hysteresis', atMs, () => { - if (inflightCanGrow) this.currentInflight += 1; - if (coverageCanGrow) this.currentCoverageBatch += 1; + const transition = calculateCapacityTransition({ + current: this.controllerState, + classified: classifySample(sample), + sample, + bounds: this.bounds, + atMs, + cooldownMs: this.cooldownMs, + healthySamplesToGrow: this.healthySamplesToGrow, + strainedSamplesToShrink: this.strainedSamplesToShrink, }); - this.consecutiveHealthyDemandSamples = 0; - this.cooldownUntilMs = atMs + this.cooldownMs; - this.state = 'cooldown'; + this.applyTransition(transition, atMs); return this.getStatus(); } getCurrentInflight(): number { - return this.currentInflight; + return this.controllerState.window.inflight; } getEffectiveCoverageBatch(): number { - return this.currentCoverageBatch; + return this.controllerState.window.coverageBatch; } getStatus(): AdaptiveCapacityStatus { + const { window, counters } = this.controllerState; return Object.freeze({ - state: this.state, - currentInflight: this.currentInflight, + state: this.controllerState.state, + currentInflight: window.inflight, minInflight: this.bounds.minInflight, maxInflight: this.bounds.maxInflight, - effectiveCoverageBatch: this.currentCoverageBatch, + effectiveCoverageBatch: window.coverageBatch, configuredCoverageBatch: this.bounds.configuredCoverageBatch, - storePressureTelemetryAvailable: this.storePressureTelemetryAvailable, - consecutiveStrainedSamples: this.consecutiveStrainedSamples, - consecutiveHealthyDemandSamples: this.consecutiveHealthyDemandSamples, - cooldownUntilMs: this.cooldownUntilMs, + storePressureTelemetryAvailable: this.controllerState.storePressureTelemetryAvailable, + consecutiveStrainedSamples: counters.strainedSamples, + consecutiveHealthyDemandSamples: counters.healthyDemandSamples, + cooldownUntilMs: this.controllerState.cooldownUntilMs, lastDecision: this.lastDecision, }); } @@ -480,23 +598,20 @@ export class AdaptiveCapacityController { } } - private applyDecision( - action: AdaptiveCapacityAction, - reason: AdaptiveCapacityReason, + private applyTransition( + transition: Readonly, atMs: number, - mutate?: () => void, ): void { - const previousInflight = this.currentInflight; - const previousCoverageBatch = this.currentCoverageBatch; - mutate?.(); + const previous = this.controllerState.window; + this.controllerState = transition.next; this.lastDecision = Object.freeze({ - action, - reason, + action: transition.action, + reason: transition.reason, atMs, - previousInflight, - currentInflight: this.currentInflight, - previousCoverageBatch, - currentCoverageBatch: this.currentCoverageBatch, + previousInflight: previous.inflight, + currentInflight: transition.next.window.inflight, + previousCoverageBatch: previous.coverageBatch, + currentCoverageBatch: transition.next.window.coverageBatch, }); } } diff --git a/packages/agent/test/adaptive-capacity.test.ts b/packages/agent/test/adaptive-capacity.test.ts index 76212607d3..3c06c6af09 100644 --- a/packages/agent/test/adaptive-capacity.test.ts +++ b/packages/agent/test/adaptive-capacity.test.ts @@ -252,6 +252,81 @@ describe('adaptive sync capacity controller', () => { }); }); + it('keeps decision endpoints aligned with each applied capacity window', () => { + const capacity = controller({ + initialInflight: 4, + maxInflight: 8, + configuredCoverageBatch: 4, + }); + const strained = healthySample({ cpuUtilization: 0.85 }); + + expect(capacity.observe(strained, 5_000)).toMatchObject({ + state: 'cooldown', + currentInflight: 4, + effectiveCoverageBatch: 4, + consecutiveStrainedSamples: 1, + consecutiveHealthyDemandSamples: 0, + lastDecision: { + action: 'hold', + reason: 'strained_hysteresis', + previousInflight: 4, + currentInflight: 4, + previousCoverageBatch: 4, + currentCoverageBatch: 4, + }, + }); + expect(capacity.observe(strained, 10_000)).toMatchObject({ + state: 'constrained', + currentInflight: 3, + effectiveCoverageBatch: 3, + consecutiveStrainedSamples: 0, + consecutiveHealthyDemandSamples: 0, + cooldownUntilMs: 40_000, + lastDecision: { + action: 'decrease', + reason: 'strained_cpu', + previousInflight: 4, + currentInflight: 3, + previousCoverageBatch: 4, + currentCoverageBatch: 3, + }, + }); + expect(capacity.observe(healthySample({ heapRatio: 0.82 }), 15_000)).toMatchObject({ + state: 'constrained', + currentInflight: 1, + effectiveCoverageBatch: 1, + cooldownUntilMs: 45_000, + lastDecision: { + action: 'halve', + reason: 'critical_heap', + previousInflight: 3, + currentInflight: 1, + previousCoverageBatch: 3, + currentCoverageBatch: 1, + }, + }); + + for (let sample = 1; sample <= 5; sample += 1) { + capacity.observe(healthySample(), 15_000 + sample * 5_000); + } + expect(capacity.observe(healthySample(), 45_000)).toMatchObject({ + state: 'cooldown', + currentInflight: 2, + effectiveCoverageBatch: 2, + consecutiveStrainedSamples: 0, + consecutiveHealthyDemandSamples: 0, + cooldownUntilMs: 75_000, + lastDecision: { + action: 'increase', + reason: 'healthy_hysteresis', + previousInflight: 1, + currentInflight: 2, + previousCoverageBatch: 1, + currentCoverageBatch: 2, + }, + }); + }); + it('allows host-healthy growth to two without store telemetry but never above it', () => { const capacity = controller({ initialInflight: 1, From dec30a7a56e3e0682f15c164ed0e630fb495114d Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:25:51 +0200 Subject: [PATCH 6/9] fix(sync): clarify adaptive coverage contracts --- .../sync/core-public-coverage-scheduler.ts | 23 +++++++++- .../test/adaptive-capacity-sampler.test.ts | 1 + .../core-public-coverage-scheduler.test.ts | 45 ++++++++++++------- 3 files changed, 51 insertions(+), 18 deletions(-) diff --git a/packages/agent/src/sync/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index 6f6cfc6de9..2b295f6506 100644 --- a/packages/agent/src/sync/core-public-coverage-scheduler.ts +++ b/packages/agent/src/sync/core-public-coverage-scheduler.ts @@ -19,6 +19,12 @@ export interface CorePublicSyncCoverageStatus { }; } +export interface CorePublicSyncCoveragePlanOptions { + priorities?: Readonly; + planningLane?: string; + effectiveBatchSize?: number; +} + function normalizeBatchSize(value: number | undefined): number { if (value === undefined) return DEFAULT_CORE_PUBLIC_SYNC_BATCH_SIZE; if (!Number.isInteger(value) || value < 0) { @@ -122,8 +128,23 @@ export class CorePublicSyncCoverageScheduler { selectedContextGraphIds: readonly string[], priorities?: Readonly, planningLane = 'default', - effectiveBatchSize?: number, ): string[] { + return this.planAutomaticCoverageWithOptions(selectedContextGraphIds, { + priorities, + planningLane, + }); + } + + /** Named adaptive planning boundary; existing static callers keep their exact call shape. */ + planAutomaticCoverageWithOptions( + selectedContextGraphIds: readonly string[], + options: Readonly = {}, + ): string[] { + const { + priorities, + planningLane = 'default', + effectiveBatchSize, + } = options; const selected = [...new Set( selectedContextGraphIds.map((id) => id.trim()).filter(Boolean), )]; diff --git a/packages/agent/test/adaptive-capacity-sampler.test.ts b/packages/agent/test/adaptive-capacity-sampler.test.ts index 1734cae493..464002af3a 100644 --- a/packages/agent/test/adaptive-capacity-sampler.test.ts +++ b/packages/agent/test/adaptive-capacity-sampler.test.ts @@ -67,6 +67,7 @@ describe('adaptive capacity sampler', () => { }); it('caps the hard maximum by operator, hardware, and store capacity', () => { + expect(deriveAdaptiveInflightHardMax({ operatorMax: 3, parallelism: 16 })).toBe(3); expect(deriveAdaptiveInflightHardMax({ operatorMax: 7, parallelism: 12, diff --git a/packages/agent/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index 07e0b0fd9c..489fd5407c 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -151,20 +151,16 @@ describe('Core public Context Graph coverage scheduler', () => { scheduler.register(contextGraphId); } - const constrained = scheduler.planAutomaticCoverage( - ['cg:selected'], - undefined, - 'peer-a', - 1, - ); + const constrained = scheduler.planAutomaticCoverageWithOptions(['cg:selected'], { + planningLane: 'peer-a', + effectiveBatchSize: 1, + }); expect(constrained).toHaveLength(1); - const recovered = scheduler.planAutomaticCoverage( - ['cg:selected'], - undefined, - 'peer-a', - 20, - ); + const recovered = scheduler.planAutomaticCoverageWithOptions(['cg:selected'], { + planningLane: 'peer-a', + effectiveBatchSize: 20, + }); expect(recovered).toHaveLength(4); expect(recovered).not.toContain('cg:selected'); }); @@ -175,9 +171,18 @@ describe('Core public Context Graph coverage scheduler', () => { scheduler.register(contextGraphId); } - const first = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1); - const second = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 2); - const third = scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1); + const first = scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: 1, + }); + const second = scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: 2, + }); + const third = scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: 1, + }); expect(first).toEqual(['cg:a']); expect(second).toEqual(['cg:c', 'cg:d']); @@ -188,10 +193,16 @@ describe('Core public Context Graph coverage scheduler', () => { const scheduler = new CorePublicSyncCoverageScheduler(3); scheduler.register('cg:a'); - expect(() => scheduler.planAutomaticCoverage([], undefined, 'peer-a', -1)).toThrow( + expect(() => scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: -1, + })).toThrow( /effective Core public sync batch size/, ); - expect(() => scheduler.planAutomaticCoverage([], undefined, 'peer-a', 1.5)).toThrow( + expect(() => scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: 1.5, + })).toThrow( /effective Core public sync batch size/, ); }); From 2b24b2e6e0b5f600ee019fd8785efc904a3ebb4b Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:44:00 +0200 Subject: [PATCH 7/9] fix(sync): preserve normal store headroom --- .../src/sync/adaptive-capacity-sampler.ts | 3 ++- .../test/adaptive-capacity-sampler.test.ts | 18 ++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/packages/agent/src/sync/adaptive-capacity-sampler.ts b/packages/agent/src/sync/adaptive-capacity-sampler.ts index b7811c5912..19a4c6f85b 100644 --- a/packages/agent/src/sync/adaptive-capacity-sampler.ts +++ b/packages/agent/src/sync/adaptive-capacity-sampler.ts @@ -145,7 +145,8 @@ export function deriveAdaptiveInflightHardMax( 1, input.storePressure.maxConcurrent - input.storePressure.ackReservedSlots - - (input.storePressure.healthReservedSlots ?? 0), + - (input.storePressure.healthReservedSlots ?? 0) + - (input.storePressure.normalReservedSlots ?? 0), ) : MAX_SYNC_ADAPTIVE_INFLIGHT; return Math.min(MAX_SYNC_ADAPTIVE_INFLIGHT, operatorMax, hardwareMax, storeMax); diff --git a/packages/agent/test/adaptive-capacity-sampler.test.ts b/packages/agent/test/adaptive-capacity-sampler.test.ts index 464002af3a..ba66b8c3b8 100644 --- a/packages/agent/test/adaptive-capacity-sampler.test.ts +++ b/packages/agent/test/adaptive-capacity-sampler.test.ts @@ -68,6 +68,24 @@ describe('adaptive capacity sampler', () => { it('caps the hard maximum by operator, hardware, and store capacity', () => { expect(deriveAdaptiveInflightHardMax({ operatorMax: 3, parallelism: 16 })).toBe(3); + expect(deriveAdaptiveInflightHardMax({ + operatorMax: 8, + parallelism: 16, + storePressure: { + ackInflight: 0, + healthInflight: 0, + normalInflight: 0, + backgroundInflight: 0, + ackQueued: 0, + healthQueued: 0, + normalQueued: 0, + backgroundQueued: 0, + maxConcurrent: 4, + ackReservedSlots: 1, + healthReservedSlots: 1, + normalReservedSlots: 1, + }, + })).toBe(1); expect(deriveAdaptiveInflightHardMax({ operatorMax: 7, parallelism: 12, From 58b9656b7ce83f539633893c6f9d955089dfe92c Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 14:55:21 +0200 Subject: [PATCH 8/9] test(sync): preserve health queue pressure --- packages/agent/test/adaptive-capacity-sampler.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/agent/test/adaptive-capacity-sampler.test.ts b/packages/agent/test/adaptive-capacity-sampler.test.ts index ba66b8c3b8..ba1f613224 100644 --- a/packages/agent/test/adaptive-capacity-sampler.test.ts +++ b/packages/agent/test/adaptive-capacity-sampler.test.ts @@ -26,6 +26,7 @@ describe('adaptive capacity sampler', () => { normalInflight: 1, backgroundInflight: 0, ackQueued: 0, + healthQueued: 2, normalQueued: 2, backgroundQueued: 0, maxConcurrent: 4, @@ -45,7 +46,7 @@ describe('adaptive capacity sampler', () => { store: { telemetryAvailable: true, ackQueued: 0, - healthQueued: 0, + healthQueued: 2, normalQueued: 2, backgroundQueued: 0, }, From 43768351aee0d4c93e86bddf065197ed6aced496 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Sun, 2 Aug 2026 17:02:49 +0200 Subject: [PATCH 9/9] refactor(sync): derive adaptive capacity status state --- packages/agent/src/sync/adaptive-capacity.ts | 56 +++++++++++++------ packages/agent/test/adaptive-capacity.test.ts | 45 +++++++++++++++ 2 files changed, 85 insertions(+), 16 deletions(-) diff --git a/packages/agent/src/sync/adaptive-capacity.ts b/packages/agent/src/sync/adaptive-capacity.ts index ee3c039451..69038cfb55 100644 --- a/packages/agent/src/sync/adaptive-capacity.ts +++ b/packages/agent/src/sync/adaptive-capacity.ts @@ -211,7 +211,6 @@ interface CapacityCounters { interface CapacityControllerState { readonly window: Readonly; readonly counters: Readonly; - readonly state: AdaptiveCapacityState; readonly cooldownUntilMs: number; readonly storePressureTelemetryAvailable: boolean; } @@ -324,7 +323,6 @@ function buildTransition( next: { window?: Readonly; counters?: Readonly; - state: AdaptiveCapacityState; cooldownUntilMs?: number; }, ): CapacityTransition { @@ -334,7 +332,6 @@ function buildTransition( next: { window: next.window ?? context.current.window, counters: next.counters ?? context.current.counters, - state: next.state, cooldownUntilMs: next.cooldownUntilMs ?? context.current.cooldownUntilMs, storePressureTelemetryAvailable: context.sample.store.telemetryAvailable, }, @@ -344,10 +341,9 @@ function buildTransition( function holdCapacity( context: CapacityTransitionContext, reason: AdaptiveCapacityReason, - state: AdaptiveCapacityState, counters: Readonly = context.current.counters, ): CapacityTransition { - return buildTransition(context, 'hold', reason, { state, counters }); + return buildTransition(context, 'hold', reason, { counters }); } function halveCapacity( @@ -362,7 +358,6 @@ function halveCapacity( : 0, }, counters: { strainedSamples: 0, healthyDemandSamples: 0 }, - state: 'constrained', cooldownUntilMs: atMs + cooldownMs, }); } @@ -379,7 +374,6 @@ function decreaseCapacity( : 0, }, counters: { strainedSamples: 0, healthyDemandSamples: 0 }, - state: 'constrained', cooldownUntilMs: atMs + cooldownMs, }); } @@ -396,7 +390,6 @@ function increaseCapacity( coverageBatch: current.window.coverageBatch + (coverageCanGrow ? 1 : 0), }, counters: { strainedSamples: 0, healthyDemandSamples: 0 }, - state: 'cooldown', cooldownUntilMs: atMs + cooldownMs, }); } @@ -423,7 +416,6 @@ function calculateCapacityTransition( return holdCapacity( context, 'strained_hysteresis', - waitingForCooldown ? 'cooldown' : 'warming', { strainedSamples, healthyDemandSamples: 0 }, ); } @@ -438,7 +430,6 @@ function calculateCapacityTransition( return holdCapacity( context, classified.reason, - waitingForCooldown ? 'cooldown' : 'warming', { strainedSamples: 0, healthyDemandSamples: 0 }, ); } @@ -447,7 +438,6 @@ function calculateCapacityTransition( return holdCapacity( context, 'no_demand', - waitingForCooldown ? 'cooldown' : 'healthy', { strainedSamples: 0, healthyDemandSamples: 0 }, ); } @@ -461,12 +451,11 @@ function calculateCapacityTransition( return holdCapacity( context, 'healthy_hysteresis', - waitingForCooldown ? 'cooldown' : 'warming', healthyCounters, ); } if (waitingForCooldown) { - return holdCapacity(context, 'cooldown', 'cooldown', healthyCounters); + return holdCapacity(context, 'cooldown', healthyCounters); } const growthInflightCeiling = sample.store.telemetryAvailable @@ -481,7 +470,6 @@ function calculateCapacityTransition( !sample.store.telemetryAvailable && current.window.inflight < bounds.maxInflight ? 'store_telemetry_growth_cap' : 'at_maximum', - sample.store.telemetryAvailable ? 'healthy' : 'constrained', healthyCounters, ); } @@ -489,6 +477,43 @@ function calculateCapacityTransition( return increaseCapacity(context, inflightCanGrow, coverageCanGrow); } +function deriveAdaptiveCapacityState( + current: Readonly, + lastDecision: Readonly, +): AdaptiveCapacityState { + if (lastDecision.action === 'halve' || lastDecision.action === 'decrease') { + return 'constrained'; + } + if (lastDecision.action === 'increase') return 'cooldown'; + + switch (lastDecision.reason) { + case 'warming': + return 'warming'; + case 'cooldown': + return 'cooldown'; + case 'no_demand': + return lastDecision.atMs < current.cooldownUntilMs ? 'cooldown' : 'healthy'; + case 'ambiguous_signals': + case 'strained_hysteresis': + case 'healthy_hysteresis': + return lastDecision.atMs < current.cooldownUntilMs ? 'cooldown' : 'warming'; + case 'at_maximum': + return current.storePressureTelemetryAvailable ? 'healthy' : 'constrained'; + case 'store_telemetry_growth_cap': + case 'critical_ack_queue': + case 'critical_health_queue': + case 'critical_store_saturated': + case 'critical_store_stalled': + case 'critical_heap': + case 'critical_event_loop': + case 'strained_store_queue': + case 'strained_cpu': + case 'strained_heap': + case 'strained_event_loop': + return 'constrained'; + } +} + /** * Pure fast-down/slow-up AIMD controller. Sampling and application of the * returned limits are deliberately owned by the DKGAgent integration layer. @@ -529,7 +554,6 @@ export class AdaptiveCapacityController { coverageBatch: bounds.configuredCoverageBatch, }, counters: { strainedSamples: 0, healthyDemandSamples: 0 }, - state: 'warming', cooldownUntilMs: createdAt + this.cooldownMs, storePressureTelemetryAvailable: false, }; @@ -571,7 +595,7 @@ export class AdaptiveCapacityController { getStatus(): AdaptiveCapacityStatus { const { window, counters } = this.controllerState; return Object.freeze({ - state: this.controllerState.state, + state: deriveAdaptiveCapacityState(this.controllerState, this.lastDecision), currentInflight: window.inflight, minInflight: this.bounds.minInflight, maxInflight: this.bounds.maxInflight, diff --git a/packages/agent/test/adaptive-capacity.test.ts b/packages/agent/test/adaptive-capacity.test.ts index 3c06c6af09..6e7f1ee582 100644 --- a/packages/agent/test/adaptive-capacity.test.ts +++ b/packages/agent/test/adaptive-capacity.test.ts @@ -151,6 +151,51 @@ describe('adaptive sync capacity controller', () => { }); }); + it('derives presentation state from cooldown, decision, and telemetry facts', () => { + const idle = controller({ initialInflight: 2, maxInflight: 2 }); + + expect(idle.observe(healthySample({ demand: false }), 1_000)).toMatchObject({ + state: 'cooldown', + cooldownUntilMs: 30_000, + lastDecision: { action: 'hold', reason: 'no_demand', atMs: 1_000 }, + }); + expect(idle.observe(healthySample({ demand: false }), 30_000)).toMatchObject({ + state: 'healthy', + cooldownUntilMs: 30_000, + lastDecision: { action: 'hold', reason: 'no_demand', atMs: 30_000 }, + }); + expect(idle.observe({ + demand: true, + heapRatio: 0.65, + store: { telemetryAvailable: false }, + }, 31_000)).toMatchObject({ + state: 'warming', + lastDecision: { action: 'hold', reason: 'ambiguous_signals' }, + }); + + const withStoreTelemetry = controller({ initialInflight: 2, maxInflight: 2 }); + const withoutStoreTelemetry = controller({ initialInflight: 2, maxInflight: 2 }); + let telemetryStatus = withStoreTelemetry.getStatus(); + let noTelemetryStatus = withoutStoreTelemetry.getStatus(); + for (let sample = 1; sample <= 6; sample += 1) { + telemetryStatus = withStoreTelemetry.observe(healthySample(), sample * 5_000); + noTelemetryStatus = withoutStoreTelemetry.observe(healthySample({ + store: { telemetryAvailable: false }, + }), sample * 5_000); + } + + expect(telemetryStatus).toMatchObject({ + state: 'healthy', + storePressureTelemetryAvailable: true, + lastDecision: { action: 'hold', reason: 'at_maximum' }, + }); + expect(noTelemetryStatus).toMatchObject({ + state: 'constrained', + storePressureTelemetryAvailable: false, + lastDecision: { action: 'hold', reason: 'at_maximum' }, + }); + }); + it.each([ ['ACK work', healthySample({ store: { ...healthyStore, ackQueued: 1 } }), 'critical_ack_queue'], ['health work', healthySample({ store: { ...healthyStore, healthQueued: 1 } }), 'critical_health_queue'],