From 412550c36fd5e3625c8050aeb4856837d06c4e09 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 5 Nov 2026 20:57:13 +0700 Subject: [PATCH] Keep active transfers connected when the app returns to foreground (#8651) * fix(client): keep active transfers alive during foreground checks Foreground used an independent three-second probe that closed live connections while a pong waited behind bulk data. Share the heartbeat clock and refresh its outstanding deadline on validated inbound frames, including incomplete chunk assemblies. * test(client): deliver bulk response before the queued pong * fix(client): preserve latency deadlines when sharing heartbeat pings * fix(client): keep heartbeat cadence during active downloads --- packages/client/src/daemon-client.test.ts | 158 ++++++++++++++++++++- packages/client/src/daemon-client.ts | 82 +++++------ packages/protocol/src/chunked-transport.ts | 2 + 3 files changed, 199 insertions(+), 43 deletions(-) diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index eb0dcdb6ca..87719a0a56 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -657,6 +657,38 @@ class FakeDaemon { }, }; + finishTransfer(): void { + // Four MiB of JSON whitespace precedes this valid application message. + const bytes = new TextEncoder().encode( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "srv_heartbeat_test", + hostname: null, + version: null, + features: { binaryChunkingV1: true }, + }, + }, + }), + ); + const frame = new Uint8Array(bytes.length + 4); + frame.set([0x31, 1, 0, 2]); + frame.set(bytes, 4); + this.onMessage(frame); + this.onMessage(JSON.stringify({ type: "pong" })); + this.daemonAnswersPingsAfter("fast"); + } + + receiveChunk(start: boolean): void { + const frame = new Uint8Array(1024 * 1024 + 4); + frame.set([0x31, 1, 0, start ? 1 : 0]); + frame.fill(32, 4); + this.onMessage(frame); + } + openConnection(): void { this.onOpen(); this.onMessage( @@ -667,6 +699,7 @@ class FakeDaemon { payload: { status: "server_info", serverId: "srv_heartbeat_test", + features: { binaryChunkingV1: true }, hostname: null, version: null, }, @@ -2572,7 +2605,7 @@ test("keeps default connect timeout shorter than session RPC waiters", async () } }); -test("foreground verification replaces a silently broken socket without waiting for heartbeats", async () => { +test("foreground checks a silently broken socket through two normal heartbeat misses", async () => { useHeartbeatClock(); const first = new FakeDaemon(); const second = new FakeDaemon(); @@ -2590,7 +2623,11 @@ test("foreground verification replaces a silently broken socket without waiting await connection; client.ensureConnected({ verify: true }); - await vi.advanceTimersByTimeAsync(3_000); + await vi.advanceTimersByTimeAsync(15_000); + expect(first.closesFromClient()).toEqual([]); + await vi.advanceTimersByTimeAsync(25_000); + expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]); + client.ensureConnected(); expect(attempts).toBe(2); second.openConnection(); expect(client.getConnectionState()).toEqual({ status: "connected" }); @@ -2618,7 +2655,110 @@ test("foreground verification preserves a healthy socket and deduplicates simult expect(daemon.closesFromClient()).toEqual([]); }); -test("an obsolete foreground probe cannot close a replacement connection", async () => { +test("chunks prevent heartbeat misses until silence causes two misses", async () => { + useHeartbeatClock(); + const daemon = new FakeDaemon(); + daemon.daemonGoesSilent(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "bulk-resume", + logger: noopLogger, + reconnect: { enabled: false }, + transportFactory: () => daemon.transport, + }); + clients.push(client); + const connection = client.connect(); + daemon.openConnection(); + await connection; + client.ensureConnected({ verify: true }); + let measurementError: unknown; + void client.measureLatency({ timeoutMs: 5_000 }).catch((error: unknown) => { + measurementError = error; + }); + await vi.advanceTimersByTimeAsync(5_000); + expect(measurementError).toEqual( + expect.objectContaining({ + message: "Latency measurement timed out (5000ms)", + }), + ); + for (let index = 0; index < 4; index++) { + await vi.advanceTimersByTimeAsync(index === 0 ? 5_000 : 10_000); + daemon.receiveChunk(index === 0); + expect(daemon.closesFromClient()).toEqual([]); + } + // No complete message or pong has arrived, but progress prevents misses. + expect(daemon.pingTimestamps()).toEqual(["0s", "25s"]); + await vi.advanceTimersByTimeAsync(25_000); + expect(daemon.closesFromClient()).toEqual([]); + await vi.advanceTimersByTimeAsync(25_000); + expect(daemon.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]); +}); + +test("foreground preserves a multi-MB transfer with its pong queued behind chunks", async () => { + useHeartbeatClock(); + const daemon = new FakeDaemon(); + daemon.daemonGoesSilent(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "bulk-resume", + logger: noopLogger, + reconnect: { enabled: false }, + transportFactory: () => daemon.transport, + }); + clients.push(client); + const connection = client.connect(); + daemon.openConnection(); + await connection; + client.ensureConnected({ verify: true }); + await vi.advanceTimersByTimeAsync(3_001); + expect(daemon.closesFromClient()).toEqual([]); + for (let index = 0; index < 4; index++) { + await vi.advanceTimersByTimeAsync(10_000); + daemon.receiveChunk(index === 0); + expect(daemon.closesFromClient()).toEqual([]); + } + // No complete message or pong has arrived, but progress prevents misses. + expect(daemon.pingTimestamps()).toEqual(["0s", "25s"]); + daemon.finishTransfer(); + await vi.advanceTimersByTimeAsync(30_000); + expect(client.getConnectionState()).toEqual({ status: "connected" }); + expect(daemon.closesFromClient()).toEqual([]); +}); + +test("chunk progress beyond the daemon lease keeps sending heartbeat pings", async () => { + useHeartbeatClock(); + const daemon = new FakeDaemon(); + daemon.daemonGoesSilent(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "long-download", + logger: noopLogger, + reconnect: { enabled: false }, + transportFactory: () => daemon.transport, + }); + clients.push(client); + const connection = client.connect(); + daemon.openConnection(); + await connection; + client.ensureConnected({ verify: true }); + for (let index = 0; index < 12; index++) { + await vi.advanceTimersByTimeAsync(5_000); + daemon.receiveChunk(index === 0); + // Each completed window starts the next ping ten seconds later, + // even though the response and all queued pongs are still in flight. + const expected = ["0s"]; + if (index >= 4) expected.push("25s"); + if (index >= 9) expected.push("50s"); + expect(daemon.pingTimestamps()).toEqual(expected); + expect(daemon.closesFromClient()).toEqual([]); + } + daemon.finishTransfer(); + await vi.advanceTimersByTimeAsync(30_000); + expect(client.getConnectionState()).toEqual({ status: "connected" }); + expect(daemon.closesFromClient()).toEqual([]); +}); + +test("an obsolete heartbeat cannot close a replacement connection", async () => { useHeartbeatClock(); const first = new FakeDaemon(); const second = new FakeDaemon(); @@ -2735,6 +2875,18 @@ test("sends only one ping while a slow pong is still outstanding", async () => { expect(session.pingTimestamps()).toEqual(["10s"]); }); +test("a heartbeat sharing a latency measurement still saves the RTT", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("2s"); + await session.connect(); + await session.advance(9_000); + const measurement = session.measureLatency({ timeoutMs: 5_000 }); + await session.advance(2_000); + await expect(measurement).resolves.toBe(2000); + expect(session.lastLivenessRttMs()).toBe(2000); +}); + test("reports the round-trip time of the last successful heartbeat", async () => { useHeartbeatClock(); const session = new DaemonClientSession(); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 7454c8448f..68677c3f24 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1296,6 +1296,7 @@ interface PingProbe { reject: (error: Error) => void; timeoutHandle: ReturnType; startedAt: number; + receivedActivity: boolean; // Whether a timeout on this ping should be recorded as a liveness failure. Only the // heartbeat sets this; a latency measurement never drives teardown, even when a // heartbeat tick shares (dedupes onto) an in-flight measurement ping. @@ -1439,7 +1440,6 @@ export class DaemonClient { private runtimeMetricsInterval: ReturnType | null = null; private runtimeMetrics: DaemonClientRuntimeMetrics | null = null; private pingProbe: PingProbe | null = null; - private connectionVerification: DaemonTransport | null = null; private livenessHeartbeatTimer: ReturnType | null = null; private lastLivenessRttMs: number | null = null; private consecutiveLivenessFailures = 0; @@ -1589,6 +1589,7 @@ export class DaemonClient { send: (frame) => transport.send(frame), onFailure: () => transport.close(4003, "Transport send failed"), deliver: (frame) => this.handleTransportMessage(frame), + onActivity: () => this.recordTransportActivity(), }); this.lastServerInfoMessage = null; this.authFailureReasonValue = null; @@ -1785,7 +1786,7 @@ export class DaemonClient { this.shouldReconnect = true; } if (this.connectionState.status === "connected") { - if (options?.verify) this.verifyConnection(); + if (options?.verify) this.tickLivenessHeartbeat(); return; } if (this.connectionState.status === "connecting") return; @@ -1798,26 +1799,9 @@ export class DaemonClient { ); } - private verifyConnection(): void { - const transport = this.transport; - if (!transport || this.connectionVerification === transport) return; - this.connectionVerification = transport; - // A session probe has its own deadline, independent of a heartbeat that the OS - // may have suspended. A successful response also proves the session can serve RPCs. - void this.ping({ timeoutMs: 3_000 }) - .catch((error: unknown) => { - if (this.transport !== transport || this.connectionState.status !== "connected") return; - this.disposeTransport(1001, "Connection verification failed"); - this.scheduleReconnect({ - reason: error instanceof Error ? error.message : String(error), - event: "CONNECTION_VERIFICATION_FAILED", - reasonCode: "liveness_timeout", - }); - this.ensureConnected(); - }) - .finally(() => { - if (this.connectionVerification === transport) this.connectionVerification = null; - }); + private recordTransportActivity(): void { + this.consecutiveLivenessFailures = 0; + if (this.pingProbe) this.pingProbe.receivedActivity = true; } requireConnectedHosts(): void { @@ -2553,9 +2537,18 @@ export class DaemonClient { measureLatency(params?: { timeoutMs?: number }): Promise { const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS); - return this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: false }).catch((error) => { - throw toTimeoutError(error, "Latency measurement", timeoutMs); + let deadline: ReturnType; + const timeout = new Promise((_resolve, reject) => { + deadline = setTimeout(() => reject(new PingTimeoutError(timeoutMs)), timeoutMs); }); + return Promise.race([ + this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: false }), + timeout, + ]) + .catch((error) => { + throw toTimeoutError(error, "Latency measurement", timeoutMs); + }) + .finally(() => clearTimeout(deadline)); } private async livenessPing(params?: { timeoutMs?: number }): Promise { @@ -2591,21 +2584,23 @@ export class DaemonClient { resolveProbe = resolve; rejectProbe = reject; }); + const expire = () => { + if (this.pingProbe !== probe) return; + this.pingProbe = null; + const error = new PingTimeoutError(timeoutMs); + probe.reject(error); + // Keep the deadline fixed so subsequent pings renew the daemon socket lease + // during a long download. Progress proves life even when its pong is queued. + if (probe.drivesLivenessFailure && !probe.receivedActivity) { + this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs)); + } + }; const probe: PingProbe = { promise, resolve: (value) => resolveProbe?.(value), reject: (error) => rejectProbe?.(error), - timeoutHandle: setTimeout(() => { - if (this.pingProbe !== probe) { - return; - } - this.pingProbe = null; - const error = new PingTimeoutError(timeoutMs); - probe.reject(error); - if (probe.drivesLivenessFailure) { - this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs)); - } - }, timeoutMs), + timeoutHandle: setTimeout(expire, timeoutMs), + receivedActivity: false, startedAt, drivesLivenessFailure: params.drivesLivenessFailure, }; @@ -2646,14 +2641,21 @@ export class DaemonClient { } this.livenessHeartbeatTimer = setTimeout(() => { this.livenessHeartbeatTimer = null; - this.livenessPing({ timeoutMs: LIVENESS_HEARTBEAT_TIMEOUT_MS }) - .catch(() => {}) - .finally(() => { - this.scheduleNextLivenessHeartbeat(); - }); + this.tickLivenessHeartbeat(); }, LIVENESS_HEARTBEAT_INTERVAL_MS); } + private tickLivenessHeartbeat(): void { + this.stopLivenessHeartbeat(); + const transport = this.transport; + const tick = this.livenessPing({ timeoutMs: LIVENESS_HEARTBEAT_TIMEOUT_MS }); + void tick + .catch(() => {}) + .finally(() => { + if (this.transport === transport) this.scheduleNextLivenessHeartbeat(); + }); + } + // ============================================================================ // Agent RPCs (requestId-correlated) // ============================================================================ diff --git a/packages/protocol/src/chunked-transport.ts b/packages/protocol/src/chunked-transport.ts index a32d68424f..8faa22cd96 100644 --- a/packages/protocol/src/chunked-transport.ts +++ b/packages/protocol/src/chunked-transport.ts @@ -25,6 +25,7 @@ export class ChunkedTransport { supportsDrain?: boolean; send: (frame: Frame, waitForDrain: boolean) => void | Promise; deliver: (frame: Frame) => void; + onActivity?: () => void; onFailure?: (error: unknown) => void; }, ) { @@ -134,6 +135,7 @@ export class ChunkedTransport { { cause: error }, ); } + this.io.onActivity?.(); if (message !== undefined) this.io.deliver(message); }