Skip to content

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

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

Open
StellaHuang95 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

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

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)
);
}

@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

isPetConnectionLostError now controls PET retry behavior but lives in a telemetry module, deepening the manager-to-telemetry dependency. Move this predicate to a PET/RPC domain error module and import it from both telemetry and NativePythonFinderImpl.

[verified]


/**
* 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';
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';
}

@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

The new resolve recovery path treats PendingResponseRejected as a recoverable connection loss, but the new tests cover that behavior only through refresh. Add a resolve-specific regression test so its restart behavior does not drift from refresh.

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();
writable.end();
};

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');
}

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) {
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