mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-09 01:45:00 +00:00
fix(client): finish closing when subscription releases are in flight
The host runtime releases a client's subscriptions and closes it without waiting. close() drops the transport and rejects outstanding requests before it marks the client disposed, so each in-flight release failed while the client still reported connected. The failure handler then published a false transport close, and the rejection propagated through owned.close(), aborting close() before the rest of its cleanup and the disposed state. A release that fails after its transport is gone has nothing to reset.
This commit is contained in:
@@ -5225,6 +5225,65 @@ test("logs configured runtime generation in connection transition events", async
|
||||
}
|
||||
});
|
||||
|
||||
test("closing a client while subscription releases are in flight reports only the disposal", async () => {
|
||||
const mock = createMockTransport();
|
||||
let transportsCreated = 0;
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger: createMockLogger(),
|
||||
reconnect: { enabled: true, baseDelayMs: 1, maxDelayMs: 1 },
|
||||
transportFactory: () => {
|
||||
transportsCreated += 1;
|
||||
return mock.transport;
|
||||
},
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const subscriptions = ["/tmp/one", "/tmp/two"].map((cwd) =>
|
||||
client.observeCheckoutDiff(cwd, { mode: "uncommitted" }),
|
||||
);
|
||||
await vi.waitFor(() => expect(mock.sent).toHaveLength(2));
|
||||
for (const [index, frame] of [...mock.sent].entries()) {
|
||||
const request = parseSentFrame(frame);
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "subscribe_checkout_diff_response",
|
||||
payload: {
|
||||
subscriptionId: `checkout-sub-${index}`,
|
||||
cwd: request.cwd,
|
||||
files: [],
|
||||
error: null,
|
||||
requestId: request.requestId,
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
await Promise.all(subscriptions.map((subscription) => subscription.ready));
|
||||
|
||||
// The host runtime releases its handlers and closes the previous client without waiting.
|
||||
const releases = subscriptions.map((subscription) =>
|
||||
subscription.release().catch(() => undefined),
|
||||
);
|
||||
await vi.waitFor(() => expect(mock.sent).toHaveLength(4));
|
||||
|
||||
const statuses: string[] = [];
|
||||
client.subscribeConnectionStatus((state) => statuses.push(state.status));
|
||||
statuses.length = 0;
|
||||
|
||||
await client.close();
|
||||
await Promise.all(releases);
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
|
||||
expect(statuses).toEqual(["disposed"]);
|
||||
expect(client.lastError).toBeNull();
|
||||
expect(transportsCreated).toBe(1);
|
||||
});
|
||||
|
||||
test("subscribes to checkout diff updates via RPC handshake", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -1400,12 +1400,16 @@ export class DaemonClient {
|
||||
),
|
||||
release: async (subscriptionId) => {
|
||||
if (!this.isConnected) return;
|
||||
const transport = this.transport;
|
||||
try {
|
||||
await this.sendCorrelatedSessionRequest({
|
||||
message: { type: "subscription.release.request", subscriptionId },
|
||||
responseType: "subscription.release.response",
|
||||
});
|
||||
} catch (error) {
|
||||
// close() rejects in-flight requests after dropping the transport. That failure says
|
||||
// nothing about a live connection, and rethrowing it would abort close() midway.
|
||||
if (this.transport !== transport) return;
|
||||
this.disposeTransport(1001, "Subscription release failed");
|
||||
this.scheduleReconnect({ reason: "Subscription release failed" });
|
||||
throw error;
|
||||
|
||||
Reference in New Issue
Block a user