Skip to content

Commit 5cebd3b

Browse files
t3dotggclaude
t3dotgg
and
claude
authored
fix(clients): dropped connections say why in the client trace (#16200)
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
1 parent 24238fe commit 5cebd3b

7 files changed

Lines changed: 235 additions & 174 deletions

File tree

‎packages/client-runtime/src/connection/supervisor.ts‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -802,7 +802,30 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
802802
resetRetry: connectedExit.value,
803803
} satisfies AttemptOutcome;
804804
}
805-
return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS);
805+
const outcome = failureFromExit(
806+
target,
807+
connectedExit,
808+
true,
809+
connectedForMs >= BACKOFF_RESET_AFTER_MS,
810+
);
811+
if (outcome._tag === "Failure") {
812+
// A live session ending is otherwise invisible in the client trace, so
813+
// record why, and how long it lasted, as its own root span.
814+
yield* Effect.void.pipe(
815+
Effect.withSpan("EnvironmentSupervisor.connectionLost", {
816+
root: true,
817+
attributes: {
818+
"environment.id": target.environmentId,
819+
"environment.label": target.label,
820+
"environment.target.kind": target._tag,
821+
"connection.connected_ms": connectedForMs,
822+
"connection.failure.reason": outcome.failure.error.reason,
823+
"connection.failure.detail": outcome.failure.error.detail,
824+
},
825+
}),
826+
);
827+
}
828+
return outcome;
806829
}, Effect.ensuring(clearLease));
807830

808831
const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) {

‎packages/client-runtime/src/errors/transport.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ describe("isTransportConnectionErrorMessage", () => {
2222

2323
it("recognizes connection errors emitted by the Effect RPC session", () => {
2424
expect(isTransportConnectionErrorMessage("Test environment disconnected.")).toBe(true);
25+
expect(isTransportConnectionErrorMessage("Test environment stopped responding.")).toBe(true);
2526
expect(
2627
isTransportConnectionErrorMessage(
2728
"Test environment could not establish a WebSocket connection.",
@@ -34,6 +35,7 @@ describe("isTransportConnectionErrorMessage", () => {
3435
it("recognizes relay connection errors that carry the network hint", () => {
3536
for (const sentence of [
3637
"Relay environment disconnected.",
38+
"Relay environment stopped responding.",
3739
"Relay environment could not establish a WebSocket connection.",
3840
]) {
3941
expect(isTransportConnectionErrorMessage(`${sentence} ${NETWORK_BLOCKING_HINT}`)).toBe(true);

‎packages/client-runtime/src/errors/transport.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ const TRANSPORT_ERROR_PATTERNS = [
1010
// The RPC session appends the network hint for relay connections. Any other
1111
// trailing text means a different error that the user should still see.
1212
new RegExp(
13-
`\\b(?:is not connected|disconnected|could not establish a WebSocket connection)\\.(?: ${escapeRegExp(NETWORK_BLOCKING_HINT)})?$`,
13+
`\\b(?:is not connected|disconnected|stopped responding|could not establish a WebSocket connection)\\.(?: ${escapeRegExp(NETWORK_BLOCKING_HINT)})?$`,
1414
"i",
1515
),
1616
/\bClientProtocolError\b/i,

‎packages/client-runtime/src/rpc/session.test.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1108,7 +1108,10 @@ describe("RpcSessionFactory", () => {
11081108
yield* TestClock.adjust("5 seconds");
11091109
const error = yield* Fiber.join(closedFiber);
11101110
expect(error).toBeInstanceOf(ConnectionTransientError);
1111-
expect(error).toMatchObject({ reason: "transport" });
1111+
expect(error).toMatchObject({
1112+
reason: "transport",
1113+
detail: "Test environment stopped responding.",
1114+
});
11121115
}).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
11131116
);
11141117

‎packages/client-runtime/src/rpc/session.ts‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -170,18 +170,24 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
170170

171171
const connected = yield* Deferred.make<void>();
172172
const disconnected = yield* Deferred.make<never, ConnectionTransientError>();
173+
// Set when the socket closes because pongs stopped, so the failure says so
174+
// instead of looking like the server closed the connection.
175+
const pingTimedOut = yield* Ref.make(false);
173176
const hooks = RpcClient.ConnectionHooks.of({
174177
onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid),
175-
onDisconnect: Deferred.isDone(connected).pipe(
176-
Effect.flatMap((wasConnected) =>
178+
onPingTimeout: Ref.set(pingTimedOut, true),
179+
onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe(
180+
Effect.flatMap(([wasConnected, timedOut]) =>
177181
Deferred.fail(
178182
disconnected,
179183
new ConnectionTransientErrorClass({
180184
reason: "transport",
181185
detail: `${
182-
wasConnected
183-
? `${connection.label} disconnected.`
184-
: `${connection.label} could not establish a WebSocket connection.`
186+
!wasConnected
187+
? `${connection.label} could not establish a WebSocket connection.`
188+
: timedOut
189+
? `${connection.label} stopped responding.`
190+
: `${connection.label} disconnected.`
185191
}${networkHint}`,
186192
}),
187193
),

‎patches/[email protected]‎

Lines changed: 33 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
diff --git a/dist/ai/McpServer.js b/dist/ai/McpServer.js
2-
index 2da5941..9fada51 100644
2+
index 2da594107273980144e84ab9b53636ef34043f6b..9fada51944f58fd0d988ae8f6b18f83e79583a25 100644
33
--- a/dist/ai/McpServer.js
44
+++ b/dist/ai/McpServer.js
55
@@ -1093,7 +1093,7 @@ export const layerHttp = options => {
@@ -30,7 +30,7 @@ index 2da5941..9fada51 100644
3030
}));
3131
const mcpHttpSerialization = /*#__PURE__*/(() => {
3232
diff --git a/dist/ai/internal/mcpRuntime.js b/dist/ai/internal/mcpRuntime.js
33-
index 06370fd..baa945b 100644
33+
index 06370fd1341ae7a6b681c513a3a3c2866a414b7b..baa945b8b4763b5472c4a900b90b817209caa4bf 100644
3434
--- a/dist/ai/internal/mcpRuntime.js
3535
+++ b/dist/ai/internal/mcpRuntime.js
3636
@@ -336,6 +336,7 @@ export const make = /*#__PURE__*/Effect.fnUntraced(function* (protocols) {
@@ -42,7 +42,7 @@ index 06370fd..baa945b 100644
4242
canDeliver: (clientId, headers, notification, fallback) => stateful?.canDeliver(clientId, headers, notification, fallback) ?? true,
4343
installHandlers: Effect.fnUntraced(function* (options) {
4444
diff --git a/dist/ai/internal/mcpStatefulRuntime.js b/dist/ai/internal/mcpStatefulRuntime.js
45-
index 9471bdf..4558603 100644
45+
index 9471bdfbcff0643b44df3a08c3701f743a299638..455860316ddfccaf98ff95f9f05da500c331869c 100644
4646
--- a/dist/ai/internal/mcpStatefulRuntime.js
4747
+++ b/dist/ai/internal/mcpStatefulRuntime.js
4848
@@ -46,6 +46,7 @@ export const make = () => {
@@ -54,7 +54,7 @@ index 9471bdf..4558603 100644
5454
const session = resolveSession(clientId, headers);
5555
if (session !== undefined) {
5656
diff --git a/dist/http/HttpClientResponse.js b/dist/http/HttpClientResponse.js
57-
index 42fba67..2b4981b 100644
57+
index 42fba673de3b6bdc089c305f2e87ad56c3bfc374..2b4981b8a88606fb93d62a29f21f9556803307c4 100644
5858
--- a/dist/http/HttpClientResponse.js
5959
+++ b/dist/http/HttpClientResponse.js
6060
@@ -195,7 +195,7 @@ class WebHttpClientResponse extends Inspectable.Class {
@@ -66,10 +66,37 @@ index 42fba67..2b4981b 100644
6666
}
6767
get remoteAddress() {
6868
return Option.none();
69+
diff --git a/dist/rpc/RpcClient.d.ts b/dist/rpc/RpcClient.d.ts
70+
index c498ff74510b784dcba791dbf6b8f23dac40a71d..8be9c910a8112a5289159251846128cd27cf32aa 100644
71+
--- a/dist/rpc/RpcClient.d.ts
72+
+++ b/dist/rpc/RpcClient.d.ts
73+
@@ -302,6 +302,8 @@ export declare const layerProtocolWorker: (options: {
74+
declare const ConnectionHooks_base: Context.ServiceClass<ConnectionHooks, "effect/rpc/RpcClient/ConnectionHooks", {
75+
readonly onConnect: Effect.Effect<void>;
76+
readonly onDisconnect: Effect.Effect<void>;
77+
+ /** Runs when the socket is dropped because pongs stopped, before `onDisconnect`. */
78+
+ readonly onPingTimeout?: Effect.Effect<void> | undefined;
79+
}>;
80+
/**
81+
* Represents optional client protocol hooks that run when a transport connects
6982
diff --git a/dist/rpc/RpcClient.js b/dist/rpc/RpcClient.js
70-
index 8af2e2e..65acae4 100644
83+
index 8af2e2efa9311d582add61fef131fd39fd459ab4..723649f05baa61ac33c9bc3634d106caa9641164 100644
7184
--- a/dist/rpc/RpcClient.js
7285
+++ b/dist/rpc/RpcClient.js
86+
@@ -679,11 +679,11 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun
87+
yield* processData(frames[i]);
88+
}
89+
}
90+
- }).pipe(Effect.scoped, Effect.raceFirst(Effect.flatMap(pinger.timeout, () => Effect.fail(new Socket.SocketError({
91+
+ }).pipe(Effect.scoped, Effect.raceFirst(Effect.flatMap(pinger.timeout, () => (Option.isSome(hooks) && hooks.value.onPingTimeout ? hooks.value.onPingTimeout : Effect.void).pipe(Effect.andThen(Effect.fail(new Socket.SocketError({
92+
reason: new Socket.SocketReadError({
93+
cause: new Error("ping timeout")
94+
})
95+
- })))));
96+
+ })))))));
97+
}).pipe(Option.isSome(hooks) ? Effect.ensuring(hooks.value.onDisconnect) : identity, Effect.tapCause(cause => {
98+
const error = Cause.findError(cause);
99+
const hasError = Result.isSuccess(error);
73100
@@ -726,17 +726,25 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun
74101
const defaultRetryPolicy = /*#__PURE__*/Schedule.min([/*#__PURE__*/Schedule.exponential(500, 1.5), /*#__PURE__*/Schedule.spaced(5000)]);
75102
const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing) {
@@ -98,7 +125,7 @@ index 8af2e2e..65acae4 100644
98125
}).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped);
99126
return {
100127
diff --git a/src/http/HttpClientResponse.ts b/src/http/HttpClientResponse.ts
101-
index c037015..a8b0d8b 100644
128+
index c037015185da1e0c664653b5a496832a9449f3c1..a8b0d8be28bbbd3a0d0d3f1a222c2f074d2d3ad5 100644
102129
--- a/src/http/HttpClientResponse.ts
103130
+++ b/src/http/HttpClientResponse.ts
104131
@@ -341,7 +341,7 @@ class WebHttpClientResponse extends Inspectable.Class implements HttpClientRespo

0 commit comments

Comments
 (0)