Skip to content

fix: promptly reject in-flight PET RPC requests on process exit/error #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
StellaHuang95 wants to merge 1 commit into from
Closed

fix: promptly reject in-flight PET RPC requests on process exit/error #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions src/common/telemetry/errorClassifier.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,18 @@ export function isTimeoutErrorType(errorType: DiscoveryErrorType): boolean {
(errorType.startsWith('rpc_') && errorType.endsWith('_timeout'));
}

/**
* True when `ex` is a lost PET JSON-RPC connection — an {@link rpc.ConnectionError} or a
* {@link rpc.ResponseError} with {@link rpc.ErrorCodes.PendingResponseRejected}. Pure classifier:
* it cannot tell an intentional disposal from a crash, so callers must gate on lifecycle state.
*/
export function isPetConnectionLostError(ex: unknown): boolean {
return (
ex instanceof rpc.ConnectionError ||
(ex instanceof rpc.ResponseError && ex.code === rpc.ErrorCodes.PendingResponseRejected)
);
}

/**
* Classifies an error into a telemetry-safe category for the `errorType` property.
* Does NOT include raw error messages — only the category.
Expand All @@ -49,8 +61,7 @@ export function classifyError(ex: unknown): DiscoveryErrorType {
}
}

// JSON-RPC connection errors (e.g., PET process died mid-request, connection closed/disposed)
if (ex instanceof rpc.ConnectionError) {
if (isPetConnectionLostError(ex)) {
return 'connection_error';
}

Expand Down
110 changes: 69 additions & 41 deletions src/managers/common/nativePythonFinder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import { getExtension } from '../../common/extension.apis';
import { traceError, traceVerbose, traceWarn } from '../../common/logging';
import { StopWatch } from '../../common/stopWatch';
import { EventNames } from '../../common/telemetry/constants';
import { classifyError, isTimeoutErrorType } from '../../common/telemetry/errorClassifier';
import { classifyError, isPetConnectionLostError, isTimeoutErrorType } from '../../common/telemetry/errorClassifier';

@StellaHuang95 StellaHuang95 Aug 22, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

Recovery policy now depends on a telemetry module that imports RpcTimeoutError from this manager, reinforcing a circular dependency. Move the PET error types and predicates into a dependency-neutral common module consumed by both layers.

@StellaHuang95 StellaHuang95 Aug 23, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

This creates a runtime dependency cycle: the finder imports recovery policy from errorClassifier, while errorClassifier imports RpcTimeoutError from the finder. Please move the PET RPC error types and predicates into a dependency-neutral common module so module initialization and future maintenance do not depend on circular-import behavior.

import { sendTelemetryEvent } from '../../common/telemetry/sender';
import { untildify, untildifyArray } from '../../common/utils/pathUtils';
import { isWindows } from '../../common/utils/platformUtils';
Expand Down Expand Up @@ -342,7 +342,10 @@ async function sendRequestWithTimeout<T>(
}
}

class NativePythonFinderImpl implements NativePythonFinder {
/**
* @internal Concrete {@link NativePythonFinder}, exported only as a test seam — not public API.
*/
export class NativePythonFinderImpl implements NativePythonFinder {
private connection: rpc.MessageConnection;
private readonly pool: WorkerPool<NativePythonEnvironmentKind | Uri[] | undefined, NativeInfo[]>;
private cache: Map<string, NativeInfo[]> = new Map();
Expand All @@ -357,6 +360,7 @@ class NativePythonFinderImpl implements NativePythonFinder {
private startFailed: boolean = false;
private restartAttempts: number = 0;
private isRestarting: boolean = false;
private disposed: boolean = false;
private processExitReason: string | undefined = undefined;
private readonly configureRetry = new ConfigureRetryState();
/**
Expand Down Expand Up @@ -403,15 +407,16 @@ class NativePythonFinderImpl implements NativePythonFinder {
});
return environment;
} catch (ex) {
// On resolve timeout or connection error (not configure — configure handles its own timeout),
// kill the hung process so next request triggers restart
if ((ex instanceof RpcTimeoutError && ex.method !== 'configure') || ex instanceof rpc.ConnectionError) {
const reason = ex instanceof rpc.ConnectionError ? 'crashed' : 'timed out';
if (
(ex instanceof RpcTimeoutError && ex.method !== 'configure') ||
this.isRecoverableConnectionLoss(ex)
) {
const reason = ex instanceof RpcTimeoutError ? 'timed out' : 'crashed';
this.outputChannel.warn(`[pet] Resolve request ${reason}, killing process for restart`);
this.killProcess();
this.processExited = true;
this.processExitReason =
ex instanceof rpc.ConnectionError ? 'rpc_connection_error' : 'rpc_resolve_timeout';
ex instanceof RpcTimeoutError ? 'rpc_resolve_timeout' : 'rpc_connection_error';
}
throw ex;
}
Expand All @@ -436,6 +441,10 @@ class NativePythonFinderImpl implements NativePythonFinder {
}
}

private isRecoverableConnectionLoss(ex: unknown): boolean {
return !this.disposed && !this.isRestarting && isPetConnectionLostError(ex);
}

/**
* Ensures the PET process is running. If it has exited or failed, attempts to restart
* with exponential backoff up to MAX_RESTART_ATTEMPTS times.
Expand Down Expand Up @@ -638,6 +647,7 @@ class NativePythonFinderImpl implements NativePythonFinder {
}

public dispose() {
this.disposed = true;
this.pool.stop();
this.startDisposables.forEach((d) => d.dispose());
this.connection.dispose();
Expand Down Expand Up @@ -673,42 +683,62 @@ class NativePythonFinderImpl implements NativePythonFinder {
const readable = new PassThrough();
const writable = new PassThrough();

// Owned by THIS child, so a dead child closes only its own resources after a later restart().
const localDisposables: Disposable[] = [];
this.startDisposables = localDisposables;

let streamsEnded = false;
let childStdout: NodeJS.ReadableStream | undefined;
const endStreams = () => {
if (streamsEnded) {
return;
}
streamsEnded = true;
// Unpipe before ending so buffered stdout can't raise a write-after-end on the ended stream.
childStdout?.unpipe(readable);
writable.unpipe();
readable.end();

@StellaHuang95 StellaHuang95 Aug 23, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

Could unpiping stdout immediately on exit discard a complete JSON-RPC response that PET wrote before exiting but which has not yet been delivered through the pipe? Please add a deterministic pre-exit-response test and preserve/drain that response if the scenario is possible.

writable.end();

@StellaHuang95 StellaHuang95 Aug 22, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

Unpiping stdout on exit can prevent a complete response already buffered in the child stream from reaching readable, converting a successful request into PendingResponseRejected. Drain stdout through its close event, with a bounded fallback, before ending the reader; add coverage for delayed delivery of a response written before exit.

};

try {
this.proc = spawnProcess(this.toolPath, ['server'], { env: process.env, stdio: 'pipe' });
const proc = spawnProcess(this.toolPath, ['server'], { env: process.env, stdio: 'pipe' });
this.proc = proc;

if (!this.proc.stdout || !this.proc.stderr || !this.proc.stdin) {
if (!proc.stdout || !proc.stderr || !proc.stdin) {
throw new Error('Failed to create stdio streams for PET process');

@StellaHuang95 StellaHuang95 Aug 22, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

If spawning succeeds but a required stdio stream is absent, this throws before registering the child cleanup disposable, leaving a live child without teardown ownership. Register cleanup immediately after spawning or explicitly terminate the child on this failure path, and cover it with a live fake child missing stdio.

}

this.proc.stdout.pipe(readable, { end: false });
this.proc.stderr.on('data', (data) => this.outputChannel.error(`[pet] ${data.toString()}`));
writable.pipe(this.proc.stdin, { end: false });
childStdout = proc.stdout;
proc.stdout.pipe(readable, { end: false });
proc.stderr.on('data', (data) => this.outputChannel.error(`[pet] ${data.toString()}`));
writable.pipe(proc.stdin, { end: false });

// Handle process exit - mark as exited so pending requests fail fast
this.proc.on('exit', (code, signal) => {
this.processExited = true;
// Preserve a more-specific reason (e.g. rpc_*) if one was already recorded before the kill.
if (this.processExitReason === undefined) {
this.processExitReason = `process_exit:${code ?? 'null'}:${signal ?? 'none'}`;
const handleChildTermination = (reason: string) => {
endStreams();
if (this.proc === proc) {
this.processExited = true;
if (this.processExitReason === undefined) {
this.processExitReason = reason;
}
}
};

proc.on('exit', (code, signal) => {
handleChildTermination(`process_exit:${code ?? 'null'}:${signal ?? 'none'}`);
if (code !== 0) {

@StellaHuang95 StellaHuang95 Aug 23, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning · Non-blocking recommendation

Unpiping stdout immediately on process termination can discard a complete RPC response that was written before exit but has not yet reached the pipe destination. Please add coverage for that ordering and clarify whether the response should be drained or deliberately rejected.

this.outputChannel.error(
`[pet] Python Environment Tools exited unexpectedly with code ${code}, signal ${signal}`,
);
}
});

// Handle process errors (e.g., ENOENT if executable not found)
this.proc.on('error', (err) => {
this.processExited = true;
if (this.processExitReason === undefined) {
this.processExitReason = 'process_error';
}
proc.on('error', (err) => {
handleChildTermination('process_error');
this.outputChannel.error('[pet] Process error:', err);
});

const proc = this.proc;
this.startDisposables.push({
localDisposables.push({
dispose: () => {
try {
if (proc.exitCode === null) {
Expand Down Expand Up @@ -742,12 +772,9 @@ class NativePythonFinderImpl implements NativePythonFinder {
new rpc.StreamMessageReader(readable),
new rpc.StreamMessageWriter(writable),
);
this.startDisposables.push(
localDisposables.push(
connection,
new Disposable(() => {
readable.end();
writable.end();
}),
new Disposable(() => endStreams()),
connection.onError((ex) => {
this.outputChannel.error('[pet] Connection Error:', ex);
}),
Expand All @@ -772,7 +799,7 @@ class NativePythonFinderImpl implements NativePythonFinder {
}),
connection.onNotification('telemetry', (data) => this.outputChannel.info('[pet] Telemetry: ', data)),
connection.onClose(() => {
this.startDisposables.forEach((d) => d.dispose());
localDisposables.forEach((d) => d.dispose());
}),
);

Expand Down Expand Up @@ -852,20 +879,20 @@ class NativePythonFinderImpl implements NativePythonFinder {
} catch (ex) {
lastError = 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;
(ex instanceof RpcTimeoutError && ex.method !== 'configure') ||
this.isRecoverableConnectionLoss(ex);
if (isRetryable) {
if (attempt < MAX_REFRESH_RETRIES) {
const reason = ex instanceof rpc.ConnectionError ? 'crashed' : 'timed out';
const reason = ex instanceof RpcTimeoutError ? 'timed out' : 'crashed';
this.outputChannel.warn(
`[pet] Refresh ${reason} (attempt ${attempt + 1}/${MAX_REFRESH_RETRIES + 1}), restarting and retrying...`,
);
// Kill and restart for retry
this.killProcess();
this.processExited = true;
this.processExitReason =
ex instanceof rpc.ConnectionError ? 'rpc_connection_error' : 'rpc_refresh_timeout';
ex instanceof RpcTimeoutError ? 'rpc_refresh_timeout' : 'rpc_connection_error';
continue;
}
// Final attempt failed
Expand Down Expand Up @@ -997,15 +1024,16 @@ class NativePythonFinderImpl implements NativePythonFinder {
},
ex instanceof Error ? ex : undefined,
);
// On refresh timeout or connection error (not configure — configure handles its own timeout),
// kill the hung process so next request triggers restart
if ((ex instanceof RpcTimeoutError && ex.method !== 'configure') || ex instanceof rpc.ConnectionError) {
const reason = ex instanceof rpc.ConnectionError ? 'crashed' : 'timed out';
if (
(ex instanceof RpcTimeoutError && ex.method !== 'configure') ||
this.isRecoverableConnectionLoss(ex)
) {
const reason = ex instanceof RpcTimeoutError ? 'timed out' : 'crashed';
this.outputChannel.warn(`[pet] PET process ${reason}, killing for restart`);
this.killProcess();
this.processExited = true;
this.processExitReason =
ex instanceof rpc.ConnectionError ? 'rpc_connection_error' : 'rpc_refresh_timeout';
ex instanceof RpcTimeoutError ? 'rpc_refresh_timeout' : 'rpc_connection_error';
}
this.outputChannel.error('[pet] Error refreshing', ex);
throw ex;
Expand Down
37 changes: 36 additions & 1 deletion src/test/common/telemetry/errorClassifier.unit.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ import assert from 'node:assert';
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 { classifyError, isPetConnectionLostError, isTimeoutErrorType } from '../../../common/telemetry/errorClassifier';
import { RpcTimeoutError } from '../../../managers/common/nativePythonFinder';

suite('Error Classifier', () => {
Expand Down Expand Up @@ -86,6 +86,13 @@ suite('Error Classifier', () => {
assert.strictEqual(classifyError(new rpc.ResponseError(-32601, 'Method not found')), 'rpc_error');
});

test('should classify PendingResponseRejected ResponseError as connection_error', () => {
assert.strictEqual(
classifyError(new rpc.ResponseError(rpc.ErrorCodes.PendingResponseRejected, 'Pending response rejected')),
'connection_error',
);
});

test('should classify BaseError subclasses as already_registered', () => {
// Using a concrete subclass to test (BaseError is abstract)
class TestRegisteredError extends BaseError {
Expand Down Expand Up @@ -130,6 +137,34 @@ suite('Error Classifier', () => {
});
});

suite('isPetConnectionLostError', () => {
test('recognizes JSON-RPC ConnectionError (transport failure)', () => {
assert.strictEqual(
isPetConnectionLostError(new rpc.ConnectionError(rpc.ConnectionErrors.Closed, 'closed')),
true,
);
assert.strictEqual(
isPetConnectionLostError(new rpc.ConnectionError(rpc.ConnectionErrors.Disposed, 'disposed')),
true,
);
});

test('recognizes PendingResponseRejected ResponseError (connection disposed mid-request)', () => {
assert.strictEqual(
isPetConnectionLostError(new rpc.ResponseError(rpc.ErrorCodes.PendingResponseRejected, 'rejected')),
true,
);
});

test('does NOT match other ResponseError codes or unrelated errors', () => {
assert.strictEqual(isPetConnectionLostError(new rpc.ResponseError(-32600, 'Invalid request')), false);
assert.strictEqual(isPetConnectionLostError(new RpcTimeoutError('refresh', 30000)), false);
assert.strictEqual(isPetConnectionLostError(new Error('boom')), false);
assert.strictEqual(isPetConnectionLostError('nope'), false);
assert.strictEqual(isPetConnectionLostError(undefined), false);
});
});

suite('isTimeoutErrorType', () => {
test('recognizes spawn and JSON-RPC timeout categories', () => {
assert.strictEqual(isTimeoutErrorType('spawn_timeout'), true);
Expand Down
Loading
Loading