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
2026-11-05 16:04:43 +00:00 · 2026-11-05 20:57:13 +07:00
parent e6064bb374
commit 412550c36f
3 changed files with 199 additions and 43 deletions
@@ -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();
@@ -1296,6 +1296,7 @@ interface PingProbe {
  reject: (error: Error) => void;
  timeoutHandle: ReturnType<typeof setTimeout>;
  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<typeof setInterval> | null = null;
  private runtimeMetrics: DaemonClientRuntimeMetrics | null = null;
  private pingProbe: PingProbe | null = null;
-  private connectionVerification: DaemonTransport | null = null;
  private livenessHeartbeatTimer: ReturnType<typeof setTimeout> | 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<number> {
    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<typeof setTimeout>;
+    const timeout = new Promise<never>((_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<number> {
@@ -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)
  // ============================================================================
@@ -25,6 +25,7 @@ export class ChunkedTransport {
      supportsDrain?: boolean;
      send: (frame: Frame, waitForDrain: boolean) => void | Promise<void>;
      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);
  }