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..19a4c6f85b --- /dev/null +++ b/packages/agent/src/sync/adaptive-capacity-sampler.ts @@ -0,0 +1,153 @@ +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) + - (input.storePressure.normalReservedSlots ?? 0), + ) + : MAX_SYNC_ADAPTIVE_INFLIGHT; + return Math.min(MAX_SYNC_ADAPTIVE_INFLIGHT, operatorMax, hardwareMax, storeMax); +} diff --git a/packages/agent/src/sync/adaptive-capacity.ts b/packages/agent/src/sync/adaptive-capacity.ts new file mode 100644 index 0000000000..69038cfb55 --- /dev/null +++ b/packages/agent/src/sync/adaptive-capacity.ts @@ -0,0 +1,641 @@ +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; +} + +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 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) { + 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' }; +} + +function buildTransition( + context: CapacityTransitionContext, + action: AdaptiveCapacityAction, + reason: AdaptiveCapacityReason, + next: { + window?: Readonly; + counters?: Readonly; + cooldownUntilMs?: number; + }, +): CapacityTransition { + return { + action, + reason, + next: { + window: next.window ?? context.current.window, + counters: next.counters ?? context.current.counters, + cooldownUntilMs: next.cooldownUntilMs ?? context.current.cooldownUntilMs, + storePressureTelemetryAvailable: context.sample.store.telemetryAvailable, + }, + }; +} + +function holdCapacity( + context: CapacityTransitionContext, + reason: AdaptiveCapacityReason, + counters: Readonly = context.current.counters, +): CapacityTransition { + return buildTransition(context, 'hold', reason, { 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 }, + 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 }, + 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 }, + 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', + { strainedSamples, healthyDemandSamples: 0 }, + ); + } + return decreaseCapacity(context); + } + + const resetStrainedCounters = { + strainedSamples: 0, + healthyDemandSamples: current.counters.healthyDemandSamples, + }; + if (classified.kind === 'ambiguous') { + return holdCapacity( + context, + classified.reason, + { strainedSamples: 0, healthyDemandSamples: 0 }, + ); + } + + if (!sample.demand) { + return holdCapacity( + context, + 'no_demand', + { 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', + healthyCounters, + ); + } + if (waitingForCooldown) { + return holdCapacity(context, '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', + healthyCounters, + ); + } + + 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. + */ +export class AdaptiveCapacityController { + private readonly now: () => number; + private readonly cooldownMs: number; + private readonly healthySamplesToGrow: number; + private readonly strainedSamplesToShrink: number; + private controllerState: Readonly; + 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, + ); + const createdAt = this.now(); + if (!Number.isFinite(createdAt)) throw new RangeError('now must return a finite timestamp'); + this.controllerState = { + window: { + inflight: bounds.initialInflight, + coverageBatch: bounds.configuredCoverageBatch, + }, + counters: { strainedSamples: 0, healthyDemandSamples: 0 }, + cooldownUntilMs: createdAt + this.cooldownMs, + storePressureTelemetryAvailable: false, + }; + this.lastDecision = Object.freeze({ + action: 'hold', + reason: 'warming', + atMs: createdAt, + previousInflight: bounds.initialInflight, + currentInflight: bounds.initialInflight, + previousCoverageBatch: bounds.configuredCoverageBatch, + currentCoverageBatch: bounds.configuredCoverageBatch, + }); + } + + observe(sample: AdaptiveCapacitySample, atMs = this.now()): AdaptiveCapacityStatus { + this.validateSample(sample, atMs); + const transition = calculateCapacityTransition({ + current: this.controllerState, + classified: classifySample(sample), + sample, + bounds: this.bounds, + atMs, + cooldownMs: this.cooldownMs, + healthySamplesToGrow: this.healthySamplesToGrow, + strainedSamplesToShrink: this.strainedSamplesToShrink, + }); + this.applyTransition(transition, atMs); + return this.getStatus(); + } + + getCurrentInflight(): number { + return this.controllerState.window.inflight; + } + + getEffectiveCoverageBatch(): number { + return this.controllerState.window.coverageBatch; + } + + getStatus(): AdaptiveCapacityStatus { + const { window, counters } = this.controllerState; + return Object.freeze({ + state: deriveAdaptiveCapacityState(this.controllerState, this.lastDecision), + currentInflight: window.inflight, + minInflight: this.bounds.minInflight, + maxInflight: this.bounds.maxInflight, + effectiveCoverageBatch: window.coverageBatch, + configuredCoverageBatch: this.bounds.configuredCoverageBatch, + storePressureTelemetryAvailable: this.controllerState.storePressureTelemetryAvailable, + consecutiveStrainedSamples: counters.strainedSamples, + consecutiveHealthyDemandSamples: counters.healthyDemandSamples, + cooldownUntilMs: this.controllerState.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 applyTransition( + transition: Readonly, + atMs: number, + ): void { + const previous = this.controllerState.window; + this.controllerState = transition.next; + this.lastDecision = Object.freeze({ + action: transition.action, + reason: transition.reason, + atMs, + previousInflight: previous.inflight, + currentInflight: transition.next.window.inflight, + previousCoverageBatch: previous.coverageBatch, + currentCoverageBatch: transition.next.window.coverageBatch, + }); + } +} 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/core-public-coverage-scheduler.ts b/packages/agent/src/sync/core-public-coverage-scheduler.ts index 99a0dba780..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) { @@ -27,6 +33,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; @@ -112,6 +129,22 @@ export class CorePublicSyncCoverageScheduler { priorities?: Readonly, planningLane = 'default', ): 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), )]; @@ -127,8 +160,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/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/adaptive-capacity-sampler.test.ts b/packages/agent/test/adaptive-capacity-sampler.test.ts new file mode 100644 index 0000000000..ba1f613224 --- /dev/null +++ b/packages/agent/test/adaptive-capacity-sampler.test.ts @@ -0,0 +1,109 @@ +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, + healthQueued: 2, + 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: 2, + 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: 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, + 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/test/adaptive-capacity.test.ts b/packages/agent/test/adaptive-capacity.test.ts new file mode 100644 index 0000000000..6e7f1ee582 --- /dev/null +++ b/packages/agent/test/adaptive-capacity.test.ts @@ -0,0 +1,455 @@ +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('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'], + ['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('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, + 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/test/core-public-coverage-scheduler.test.ts b/packages/agent/test/core-public-coverage-scheduler.test.ts index dee5a37615..489fd5407c 100644 --- a/packages/agent/test/core-public-coverage-scheduler.test.ts +++ b/packages/agent/test/core-public-coverage-scheduler.test.ts @@ -145,6 +145,68 @@ 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.planAutomaticCoverageWithOptions(['cg:selected'], { + planningLane: 'peer-a', + effectiveBatchSize: 1, + }); + expect(constrained).toHaveLength(1); + + const recovered = scheduler.planAutomaticCoverageWithOptions(['cg:selected'], { + planningLane: 'peer-a', + effectiveBatchSize: 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.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']); + expect(third).toEqual(['cg:d']); + }); + + it('rejects invalid live automatic-coverage batches', () => { + const scheduler = new CorePublicSyncCoverageScheduler(3); + scheduler.register('cg:a'); + + expect(() => scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: -1, + })).toThrow( + /effective Core public sync batch size/, + ); + expect(() => scheduler.planAutomaticCoverageWithOptions([], { + planningLane: 'peer-a', + effectiveBatchSize: 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); 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 diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 5e71c1fc41..e1e677ed50 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -86,6 +86,8 @@ 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/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",