Skip to content

fix(client-runtime): tolerate transient server stalls in connection heartbeat #3883

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
ramifara wants to merge 2 commits into from
+166 −146
Closed

fix(client-runtime): tolerate transient server stalls in connection heartbeat #3883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
28d0195
fix(client-runtime): tolerate transient server stalls in connection h…
ramifara Jul 11, 2026
986a153
fix(patches): make effect RpcClient heartbeat configurable
ramifara Jul 11, 2026
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
9 changes: 9 additions & 0 deletions packages/client-runtime/src/rpc/session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,15 @@ export const make = Effect.gen(function* () {
RpcClient.makeProtocolSocket({
retryTransientErrors: false,
retryPolicy: Schedule.recurs(0),
// With no protocol-level retry, a single missed pong escalates to a
// full session teardown, so the default 5s/5s heartbeat drops the
// connection whenever the server stalls past 5 seconds. Ping often
// enough to keep NAT/firewall state alive, but give the server room
// to stall before declaring the link dead. These options come from
// the effect patch (patches/effect@4.0.0-beta.78.patch); keep that
// patch in sync when bumping effect.
pingInterval: "10 seconds",
pingTimeout: "20 seconds",
Comment thread
cursor[bot] marked this conversation as resolved.
}),
Comment on lines 99 to 111

@ramifara ramifara Jul 11, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Good catch — confirmed. effect@4.0.0-beta.78 (and current upstream effect-smol main) does not accept these options, so as originally pushed they would only work against a locally modified install.

Fixed in 986a153 by extending the repo's existing effect patch (patches/effect@4.0.0-beta.78.patch, which already customizes this same pinger for connection hooks): makePinger now takes a configurable ping interval and pong deadline (with a per-ping generation guard), makeProtocolSocket/layerProtocolSocket accept pingInterval/pingTimeout, and the .d.ts is updated accordingly. Defaults remain 5s/5s, so nothing changes for other callers.

Verified against the reinstalled patched package (not the local drift): a raw ws test server observed 10 pings in 11s with pingInterval: "1 second" (vs 2 with defaults), and onPingTimeout fired 4.0s after the last pong with 1s interval + 3s deadline (vs 10.0s with defaults). Full vp run typecheck and the client-runtime suite (260 tests) pass.

).pipe(
Layer.provide(
Expand Down
67 changes: 39 additions & 28 deletions patches/effect@4.0.0-beta.78.patch
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
diff --git a/dist/unstable/ai/McpServer.js b/dist/unstable/ai/McpServer.js
index 4819ff0..f5e0f60 100644
--- a/dist/unstable/ai/McpServer.js
index 38eb099d99026120a437701adc4b0fc5b1d0c714..fe29f3368f6d5e52034728ce789c73a318ccc453 100644
--- a/dist/unstable/ai/McpServer.js
+++ b/dist/unstable/ai/McpServer.js
@@ -235,6 +235,26 @@ export const run = /*#__PURE__*/Effect.fnUntraced(function* (options) {
const server = yield* McpServer;
Expand Down Expand Up @@ -30,8 +30,8 @@ index 4819ff0..f5e0f60 100644
clientSessions
}));
diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.ts
index b0f61df..ead663e 100644
--- a/dist/unstable/rpc/RpcClient.d.ts
index b0f61df34613dee99a6f6f1b446a282b2319f5ab..551c3bf6392d0a946f7bff39171e9b8b01863b1d 100644
--- a/dist/unstable/rpc/RpcClient.d.ts
+++ b/dist/unstable/rpc/RpcClient.d.ts
@@ -2,6 +2,7 @@ import * as Cause from "../../Cause.ts";
import * as Context from "../../Context.ts";
Expand All @@ -41,7 +41,25 @@ index b0f61df..ead663e 100644
import * as Layer from "../../Layer.ts";
import * as Queue from "../../Queue.ts";
import * as Schedule from "../../Schedule.ts";
@@ -260,9 +261,40 @@ export declare const layerProtocolWorker: (options: {
@@ -213,6 +214,8 @@ export declare const layerProtocolHttp: (options: {
export declare const makeProtocolSocket: (options?: {
readonly retryTransientErrors?: boolean | undefined;
readonly retryPolicy?: Schedule.Schedule<any, Socket.SocketError> | undefined;
+ readonly pingInterval?: Duration.Input | undefined;
+ readonly pingTimeout?: Duration.Input | undefined;
}) => Effect.Effect<Protocol["Service"], never, Scope.Scope | RpcSerialization.RpcSerialization | Socket.Socket>;
/**
* Provides a client `Protocol` backed by the current `Socket` and
@@ -223,6 +226,8 @@ export declare const makeProtocolSocket: (options?: {
*/
export declare const layerProtocolSocket: (options?: {
readonly retryTransientErrors?: boolean | undefined;
+ readonly pingInterval?: Duration.Input | undefined;
+ readonly pingTimeout?: Duration.Input | undefined;
}) => Layer.Layer<Protocol, never, Socket.Socket | RpcSerialization.RpcSerialization>;
/**
* Creates a client `Protocol` backed by a pool of workers, routing RPC requests
@@ -260,9 +265,40 @@ export declare const layerProtocolWorker: (options: {
readonly targetUtilization?: number | undefined;
readonly timeToLive: Duration.Input;
}) => Layer.Layer<Protocol, WorkerError, Worker.WorkerPlatform | Worker.Spawner>;
Expand Down Expand Up @@ -82,16 +100,9 @@ index b0f61df..ead663e 100644
}>;
/**
* Represents optional client protocol hooks that run when a transport connects
@@ -279,4 +311,4 @@ declare const ConnectionHooks_base: Context.ServiceClass<ConnectionHooks, "effec
export declare class ConnectionHooks extends ConnectionHooks_base {
}
export {};
-//# sourceMappingURL=RpcClient.d.ts.map
\ No newline at end of file
+//# sourceMappingURL=RpcClient.d.ts.map
diff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.js
index a3161eb..0cb81f3 100644
--- a/dist/unstable/rpc/RpcClient.js
index a3161ebd4a0ef74bea2c2f3a4ec95b6dd89ace8f..30ae7f00e7aaeaf63839cbc51e6fca3856638a15 100644
--- a/dist/unstable/rpc/RpcClient.js
+++ b/dist/unstable/rpc/RpcClient.js
@@ -46,6 +46,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro
const services = yield* Effect.context();
Expand Down Expand Up @@ -247,7 +258,7 @@ index a3161eb..0cb81f3 100644
const write = yield* socket.writer;
let parser = serialization.makeUnsafe();
- const pinger = yield* makePinger(write(parser.encode(constPing)));
+ const pinger = yield* makePinger(write(parser.encode(constPing)), Option.isSome(hooks) ? hooks.value : undefined);
+ const pinger = yield* makePinger(write(parser.encode(constPing)), Option.isSome(hooks) ? hooks.value : undefined, options?.pingInterval ?? "5 seconds", options?.pingTimeout ?? "5 seconds");
let currentError;
const onOpen = Effect.suspend(() => {
currentError = undefined;
Expand Down Expand Up @@ -276,32 +287,39 @@ index a3161eb..0cb81f3 100644
}).pipe(Effect.flatMap(() => Effect.fail(new Socket.SocketError({
reason: new Socket.SocketCloseError({
code: 1000
@@ -664,20 +693,20 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun
@@ -664,21 +693,25 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun
};
}));
const defaultRetryPolicy = /*#__PURE__*/Schedule.exponential(500, 1.5).pipe(/*#__PURE__*/Schedule.either(/*#__PURE__*/Schedule.spaced(5000)));
-const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing) {
+const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing, hooks) {
+const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing, hooks, pingInterval, pingTimeout) {
let recievedPong = true;
+ let generation = 0;
const latch = Latch.makeUnsafe();
const reset = () => {
+ generation += 1;
recievedPong = true;
latch.closeUnsafe();
};
- const onPong = () => {
+ const onPong = Effect.sync(() => {
+ generation += 1;
recievedPong = true;
- };
+ }).pipe(Effect.andThen(hooks?.onPong ?? Effect.void));
yield* Effect.suspend(() => {
if (!recievedPong) return latch.open;
- if (!recievedPong) return latch.open;
+ if (!recievedPong) return Effect.void;
recievedPong = false;
- return writePing;
+ return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing));
}).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped);
- }).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped);
+ const pingGeneration = ++generation;
+ return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing), Effect.andThen(Effect.sleep(pingTimeout).pipe(Effect.andThen(Effect.suspend(() => !recievedPong && generation === pingGeneration ? latch.open : Effect.void)), Effect.forkScoped)));
+ }).pipe(Effect.delay(pingInterval), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped);
return {
timeout: latch.await,
@@ -820,6 +849,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun
reset,
@@ -820,6 +853,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun
* @since 4.0.0
*/
export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PURE__*/Layer.effect(Protocol));
Expand All @@ -313,10 +331,3 @@ index a3161eb..0cb81f3 100644
/**
* Represents optional client protocol hooks that run when a transport connects
* and disconnects.
@@ -835,4 +869,4 @@ export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PU
export class ConnectionHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/ConnectionHooks") {}
// internal
const decodeDefect = /*#__PURE__*/Schema.decodeSync(/*#__PURE__*/Schema.Defect());
-//# sourceMappingURL=RpcClient.js.map
\ No newline at end of file
+//# sourceMappingURL=RpcClient.js.map
Loading
Loading