From 286b0c81b0025080c343ee6071a7bcddc8e3fb92 Mon Sep 17 00:00:00 2001 From: Stella Huang Date: Sat, 22 Aug 2026 19:11:19 -0700 Subject: [PATCH] fix: bound end-to-end refresh discovery latency A NativePythonFinder refresh could stall discovery indefinitely: the single-worker WorkerPool queue wait was unbounded, and CLI-fallback enrichment scaled with the environment count. Capture one monotonic operation budget (184s, derived from the existing stage-timeout constants) at enqueue. The WorkerPool expires the item with QueueTaskExpiredError if it is still queued when the budget elapses, and the same Deadline clamps every extension-controlled running stage (configure, refresh, resolve, restart backoff, CLI find + enrichment) to the remaining budget, failing fast with RefreshBudgetExceededError below a 1s floor. The CLI fallback never truncates enumeration: it retains every discovered record and only stops further enrichment when the budget is spent. resolve() and all non-refresh callers pass no deadline, so their behavior is unchanged. Both new errors classify as rpc_timeout via the existing telemetry patterns. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/common/telemetry/errorClassifier.ts | 8 +- src/common/utils/workerPool.ts | 108 ++++- src/managers/common/nativePythonFinder.ts | 217 ++++++++-- .../telemetry/errorClassifier.unit.test.ts | 16 +- src/test/common/utils/workerPool.unit.test.ts | 396 ++++++++++++++++++ .../nativePythonFinder.budget.unit.test.ts | 187 +++++++++ 6 files changed, 889 insertions(+), 43 deletions(-) create mode 100644 src/test/common/utils/workerPool.unit.test.ts create mode 100644 src/test/managers/common/nativePythonFinder.budget.unit.test.ts diff --git a/src/common/telemetry/errorClassifier.ts b/src/common/telemetry/errorClassifier.ts index 6cb20a2ab..954f0874f 100644 --- a/src/common/telemetry/errorClassifier.ts +++ b/src/common/telemetry/errorClassifier.ts @@ -1,6 +1,7 @@ import { CancellationError } from 'vscode'; import * as rpc from 'vscode-jsonrpc/node'; -import { RpcTimeoutError } from '../../managers/common/nativePythonFinder'; +import { RefreshBudgetExceededError, RpcTimeoutError } from '../../managers/common/nativePythonFinder'; +import { QueueTaskExpiredError } from '../utils/workerPool'; import { BaseError } from '../errors/types'; export type DiscoveryErrorType = @@ -49,6 +50,11 @@ export function classifyError(ex: unknown): DiscoveryErrorType { } } + // Queue-expiry and refresh-budget errors are time-budget exhaustions → generic RPC timeout category. + if (ex instanceof QueueTaskExpiredError || ex instanceof RefreshBudgetExceededError) { + return 'rpc_timeout'; + } + // JSON-RPC connection errors (e.g., PET process died mid-request, connection closed/disposed) if (ex instanceof rpc.ConnectionError) { return 'connection_error'; diff --git a/src/common/utils/workerPool.ts b/src/common/utils/workerPool.ts index 4e30ce3d2..ef6ec53cf 100644 --- a/src/common/utils/workerPool.ts +++ b/src/common/utils/workerPool.ts @@ -4,6 +4,14 @@ import { traceError } from '../logging'; import { createDeferred, Deferred } from './deferred'; +/** Rejects a queued work item that expired before a worker could dequeue it. */ +export class QueueTaskExpiredError extends Error { + constructor(expiresInMs: number) { + super(`Queued task expired after ${expiresInMs}ms before it could start`); + this.name = this.constructor.name; + } +} + interface Worker { /** * Start processing of items. @@ -23,8 +31,15 @@ type PostResult = (item: T, result?: R, err?: Error) => void; interface IWorkItem { item: T; + running: boolean; + expired: boolean; + expiryTimer?: ReturnType; + expiresAt?: number; + expiresInMs?: number; } +export type QueueClock = () => number; + export enum QueuePosition { back, front, @@ -36,9 +51,11 @@ export interface WorkerPool extends Worker { * @method addToQueue * @param {T} item: Item to process * @param {QueuePosition} position: Add items to the front or back of the queue. + * @param {number} expiresInMs: Optional. When set, a still-queued item is rejected with + * {@link QueueTaskExpiredError} after this many ms and never runs; omit to queue unbounded. * @returns A promise that when resolved gets the result from running the worker function. */ - addToQueue(item: T, position?: QueuePosition): Promise; + addToQueue(item: T, position?: QueuePosition, expiresInMs?: number): Promise; } class WorkerImpl implements Worker { @@ -76,14 +93,17 @@ class WorkerImpl implements Worker { class WorkQueue { private readonly items: IWorkItem[] = []; private readonly results: Map, Deferred> = new Map(); - public add(item: T, position?: QueuePosition): Promise { + + public constructor(private readonly now: QueueClock = Date.now) {} + + public add(item: T, position?: QueuePosition, expiresInMs?: number): Promise { // Wrap the user provided item in a wrapper object. This will allow us to track multiple // submissions of the same item. For example, addToQueue(2), addToQueue(2). If we did not // wrap this, then from the map both submissions will look the same. Since this is a generic // worker pool, we do not know if we can resolve both using the same promise. So, a better // approach is to ensure each gets a unique promise, and let the worker function figure out // how to handle repeat submissions. - const workItem: IWorkItem = { item }; + const workItem: IWorkItem = { item, running: false, expired: false }; if (position === QueuePosition.front) { this.items.unshift(workItem); } else { @@ -96,29 +116,88 @@ class WorkQueue { const deferred = createDeferred(); this.results.set(workItem, deferred); + if (expiresInMs !== undefined) { + workItem.expiresInMs = expiresInMs; + workItem.expiresAt = this.now() + expiresInMs; + workItem.expiryTimer = setTimeout(() => this.expire(workItem), Math.max(0, expiresInMs)); + } + return deferred.promise; } + private clearExpiry(workItem: IWorkItem): void { + if (workItem.expiryTimer !== undefined) { + clearTimeout(workItem.expiryTimer); + workItem.expiryTimer = undefined; + } + } + + private settleExpired(workItem: IWorkItem): void { + this.clearExpiry(workItem); + if (workItem.running || workItem.expired) { + return; + } + workItem.expired = true; + const deferred = this.results.get(workItem); + if (deferred !== undefined) { + this.results.delete(workItem); + deferred.reject(new QueueTaskExpiredError(workItem.expiresInMs ?? 0)); + } + } + + private expire(workItem: IWorkItem): void { + this.clearExpiry(workItem); + if (workItem.running || workItem.expired) { + return; + } + const index = this.items.indexOf(workItem); + if (index < 0) { + return; + } + this.items.splice(index, 1); + this.settleExpired(workItem); + } + public completed(workItem: IWorkItem, result?: R, error?: Error): void { + this.clearExpiry(workItem); const deferred = this.results.get(workItem); if (deferred !== undefined) { this.results.delete(workItem); if (error !== undefined) { deferred.reject(error); + } else { + deferred.resolve(result); } - deferred.resolve(result); } } public next(): IWorkItem | undefined { - return this.items.shift(); + let workItem = this.items.shift(); + while (workItem !== undefined) { + if (workItem.expired) { + workItem = this.items.shift(); + continue; + } + // Absolute-deadline recheck: never start an item past its deadline even if the timer hasn't fired. + if (workItem.expiresAt !== undefined && this.now() >= workItem.expiresAt) { + this.settleExpired(workItem); + workItem = this.items.shift(); + continue; + } + workItem.running = true; + this.clearExpiry(workItem); + return workItem; + } + return undefined; } public clear(): void { this.results.forEach((v: Deferred, k: IWorkItem, map: Map, Deferred>) => { + this.clearExpiry(k); v.reject(Error('Queue stopped processing')); map.delete(k); }); + this.items.length = 0; } } @@ -131,7 +210,7 @@ class WorkerPoolImpl implements WorkerPool { private readonly waitingWorkersUnblockQueue: { unblock(w: IWorkItem): void; stop(): void }[] = []; // A collection that manages the work items. - private readonly queue = new WorkQueue(); + private readonly queue: WorkQueue; // State of the pool manages via stop(), start() private stopProcessing = false; @@ -140,16 +219,19 @@ class WorkerPoolImpl implements WorkerPool { private readonly workerFunc: WorkFunc, private readonly numWorkers: number = 2, private readonly name: string = 'Worker', - ) {} + now?: QueueClock, + ) { + this.queue = new WorkQueue(now); + } - public addToQueue(item: T, position?: QueuePosition): Promise { + public addToQueue(item: T, position?: QueuePosition, expiresInMs?: number): Promise { if (this.stopProcessing) { throw Error('Queue is stopped'); } // This promise when resolved should return the processed result of the item // being added to the queue. - const deferred = this.queue.add(item, position); + const deferred = this.queue.add(item, position, expiresInMs); const worker = this.waitingWorkersUnblockQueue.shift(); if (worker) { @@ -160,9 +242,8 @@ class WorkerPoolImpl implements WorkerPool { // and give the worker the newly added item. worker.unblock(workItem); } else { - // Something is wrong, we should not be here. we just added an item to - // the queue. It should not be empty. - traceError('Work queue was empty immediately after adding item.'); + // next() dropped the just-added item as already expired; re-park the worker. + this.waitingWorkersUnblockQueue.unshift(worker); } } @@ -243,8 +324,9 @@ export function createRunningWorkerPool( workerFunc: WorkFunc, numWorkers?: number, name?: string, + now?: QueueClock, ): WorkerPool { - const pool = new WorkerPoolImpl(workerFunc, numWorkers, name); + const pool = new WorkerPoolImpl(workerFunc, numWorkers, name, now); pool.start(); return pool; } diff --git a/src/managers/common/nativePythonFinder.ts b/src/managers/common/nativePythonFinder.ts index a09b83e9e..80dcaedcf 100644 --- a/src/managers/common/nativePythonFinder.ts +++ b/src/managers/common/nativePythonFinder.ts @@ -1,6 +1,7 @@ import { ChildProcess } from 'child_process'; import * as fs from 'fs-extra'; import * as path from 'path'; +import { performance } from 'perf_hooks'; import { PassThrough } from 'stream'; import { CancellationTokenSource, Disposable, ExtensionContext, LogOutputChannel, Uri } from 'vscode'; import * as rpc from 'vscode-jsonrpc/node'; @@ -15,7 +16,7 @@ import { classifyError, isTimeoutErrorType } from '../../common/telemetry/errorC import { sendTelemetryEvent } from '../../common/telemetry/sender'; import { untildify, untildifyArray } from '../../common/utils/pathUtils'; import { isWindows } from '../../common/utils/platformUtils'; -import { createRunningWorkerPool, WorkerPool } from '../../common/utils/workerPool'; +import { createRunningWorkerPool, QueuePosition, WorkerPool } from '../../common/utils/workerPool'; import { getConfiguration, getWorkspaceFolders } from '../../common/workspace.apis'; import { getRefreshTelemetryMeasures, @@ -92,6 +93,74 @@ export class ConfigureRetryState { } } +export const MIN_STAGE_BUDGET_MS = 1_000; + +/** Worst-case wall-clock of a *successful* bounded refresh, so the cap never truncates a valid flow. */ +export function computeRefreshOperationBudgetMs(): number { + const maxRestartBackoffMs = RESTART_BACKOFF_BASE_MS * Math.pow(2, MAX_RESTART_ATTEMPTS - 1); + const firstFailingAttemptMs = MAX_CONFIGURE_TIMEOUT_MS + REFRESH_TIMEOUT_MS; + const additionalFailingAttemptMs = maxRestartBackoffMs + CONFIGURE_TIMEOUT_MS + REFRESH_TIMEOUT_MS; + const failingAttemptsMs = + MAX_REFRESH_RETRIES > 0 + ? firstFailingAttemptMs + (MAX_REFRESH_RETRIES - 1) * additionalFailingAttemptMs + : 0; + const succeedingAttemptMs = maxRestartBackoffMs + CONFIGURE_TIMEOUT_MS + REFRESH_TIMEOUT_MS + RESOLVE_TIMEOUT_MS; + return failingAttemptsMs + succeedingAttemptMs; +} + +export const REFRESH_OPERATION_BUDGET_MS = computeRefreshOperationBudgetMs(); + +export type MonotonicClock = () => number; + +const defaultMonotonicClock: MonotonicClock = () => performance.now(); + +/** Absolute, monotonic deadline captured at enqueue; stages clamp their timeouts to {@link remainingMs}. */ +export class Deadline { + private readonly deadlineAt: number; + + constructor( + budgetMs: number, + private readonly now: MonotonicClock = defaultMonotonicClock, + ) { + this.deadlineAt = this.now() + budgetMs; + } + + remainingMs(): number { + return this.deadlineAt - this.now(); + } + + isExhausted(floorMs: number = MIN_STAGE_BUDGET_MS): boolean { + return this.remainingMs() < floorMs; + } +} + +/** Rejects a bounded refresh (or one of its stages) once the operation budget is spent. */ +export class RefreshBudgetExceededError extends Error { + constructor( + public readonly stage: string, + remainingMs: number, + ) { + super(`Refresh operation budget exceeded at stage '${stage}' (remaining ${Math.round(remainingMs)}ms)`); + this.name = this.constructor.name; + } +} + +export function clampTimeoutToRemaining( + baseTimeoutMs: number, + deadline: Deadline | undefined, + stage: string, + floorMs: number = MIN_STAGE_BUDGET_MS, +): number { + if (deadline === undefined) { + return baseTimeoutMs; + } + const remaining = deadline.remainingMs(); + if (remaining < floorMs) { + throw new RefreshBudgetExceededError(stage, remaining); + } + return Math.min(baseTimeoutMs, remaining); +} + export type NativePythonToolsSource = 'envs_extension' | 'python_extension'; export async function getNativePythonToolsPath(): Promise { @@ -342,9 +411,27 @@ async function sendRequestWithTimeout( } } +export async function backoffThenCheckBudget( + waitMs: number, + deadline: Deadline | undefined, + sleep: (ms: number) => Promise = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), +): Promise { + if (waitMs > 0) { + await sleep(waitMs); + } + if (deadline?.isExhausted()) { + throw new RefreshBudgetExceededError('restart', deadline.remainingMs()); + } +} + +interface RefreshWorkItem { + options?: NativePythonEnvironmentKind | Uri[]; + deadline?: Deadline; +} + class NativePythonFinderImpl implements NativePythonFinder { private connection: rpc.MessageConnection; - private readonly pool: WorkerPool; + private readonly pool: WorkerPool; private cache: Map = new Map(); /** * Tracks in-flight hard refreshes by cache key so concurrent callers share a @@ -374,8 +461,8 @@ class NativePythonFinderImpl implements NativePythonFinder { private readonly cacheDirectory?: Uri, ) { this.connection = this.start(); - this.pool = createRunningWorkerPool( - async (options) => await this.doRefresh(options), + this.pool = createRunningWorkerPool( + async (work) => await this.doRefresh(work.options, work.deadline), 1, 'NativeRefresh-task', ); @@ -441,7 +528,7 @@ class NativePythonFinderImpl implements NativePythonFinder { * with exponential backoff up to MAX_RESTART_ATTEMPTS times. * @throws Error if the process cannot be started after all retry attempts */ - private async ensureProcessRunning(): Promise { + private async ensureProcessRunning(deadline?: Deadline): Promise { // Process is running fine if (!this.startFailed && !this.processExited) { return; @@ -462,27 +549,34 @@ class NativePythonFinderImpl implements NativePythonFinder { } // Attempt restart with exponential backoff - await this.restart(); + await this.restart(deadline); } /** * Kills the current PET process (if running) and starts a fresh one. * Implements exponential backoff between restart attempts. */ - private async restart(): Promise { + private async restart(deadline?: Deadline): Promise { + if (deadline?.isExhausted()) { + throw new RefreshBudgetExceededError('restart', deadline.remainingMs()); + } + this.isRestarting = true; this.restartAttempts++; const attempt = this.restartAttempts; const triggerReason = this.processExitReason ?? (this.startFailed ? 'start_failed' : 'unknown'); const backoffMs = RESTART_BACKOFF_BASE_MS * Math.pow(2, this.restartAttempts - 1); + const waitMs = deadline ? Math.min(backoffMs, Math.max(0, deadline.remainingMs())) : backoffMs; this.outputChannel.warn( `[pet] Restarting Python Environment Tools (attempt ${this.restartAttempts}/${MAX_RESTART_ATTEMPTS}, ` + - `waiting ${backoffMs}ms)`, + `waiting ${waitMs}ms)`, ); const sw = new StopWatch(); try { + await backoffThenCheckBudget(waitMs, deadline); + // Kill existing process if still running this.killProcess(); @@ -490,9 +584,6 @@ class NativePythonFinderImpl implements NativePythonFinder { this.startDisposables.forEach((d) => d.dispose()); this.startDisposables = []; - // Wait with exponential backoff before restarting - await new Promise((resolve) => setTimeout(resolve, backoffMs)); - // Reset state flags this.processExited = false; this.startFailed = false; @@ -517,6 +608,11 @@ class NativePythonFinderImpl implements NativePythonFinder { // Reset restart attempts on successful start (process didn't immediately fail) // We'll reset this only after a successful request completes } catch (ex) { + if (ex instanceof RefreshBudgetExceededError) { + this.restartAttempts--; + this.outputChannel.warn(`[pet] Restart aborted before spawn: ${ex.message}`); + throw ex; + } sendTelemetryEvent( EventNames.PET_PROCESS_RESTART, { duration: sw.elapsedTime, attempt }, @@ -596,11 +692,14 @@ class NativePythonFinderImpl implements NativePythonFinder { this.outputChannel.debug(`[Finder] Hard refresh for key: ${key}`); } + // One deadline captured at enqueue: the pool expires the queued item and the same deadline clamps every running stage. + const deadline = new Deadline(REFRESH_OPERATION_BUDGET_MS); + // .finally clears the in-flight slot on both success AND failure paths so // a rejected refresh does not poison the cache — the next call after a // failure starts a fresh attempt, matching today's behavior. const refreshPromise = this.pool - .addToQueue(options) + .addToQueue({ options, deadline }, QueuePosition.back, REFRESH_OPERATION_BUDGET_MS) .then((result) => { if (!result || !Array.isArray(result)) { this.outputChannel.warn(`[pet] Worker pool returned invalid result type: ${typeof result}`); @@ -843,20 +942,31 @@ class NativePythonFinderImpl implements NativePythonFinder { }; } - private async doRefresh(options?: NativePythonEnvironmentKind | Uri[]): Promise { + private async doRefresh( + options?: NativePythonEnvironmentKind | Uri[], + deadline?: Deadline, + ): Promise { let lastError: unknown; for (let attempt = 0; attempt <= MAX_REFRESH_RETRIES; attempt++) { try { - return await this.doRefreshAttempt(options, attempt); + return await this.doRefreshAttempt(options, attempt, deadline); } catch (ex) { lastError = ex; + if (ex instanceof RefreshBudgetExceededError) { + this.outputChannel.warn(`[pet] Refresh operation budget exhausted (${ex.message}), aborting`); + throw ex; + } + // Retry on timeout or connection errors (PET hung or crashed mid-request) const isRetryable = (ex instanceof RpcTimeoutError && ex.method !== 'configure') || ex instanceof rpc.ConnectionError; if (isRetryable) { if (attempt < MAX_REFRESH_RETRIES) { + if (deadline?.isExhausted()) { + throw new RefreshBudgetExceededError('refresh_retry', deadline.remainingMs()); + } const reason = ex instanceof rpc.ConnectionError ? 'crashed' : 'timed out'; this.outputChannel.warn( `[pet] Refresh ${reason} (attempt ${attempt + 1}/${MAX_REFRESH_RETRIES + 1}), restarting and retrying...`, @@ -874,7 +984,7 @@ class NativePythonFinderImpl implements NativePythonFinder { // Non-timeout errors or final timeout — check if server is fully exhausted if (this.isServerExhausted()) { this.outputChannel.warn('[pet] Server mode exhausted, falling back to JSON CLI for refresh'); - return this.refreshViaJsonCli(options); + return this.refreshViaJsonCli(options, deadline); } throw ex; } @@ -883,7 +993,7 @@ class NativePythonFinderImpl implements NativePythonFinder { // Should not reach here, but TypeScript needs this if (this.isServerExhausted()) { this.outputChannel.warn('[pet] Server mode exhausted, falling back to JSON CLI for refresh (final)'); - return this.refreshViaJsonCli(options); + return this.refreshViaJsonCli(options, deadline); } throw lastError; } @@ -891,8 +1001,9 @@ class NativePythonFinderImpl implements NativePythonFinder { private async doRefreshAttempt( options: NativePythonEnvironmentKind | Uri[] | undefined, attempt: number, + deadline?: Deadline, ): Promise { - await this.ensureProcessRunning(); + await this.ensureProcessRunning(deadline); const disposables: Disposable[] = []; const unresolved: Promise[] = []; const nativeInfo: NativeInfo[] = []; @@ -905,19 +1016,25 @@ class NativePythonFinderImpl implements NativePythonFinder { const configuration = await this.buildConfigurationOptions(); workspaceDirCount = configuration.workspaceDirectories.length; searchPathCount = configuration.environmentDirectories.length; - await this.configure(configuration); + await this.configure(configuration, deadline); const refreshOptions = this.getRefreshOptions(options); disposables.push( this.connection.onNotification('environment', (data: NativeEnvInfo) => { this.outputChannel.info(`Discovered env: ${data.executable || data.prefix}`); if (data.executable && (!data.version || !data.prefix)) { unresolvedCount++; + let resolveTimeout: number; + try { + resolveTimeout = clampTimeoutToRemaining(RESOLVE_TIMEOUT_MS, deadline, 'refresh_resolve'); + } catch { + return; + } unresolved.push( sendRequestWithTimeout( this.connection, 'resolve', { executable: data.executable }, - RESOLVE_TIMEOUT_MS, + resolveTimeout, ) .then((environment: NativeEnvInfo) => { this.outputChannel.info( @@ -943,11 +1060,12 @@ class NativePythonFinderImpl implements NativePythonFinder { } }), ); + const refreshTimeoutMs = clampTimeoutToRemaining(REFRESH_TIMEOUT_MS, deadline, 'refresh'); await sendRequestWithTimeout<{ duration: number }>( this.connection, 'refresh', refreshOptions, - REFRESH_TIMEOUT_MS, + refreshTimeoutMs, ); await Promise.all(unresolved); @@ -976,6 +1094,11 @@ class NativePythonFinderImpl implements NativePythonFinder { }, ); } catch (ex) { + // Budget errors bypass stage-timeout telemetry and retry-counter mutation; the RPC was already cancelled. + if (ex instanceof RefreshBudgetExceededError) { + this.outputChannel.warn(`[pet] Refresh attempt aborted by operation budget: ${ex.message}`); + throw ex; + } const errorType = classifyError(ex); sendTelemetryEvent( EventNames.PET_REFRESH, @@ -1022,7 +1145,7 @@ class NativePythonFinderImpl implements NativePythonFinder { * Configuration request, this must always be invoked before any other request. * Must be invoked when ever there are changes to any data related to the configuration details. */ - private async configure(options?: ConfigurationOptions) { + private async configure(options?: ConfigurationOptions, deadline?: Deadline) { const configuration = options ?? (await this.buildConfigurationOptions()); const workspaceDirCount = configuration.workspaceDirectories.length; const envDirCount = configuration.environmentDirectories.length; @@ -1037,8 +1160,7 @@ class NativePythonFinderImpl implements NativePythonFinder { return; } this.outputChannel.info('[pet] configure: Sending configuration update:', JSON.stringify(configuration)); - // Exponential backoff: 30s, 60s on retry. Capped at REFRESH_TIMEOUT_MS. - const timeoutMs = this.configureRetry.getTimeoutMs(); + const timeoutMs = clampTimeoutToRemaining(this.configureRetry.getTimeoutMs(), deadline, 'configure'); if (this.configureRetry.timeoutCount > 0) { this.outputChannel.info( `[pet] configure: Using extended timeout of ${timeoutMs}ms (retry ${this.configureRetry.timeoutCount})`, @@ -1057,6 +1179,11 @@ class NativePythonFinderImpl implements NativePythonFinder { { result: 'success' }, ); } catch (ex) { + if (ex instanceof RefreshBudgetExceededError) { + this.lastConfiguration = undefined; + this.outputChannel.warn(`[pet] Configure aborted by operation budget: ${ex.message}`); + throw ex; + } const errorType = classifyError(ex); sendTelemetryEvent( EventNames.PET_CONFIGURE, @@ -1205,7 +1332,10 @@ class NativePythonFinderImpl implements NativePythonFinder { * @param options Optional kind filter or URI search paths (same semantics as refresh()). * @returns NativeInfo[] containing managers and environments, same as server mode. */ - private async refreshViaJsonCli(options?: NativePythonEnvironmentKind | Uri[]): Promise { + private async refreshViaJsonCli( + options?: NativePythonEnvironmentKind | Uri[], + deadline?: Deadline, + ): Promise { const config = await this.buildConfigurationOptions(); // venvFolders must be included explicitly as search paths when options is Uri[], // mirroring getRefreshOptions() server-mode behaviour (searchPaths may override environmentDirectories). @@ -1215,15 +1345,20 @@ class NativePythonFinderImpl implements NativePythonFinder { this.outputChannel.info(`[pet] JSON CLI fallback refresh: ${this.toolPath} ${args.join(' ')}`); const stopWatch = new StopWatch(); + const findTimeout = clampTimeoutToRemaining(CLI_FALLBACK_TIMEOUT_MS, deadline, 'cli_find'); + let stdout: string; try { - stdout = await this.runPetCliProcess(args, CLI_FALLBACK_TIMEOUT_MS); + stdout = await this.runPetCliProcess(args, findTimeout); } catch (ex) { sendTelemetryEvent(EventNames.PET_JSON_CLI_FALLBACK, stopWatch.elapsedTime, { operation: 'refresh', result: 'error', }); this.outputChannel.error('[pet] JSON CLI fallback refresh failed:', ex); + if (deadline?.isExhausted()) { + throw new RefreshBudgetExceededError('cli_find', deadline.remainingMs()); + } throw ex; } @@ -1267,11 +1402,34 @@ class NativePythonFinderImpl implements NativePythonFinder { // Each resolveViaJsonCli() spawns a new OS process, unlike server mode where all resolve // calls share a single long-lived process — so unbounded parallelism would cause CPU/memory // pressure. Process in batches of CLI_RESOLVE_CONCURRENCY. + const retainRemainingUnresolved = (fromIndex: number): void => { + const remaining = toResolve.slice(fromIndex); + this.outputChannel.warn( + `[pet CLI] Refresh budget exhausted; retaining ${remaining.length} unresolved env(s) without enrichment`, + ); + for (const env of remaining) { + nativeInfo.push(env); + } + }; for (let i = 0; i < toResolve.length; i += CLI_RESOLVE_CONCURRENCY) { + if (deadline?.isExhausted()) { + retainRemainingUnresolved(i); + break; + } const batch = toResolve.slice(i, i + CLI_RESOLVE_CONCURRENCY); + let resolveTimeout: number; + try { + resolveTimeout = clampTimeoutToRemaining(CLI_FALLBACK_TIMEOUT_MS, deadline, 'cli_resolve'); + } catch (ex) { + if (ex instanceof RefreshBudgetExceededError) { + retainRemainingUnresolved(i); + break; + } + throw ex; + } await Promise.all( batch.map((env) => - this.resolveViaJsonCli(env.executable!) + this.resolveViaJsonCli(env.executable!, resolveTimeout) .then((resolved) => { this.outputChannel.info(`[pet CLI] Resolved env: ${resolved.executable}`); nativeInfo.push(resolved); @@ -1302,7 +1460,10 @@ class NativePythonFinderImpl implements NativePythonFinder { * @returns The resolved NativeEnvInfo. * @throws Error if PET cannot identify the environment or if the output cannot be parsed. */ - private async resolveViaJsonCli(executable: string): Promise { + private async resolveViaJsonCli( + executable: string, + timeoutMs: number = CLI_FALLBACK_TIMEOUT_MS, + ): Promise { const args = ['resolve', executable, '--json']; if (this.cacheDirectory) { args.push('--cache-directory', this.cacheDirectory.fsPath); @@ -1313,7 +1474,7 @@ class NativePythonFinderImpl implements NativePythonFinder { let stdout: string; try { - stdout = await this.runPetCliProcess(args, CLI_FALLBACK_TIMEOUT_MS); + stdout = await this.runPetCliProcess(args, timeoutMs); } catch (ex) { sendTelemetryEvent(EventNames.PET_JSON_CLI_FALLBACK, stopWatch.elapsedTime, { operation: 'resolve', diff --git a/src/test/common/telemetry/errorClassifier.unit.test.ts b/src/test/common/telemetry/errorClassifier.unit.test.ts index 9e429a85f..eb1b8cef5 100644 --- a/src/test/common/telemetry/errorClassifier.unit.test.ts +++ b/src/test/common/telemetry/errorClassifier.unit.test.ts @@ -3,7 +3,8 @@ import { CancellationError } from 'vscode'; import * as rpc from 'vscode-jsonrpc/node'; import { BaseError } from '../../../common/errors/types'; import { classifyError, isTimeoutErrorType } from '../../../common/telemetry/errorClassifier'; -import { RpcTimeoutError } from '../../../managers/common/nativePythonFinder'; +import { QueueTaskExpiredError } from '../../../common/utils/workerPool'; +import { RefreshBudgetExceededError, RpcTimeoutError } from '../../../managers/common/nativePythonFinder'; suite('Error Classifier', () => { suite('classifyError', () => { @@ -18,6 +19,19 @@ suite('Error Classifier', () => { assert.strictEqual(classifyError(new RpcTimeoutError('info', 2000)), 'rpc_timeout'); }); + test('should classify a QueueTaskExpiredError as a timeout (rpc_timeout)', () => { + const errorType = classifyError(new QueueTaskExpiredError(5_000)); + assert.strictEqual(errorType, 'rpc_timeout'); + assert.ok(isTimeoutErrorType(errorType), 'queue expiration should record as a timeout'); + }); + + test('should classify a RefreshBudgetExceededError as a timeout (rpc_timeout)', () => { + // 'restart' would match the process_crash pattern; the instanceof branch must win. + const errorType = classifyError(new RefreshBudgetExceededError('restart', 250)); + assert.strictEqual(errorType, 'rpc_timeout'); + assert.ok(isTimeoutErrorType(errorType), 'budget exhaustion should record as a timeout'); + }); + test('should classify non-Error values as unknown', () => { assert.strictEqual(classifyError('string error'), 'unknown'); assert.strictEqual(classifyError(42), 'unknown'); diff --git a/src/test/common/utils/workerPool.unit.test.ts b/src/test/common/utils/workerPool.unit.test.ts new file mode 100644 index 000000000..79eb85df4 --- /dev/null +++ b/src/test/common/utils/workerPool.unit.test.ts @@ -0,0 +1,396 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import assert from 'node:assert'; +import * as sinon from 'sinon'; +import * as logging from '../../../common/logging'; +import { createDeferred, Deferred } from '../../../common/utils/deferred'; +import { + createRunningWorkerPool, + QueuePosition, + QueueTaskExpiredError, + WorkerPool, +} from '../../../common/utils/workerPool'; + +suite('WorkerPool — pending-task expiration', () => { + let clock: sinon.SinonFakeTimers; + + setup(() => { + clock = sinon.useFakeTimers(); + sinon.stub(logging, 'traceError'); + }); + + teardown(() => { + clock.restore(); + sinon.restore(); + }); + + function makeBlockingPool(): { + pool: WorkerPool; + started: string[]; + blockerGate: Deferred; + } { + const started: string[] = []; + const blockerGate = createDeferred(); + const pool = createRunningWorkerPool( + async (item: string): Promise => { + started.push(item); + if (item === 'blocker') { + return blockerGate.promise; + } + return item; + }, + 1, + 'test-pool', + ); + return { pool, started, blockerGate }; + } + + test('queued behind never-resolving work expires and never runs', async () => { + const { pool, started, blockerGate } = makeBlockingPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + assert.deepStrictEqual(started, ['blocker'], 'worker should be busy on blocker'); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + const outcome = pExpire.then( + () => ({ ok: true as const }), + (e: unknown) => ({ ok: false as const, err: e }), + ); + + await clock.tickAsync(5_000); + + const result = await outcome; + assert.strictEqual(result.ok, false, 'queued item should have rejected'); + assert.ok( + !result.ok && result.err instanceof QueueTaskExpiredError, + 'should reject with QueueTaskExpiredError', + ); + assert.deepStrictEqual(started, ['blocker'], 'expired item must never execute'); + } finally { + void blockerGate; + pool.stop(); + } + }); + + test('dequeue clears timer — an immediately dequeued item resolves instead of expiring', async () => { + const pool = createRunningWorkerPool(async (i: string) => i, 1, 'test-pool'); + try { + const p = pool.addToQueue('quick', QueuePosition.back, 5_000); + await clock.tickAsync(10_000); + assert.strictEqual(await p, 'quick', 'dequeued item should resolve normally, not expire'); + } finally { + pool.stop(); + } + }); + + test('expiry/dequeue boundary — dequeue wins: item runs and a later expiry is a no-op (settles once)', async () => { + const { pool, started, blockerGate } = makeBlockingPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + let settleCount = 0; + pExpire.then( + () => (settleCount += 1), + () => (settleCount += 1), + ); + + await clock.tickAsync(2_000); + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + + assert.strictEqual(await pExpire, 'expireme', 'dequeued item should resolve with its result'); + assert.ok(started.includes('expireme'), 'item should have executed'); + + await clock.tickAsync(10_000); + assert.strictEqual(settleCount, 1, 'the stale expiry timer must not settle the item a second time'); + } finally { + pool.stop(); + } + }); + + test('expiry/dequeue boundary — expiry wins: item never runs and stays rejected (settles once)', async () => { + const { pool, started, blockerGate } = makeBlockingPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + let settleCount = 0; + let settledErr: unknown; + pExpire.then( + () => (settleCount += 1), + (e: unknown) => { + settleCount += 1; + settledErr = e; + }, + ); + + await clock.tickAsync(5_000); + assert.ok(settledErr instanceof QueueTaskExpiredError, 'should reject with QueueTaskExpiredError'); + + blockerGate.resolve('blocker'); + await clock.tickAsync(10_000); + + assert.ok(!started.includes('expireme'), 'an expired item must never execute, even after the worker frees up'); + assert.strictEqual(settleCount, 1, 'the item must settle exactly once'); + } finally { + pool.stop(); + } + }); + + test('stop clears timer — no stale expiry fires after stop, and the item settles once', async () => { + const { pool, blockerGate } = makeBlockingPool(); + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + let settleCount = 0; + let err: unknown; + pExpire.then( + () => (settleCount += 1), + (e: unknown) => { + settleCount += 1; + err = e; + }, + ); + + pool.stop(); + await clock.tickAsync(0); + + assert.strictEqual(settleCount, 1, 'stop should settle the queued item once'); + assert.ok(err instanceof Error, 'should reject with an Error'); + assert.ok(!(err instanceof QueueTaskExpiredError), 'stop must not surface an expiry error'); + + await clock.tickAsync(10_000); + assert.strictEqual(settleCount, 1, 'a cleared expiry timer must not fire after stop'); + void blockerGate; + }); + + test('later tasks still run after a queued task expired', async () => { + const { pool, started, blockerGate } = makeBlockingPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + pExpire.catch(() => undefined); + await clock.tickAsync(5_000); + + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + + const pLater = pool.addToQueue('later'); + assert.strictEqual(await pLater, 'later', 'the pool should keep processing new work after an expiry'); + assert.ok(started.includes('later'), 'later task should have executed'); + } finally { + pool.stop(); + } + }); + + test('omitting expiresInMs preserves the original unbounded queueing behavior', async () => { + const { pool, started, blockerGate } = makeBlockingPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + // No expiresInMs → the queued item must never expire. + const pQueued = pool.addToQueue('patient', QueuePosition.back); + let settled = false; + pQueued.then( + () => (settled = true), + () => (settled = true), + ); + + await clock.tickAsync(60 * 60 * 1000); + assert.strictEqual(settled, false, 'a task without expiresInMs must not expire while queued'); + assert.ok(!started.includes('patient'), 'still queued behind the blocker'); + + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + assert.strictEqual(await pQueued, 'patient', 'it should run once the worker frees up'); + } finally { + pool.stop(); + } + }); +}); + +/** + * Absolute-deadline tests: the pool reads an injected `now` clock, so advancing it past a deadline + * without firing sinon's faked timer reproduces an event-loop stall and proves the recheck in next(). + */ +suite('WorkerPool — absolute-deadline expiration', () => { + let clock: sinon.SinonFakeTimers; + + setup(() => { + clock = sinon.useFakeTimers(); + sinon.stub(logging, 'traceError'); + }); + + teardown(() => { + clock.restore(); + sinon.restore(); + }); + + function makeInjectedClockPool(): { + pool: WorkerPool; + started: string[]; + blockerGate: Deferred; + setNow: (ms: number) => void; + } { + const started: string[] = []; + const blockerGate = createDeferred(); + let nowMs = 0; + const pool = createRunningWorkerPool( + async (item: string): Promise => { + started.push(item); + if (item === 'blocker') { + return blockerGate.promise; + } + return item; + }, + 1, + 'test-pool', + () => nowMs, + ); + return { + pool, + started, + blockerGate, + setNow: (ms: number) => { + nowMs = ms; + }, + }; + } + + test('event-loop stall: absolute recheck expires a queued item even when its timer is delayed', async () => { + const { pool, started, blockerGate, setNow } = makeInjectedClockPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + assert.deepStrictEqual(started, ['blocker']); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + const outcome = pExpire.then( + () => ({ ok: true as const }), + (e: unknown) => ({ ok: false as const, err: e }), + ); + + // Injected clock jumps past the deadline, but sinon's timer never fires; freeing the worker forces the next() recheck. + setNow(6_000); + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + + const result = await outcome; + assert.strictEqual(result.ok, false, 'stalled-past-deadline item must be rejected, not run'); + assert.ok( + !result.ok && result.err instanceof QueueTaskExpiredError, + 'should reject with QueueTaskExpiredError', + ); + assert.ok(!started.includes('expireme'), 'expired item must never execute despite a delayed timer'); + + setNow(7_000); + const pLater = pool.addToQueue('later'); + assert.strictEqual(await pLater, 'later', 'the pool keeps processing after an absolute-deadline expiry'); + assert.ok(started.includes('later')); + } finally { + pool.stop(); + } + }); + + test('boundary: an item whose deadline exactly equals now expires (>=) and does not run', async () => { + const { pool, started, blockerGate, setNow } = makeInjectedClockPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); // expiresAt = 5000, injected clock + const outcome = pExpire.then( + () => ({ ok: true as const }), + (e: unknown) => ({ ok: false as const, err: e }), + ); + + setNow(5_000); // exactly at the deadline → recheck expires it (>=) + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + + const result = await outcome; + assert.strictEqual(result.ok, false, 'an item exactly at its deadline must expire (>= boundary)'); + assert.ok(!result.ok && result.err instanceof QueueTaskExpiredError); + assert.ok(!started.includes('expireme')); + } finally { + pool.stop(); + } + }); + + test('next() skips a stalled-expired item and continues to the next valid queued item', async () => { + const { pool, started, blockerGate, setNow } = makeInjectedClockPool(); + try { + const pBlocker = pool.addToQueue('blocker'); + pBlocker.catch(() => undefined); + await clock.tickAsync(0); + + const pExpire = pool.addToQueue('expireme', QueuePosition.back, 5_000); + const expireOutcome = pExpire.then( + () => ({ ok: true as const }), + (e: unknown) => ({ ok: false as const, err: e }), + ); + const pKeep = pool.addToQueue('keepme', QueuePosition.back); + + setNow(6_000); + blockerGate.resolve('blocker'); + await clock.tickAsync(0); + + const result = await expireOutcome; + assert.strictEqual(result.ok, false, 'the stalled item should be expired'); + assert.ok(!result.ok && result.err instanceof QueueTaskExpiredError); + assert.strictEqual(await pKeep, 'keepme', 'next() must continue to the next valid item after skipping an expired one'); + assert.ok(!started.includes('expireme'), 'expired item never ran'); + assert.ok(started.includes('keepme'), 'the following valid item ran'); + } finally { + pool.stop(); + } + }); + + test('enqueuing an already-expired item (non-positive expiresInMs) rejects it without stranding the parked worker', async () => { + const started: string[] = []; + const pool = createRunningWorkerPool( + async (i: string) => { + started.push(i); + return i; + }, + 1, + 'test-pool', + ); + try { + const pExpired = pool.addToQueue('expired-now', QueuePosition.back, 0); + const outcome = pExpired.then( + () => ({ ok: true as const }), + (e: unknown) => ({ ok: false as const, err: e }), + ); + await clock.tickAsync(0); + + const result = await outcome; + assert.strictEqual(result.ok, false, 'a non-positive expiry must reject immediately'); + assert.ok(!result.ok && result.err instanceof QueueTaskExpiredError); + assert.ok(!started.includes('expired-now'), 'the already-expired item never ran'); + + const pLater = pool.addToQueue('later'); + assert.strictEqual(await pLater, 'later', 'worker was re-parked and still processes new work'); + assert.ok(started.includes('later')); + } finally { + pool.stop(); + } + }); +}); diff --git a/src/test/managers/common/nativePythonFinder.budget.unit.test.ts b/src/test/managers/common/nativePythonFinder.budget.unit.test.ts new file mode 100644 index 000000000..f883100a4 --- /dev/null +++ b/src/test/managers/common/nativePythonFinder.budget.unit.test.ts @@ -0,0 +1,187 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import assert from 'node:assert'; +import { + backoffThenCheckBudget, + clampTimeoutToRemaining, + computeRefreshOperationBudgetMs, + Deadline, + MIN_STAGE_BUDGET_MS, + MonotonicClock, + REFRESH_OPERATION_BUDGET_MS, + RefreshBudgetExceededError, +} from '../../../managers/common/nativePythonFinder'; + +function makeClock(start = 0): { clock: MonotonicClock; advance(ms: number): void; set(ms: number): void } { + let t = start; + return { + clock: () => t, + advance: (ms: number) => { + t += ms; + }, + set: (ms: number) => { + t = ms; + }, + }; +} + +suite('Bounded refresh latency — operation budget', () => { + const CONFIGURE_TIMEOUT_MS = 30_000; + const MAX_CONFIGURE_TIMEOUT_MS = 60_000; + const REFRESH_TIMEOUT_MS = 30_000; + const RESOLVE_TIMEOUT_MS = 30_000; + const RESTART_BACKOFF_BASE_MS = 1_000; + const MAX_RESTART_ATTEMPTS = 3; + const maxRestartBackoffMs = RESTART_BACKOFF_BASE_MS * Math.pow(2, MAX_RESTART_ATTEMPTS - 1); // 4s + + test('computeRefreshOperationBudgetMs equals the worst-case successful server path (184s)', () => { + const failingAttemptMs = MAX_CONFIGURE_TIMEOUT_MS + REFRESH_TIMEOUT_MS; // 60 + 30 = 90s + const succeedingAttemptMs = + maxRestartBackoffMs + CONFIGURE_TIMEOUT_MS + REFRESH_TIMEOUT_MS + RESOLVE_TIMEOUT_MS; // 4 + 30 + 30 + 30 = 94s + const expected = failingAttemptMs + succeedingAttemptMs; // 184s + + assert.strictEqual(expected, 184_000, 'sanity: hand arithmetic should be 184000ms'); + assert.strictEqual(computeRefreshOperationBudgetMs(), 184_000); + assert.strictEqual(REFRESH_OPERATION_BUDGET_MS, 184_000); + }); + + test('MIN_STAGE_BUDGET_MS floor is 1s', () => { + assert.strictEqual(MIN_STAGE_BUDGET_MS, 1_000); + }); +}); + +suite('Bounded refresh latency — backoffThenCheckBudget (restart recheck)', () => { + test('resolves without throwing when no deadline is supplied (non-refresh restart path)', async () => { + let slept = 0; + await backoffThenCheckBudget(1_000, undefined, async (ms) => { + slept += ms; + }); + assert.strictEqual(slept, 1_000, 'the backoff wait still happens'); + }); + + test('rejects with RefreshBudgetExceededError when the budget expires during the wait', async () => { + const { clock, advance } = makeClock(); + const dl = new Deadline(4_000, clock); + await assert.rejects( + backoffThenCheckBudget(4_000, dl, async (ms) => { + advance(ms); // 4s elapses → remaining 0 < floor → exhausted + }), + RefreshBudgetExceededError, + ); + }); + + test('resolves when budget remains after the (clamped) backoff', async () => { + const { clock, advance } = makeClock(); + const dl = new Deadline(100_000, clock); + await backoffThenCheckBudget(4_000, dl, async (ms) => { + advance(ms); + }); + assert.ok(dl.remainingMs() > MIN_STAGE_BUDGET_MS); + }); +}); + +suite('Bounded refresh latency — Deadline', () => { + test('remainingMs counts down as the monotonic clock advances', () => { + const { clock, advance } = makeClock(); + const dl = new Deadline(10_000, clock); + assert.strictEqual(dl.remainingMs(), 10_000); + + advance(4_000); + assert.strictEqual(dl.remainingMs(), 6_000); + + advance(6_000); + assert.strictEqual(dl.remainingMs(), 0); + + advance(1_000); // past the deadline + assert.strictEqual(dl.remainingMs(), -1_000); + }); + + test('isExhausted uses the default floor (MIN_STAGE_BUDGET_MS) when none is given', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(10_000, clock); + + set(8_999); // remaining 1001 > floor 1000 + assert.strictEqual(dl.isExhausted(), false); + + set(9_000); // remaining 1000 == floor → NOT exhausted (strictly-less check) + assert.strictEqual(dl.isExhausted(), false); + + set(9_001); // remaining 999 < floor + assert.strictEqual(dl.isExhausted(), true); + }); + + test('isExhausted honors a custom floor', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(10_000, clock); + + set(9_500); // remaining 500 + assert.strictEqual(dl.isExhausted(100), false, '500 remaining is above a 100ms floor'); + assert.strictEqual(dl.isExhausted(1_000), true, '500 remaining is below a 1000ms floor'); + }); +}); + +suite('Bounded refresh latency — clampTimeoutToRemaining', () => { + test('returns the base timeout unchanged when no deadline is supplied (non-refresh callers)', () => { + assert.strictEqual(clampTimeoutToRemaining(30_000, undefined, 'configure'), 30_000); + assert.strictEqual(clampTimeoutToRemaining(120_000, undefined, 'cli_find'), 120_000); + }); + + test('returns the base timeout when it is smaller than the remaining budget', () => { + const { clock } = makeClock(); + const dl = new Deadline(100_000, clock); + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'refresh'), 30_000); + }); + + test('clamps down to the remaining budget when less than the base timeout remains', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(100_000, clock); + set(80_000); // remaining 20s + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'refresh'), 20_000); + }); + + test('throws RefreshBudgetExceededError when the remaining budget is below the floor', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(100_000, clock); + set(99_500); // remaining 500 < 1000 floor + assert.throws(() => clampTimeoutToRemaining(30_000, dl, 'resolve'), RefreshBudgetExceededError); + }); + + test('honors a custom floor', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(100_000, clock); + set(99_500); // remaining 500 + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'resolve', 100), 500); + assert.throws(() => clampTimeoutToRemaining(30_000, dl, 'resolve', 1_000), RefreshBudgetExceededError); + }); + + test('propagation across configure → refresh → resolve shrinks the clamp and finally fails fast', () => { + const { clock, set } = makeClock(); + const dl = new Deadline(REFRESH_OPERATION_BUDGET_MS, clock); // 184s + + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'configure'), 30_000); + set(30_000); + + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'refresh'), 30_000); + set(60_000); + + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'refresh_resolve'), 30_000); + + set(REFRESH_OPERATION_BUDGET_MS - 20_000); + assert.strictEqual(clampTimeoutToRemaining(30_000, dl, 'refresh'), 20_000); + + set(REFRESH_OPERATION_BUDGET_MS - 100); + assert.throws(() => clampTimeoutToRemaining(30_000, dl, 'refresh'), RefreshBudgetExceededError); + }); +}); + +suite('Bounded refresh latency — RefreshBudgetExceededError', () => { + test('carries the stage and has a stable name', () => { + const err = new RefreshBudgetExceededError('restart', 250); + assert.strictEqual(err.name, 'RefreshBudgetExceededError'); + assert.strictEqual(err.stage, 'restart'); + assert.ok(err instanceof Error); + assert.ok(err instanceof RefreshBudgetExceededError); + assert.strictEqual(err.message, "Refresh operation budget exceeded at stage 'restart' (remaining 250ms)"); + }); +});