From 1ff9119817c7ea63d5ecda1dfe3dccaa894cd78d Mon Sep 17 00:00:00 2001 From: Bri <34875062+Monkatraz@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:46:59 -0700 Subject: [PATCH 1/2] Replace per-message heartbeat timeout with a watchdog Every inbound frame cleared and re-armed a setTimeout to track the inactivity deadline, so a busy session allocated a timer and closure per message. Track the last inbound timestamp instead and let a single interval per session compare elapsed wall time against the deadline. Comparing Date.now() rather than counting interval executions keeps this safe under browser timer throttling: a delayed or suspended timer can only postpone detection of a dead connection, never report a heartbeat as missed while messages keep arriving. --- package-lock.json | 4 +- package.json | 2 +- .../sessionStateMachine/SessionConnected.ts | 45 ++++++++++++------ .../sessionStateMachine/stateMachine.test.ts | 46 +++++++++++++++++++ transport/sessionStateMachine/transitions.ts | 4 +- 5 files changed, 81 insertions(+), 20 deletions(-) diff --git a/package-lock.json b/package-lock.json index 206fa6b6..507e123e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@replit/river", - "version": "0.220.0", + "version": "0.220.1", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "@replit/river", - "version": "0.220.0", + "version": "0.220.1", "license": "MIT", "dependencies": { "@bufbuild/protobuf": "^2.11.0", diff --git a/package.json b/package.json index 9fbdadad..de8dbac2 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "@replit/river", "description": "It's like tRPC but... with JSON Schema Support, duplex streaming and support for service multiplexing. Transport agnostic!", - "version": "0.220.0", + "version": "0.220.1", "repository": { "type": "git", "url": "git+https://github.com/replit/river.git" diff --git a/transport/sessionStateMachine/SessionConnected.ts b/transport/sessionStateMachine/SessionConnected.ts index a4997626..f0ce6a8b 100644 --- a/transport/sessionStateMachine/SessionConnected.ts +++ b/transport/sessionStateMachine/SessionConnected.ts @@ -55,7 +55,8 @@ export class SessionConnected< listeners: SessionConnectedListeners; private heartbeatHandle?: ReturnType | undefined; - private heartbeatMissTimeout?: ReturnType | undefined; + private heartbeatWatchdog?: ReturnType | undefined; + private lastInboundAt = Date.now(); private isActivelyHeartbeating = false; private rehandshakeTimer?: ReturnType | undefined; private credentialExpiry?: number | undefined; @@ -63,16 +64,11 @@ export class SessionConnected< updateBookkeeping(ack: number, seq: number) { this.sendBuffer = this.sendBuffer.filter((unacked) => unacked.seq >= ack); this.ack = seq + 1; + this.lastInboundAt = Date.now(); if (this.sendBuffer.length < this.options.sendBufferHighWaterMark) { this.notifySendBufferDrain(); } - - if (this.heartbeatMissTimeout) { - clearTimeout(this.heartbeatMissTimeout); - } - - this.startMissingHeartbeatTimeout(); } private assertSendOrdering(encodedMsg: EncodedTransportMessage) { @@ -163,10 +159,26 @@ export class SessionConnected< }; } - startMissingHeartbeatTimeout() { + /** + * Arms the watchdog that closes the connection once the peer stops sending. + * + * A single interval for the lifetime of the session compares wall time against + * the last inbound message, so a busy session does not allocate a timer per + * frame. Elapsed time comes from {@link Date.now}, thus a throttled or suspended + * timer can only delay detection of a dead connection, never report a heartbeat + * as missed while messages keep arriving. + */ + startHeartbeatWatchdog() { const maxMisses = this.options.heartbeatsUntilDead; const missDuration = maxMisses * this.options.heartbeatIntervalMs; - this.heartbeatMissTimeout = setTimeout(() => { + this.heartbeatWatchdog = setInterval(() => { + if (Date.now() - this.lastInboundAt < missDuration) { + return; + } + + // the peer is gone, so nothing will refresh the deadline: stop checking + this.clearHeartbeatWatchdog(); + this.log?.info( `closing connection to ${this.to} due to inactivity (missed ${maxMisses} heartbeats which is ${missDuration}ms)`, this.loggingMetadata, @@ -176,7 +188,14 @@ export class SessionConnected< ); this.conn.close(); - }, missDuration); + }, this.options.heartbeatIntervalMs); + } + + private clearHeartbeatWatchdog() { + if (this.heartbeatWatchdog) { + clearInterval(this.heartbeatWatchdog); + this.heartbeatWatchdog = undefined; + } } startActiveHeartbeat() { @@ -369,11 +388,7 @@ export class SessionConnected< this.heartbeatHandle = undefined; } - if (this.heartbeatMissTimeout) { - clearTimeout(this.heartbeatMissTimeout); - this.heartbeatMissTimeout = undefined; - } - + this.clearHeartbeatWatchdog(); this.clearRehandshakeTimer(); } diff --git a/transport/sessionStateMachine/stateMachine.test.ts b/transport/sessionStateMachine/stateMachine.test.ts index 9a8c6889..c6546079 100644 --- a/transport/sessionStateMachine/stateMachine.test.ts +++ b/transport/sessionStateMachine/stateMachine.test.ts @@ -2046,6 +2046,52 @@ describe('session state machine', () => { }); }); + test('closes the connection once nothing arrives before the deadline', async () => { + const sessionHandle = await createSessionConnected(); + const missDuration = + testingSessionOptions.heartbeatsUntilDead * + testingSessionOptions.heartbeatIntervalMs; + + await vi.advanceTimersByTimeAsync(missDuration - 1); + expect(sessionHandle.onConnectionClosed).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync( + testingSessionOptions.heartbeatIntervalMs, + ); + expect(sessionHandle.onConnectionClosed).toHaveBeenCalledTimes(1); + }); + + test('inbound messages keep the connection past the deadline', async () => { + const sessionHandle = await createSessionConnected(); + const session = sessionHandle.session; + + // stay quiet for just under the deadline between each message, so the + // session only survives if every message pushes the deadline out + const missDuration = + testingSessionOptions.heartbeatsUntilDead * + testingSessionOptions.heartbeatIntervalMs; + for (let seq = 0; seq < 5; seq++) { + session.conn.onData( + session.options.codec.toBuffer({ + id: `msgid-${seq}`, + to: session.from, + from: session.to, + seq, + ack: 0, + streamId: 'heartbeat', + controlFlags: ControlFlags.AckBit, + payload: { + type: 'ACK', + } satisfies Static, + }), + ); + + await vi.advanceTimersByTimeAsync(missDuration - 1); + } + + expect(sessionHandle.onConnectionClosed).not.toHaveBeenCalled(); + }); + test('does not dispatch acks', async () => { const sessionHandle = await createSessionConnected(); const session = sessionHandle.session; diff --git a/transport/sessionStateMachine/transitions.ts b/transport/sessionStateMachine/transitions.ts index f0df3e65..556d7740 100644 --- a/transport/sessionStateMachine/transitions.ts +++ b/transport/sessionStateMachine/transitions.ts @@ -234,7 +234,7 @@ export const SessionStateGraph = { ...carriedState, }); - session.startMissingHeartbeatTimeout(); + session.startHeartbeatWatchdog(); session.log?.info( `session ${session.id} transition from Handshaking to Connected`, @@ -293,7 +293,7 @@ export const SessionStateGraph = { ...carriedState, }); - session.startMissingHeartbeatTimeout(); + session.startHeartbeatWatchdog(); conn.telemetry = createConnectionTelemetryInfo( session.tracer, From ea76aabf5d642610ede91421b5f2d21eddd04413 Mon Sep 17 00:00:00 2001 From: Bri <34875062+Monkatraz@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:05:32 -0700 Subject: [PATCH 2/2] Give the heartbeat watchdog tests realistic timing margins The tests asserted the connection was still alive 1ms before the inactivity deadline. The suite runs with shouldAdvanceTime, so the fake clock also moves with real time, and on a contended runner it crossed that 1ms margin and fired the watchdog early. Use the existing heartbeat and disconnect grace helpers instead, which leaves a full heartbeat interval of slack on either side of the deadline. Both tests still fail if an inbound message stops refreshing the timestamp. --- .../sessionStateMachine/stateMachine.test.ts | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/transport/sessionStateMachine/stateMachine.test.ts b/transport/sessionStateMachine/stateMachine.test.ts index c6546079..8d83b0d2 100644 --- a/transport/sessionStateMachine/stateMachine.test.ts +++ b/transport/sessionStateMachine/stateMachine.test.ts @@ -3,7 +3,11 @@ import { payloadToTransportMessage, testingSessionOptions, } from '../../testUtil'; -import { waitFor } from '../../testUtil/fixtures/cleanup'; +import { + advanceFakeTimersByDisconnectGrace, + advanceFakeTimersByHeartbeat, + waitFor, +} from '../../testUtil/fixtures/cleanup'; import { ControlFlags, ControlMessageAckSchema, @@ -2048,16 +2052,15 @@ describe('session state machine', () => { test('closes the connection once nothing arrives before the deadline', async () => { const sessionHandle = await createSessionConnected(); - const missDuration = - testingSessionOptions.heartbeatsUntilDead * - testingSessionOptions.heartbeatIntervalMs; - await vi.advanceTimersByTimeAsync(missDuration - 1); + // a heartbeat of silence is still well inside the deadline + await advanceFakeTimersByHeartbeat(); expect(sessionHandle.onConnectionClosed).not.toHaveBeenCalled(); - await vi.advanceTimersByTimeAsync( - testingSessionOptions.heartbeatIntervalMs, - ); + await advanceFakeTimersByDisconnectGrace(); + + // the watchdog stops itself once it fires, so the peer is only declared + // dead once rather than on every subsequent check expect(sessionHandle.onConnectionClosed).toHaveBeenCalledTimes(1); }); @@ -2065,11 +2068,8 @@ describe('session state machine', () => { const sessionHandle = await createSessionConnected(); const session = sessionHandle.session; - // stay quiet for just under the deadline between each message, so the - // session only survives if every message pushes the deadline out - const missDuration = - testingSessionOptions.heartbeatsUntilDead * - testingSessionOptions.heartbeatIntervalMs; + // a message every heartbeat for well past the deadline: the session only + // survives if each one pushes the deadline out for (let seq = 0; seq < 5; seq++) { session.conn.onData( session.options.codec.toBuffer({ @@ -2086,7 +2086,7 @@ describe('session state machine', () => { }), ); - await vi.advanceTimersByTimeAsync(missDuration - 1); + await advanceFakeTimersByHeartbeat(); } expect(sessionHandle.onConnectionClosed).not.toHaveBeenCalled();