diff --git a/docs/ENGINE.md b/docs/ENGINE.md index f4423d68..068be6b1 100644 --- a/docs/ENGINE.md +++ b/docs/ENGINE.md @@ -635,7 +635,7 @@ Client text control frames: ```json {"type":"resize","cols":120,"rows":40} -{"type":"ping"} +{"type":"ping","id":1} ``` Server text control frames: @@ -643,9 +643,18 @@ Server text control frames: ```json {"type":"snapshot-start","session_id":"uuid","scrollback_bytes":0,"scrollback_pos":0,"reset":true} {"type":"snapshot-done"} +{"type":"pong","id":1,"pos":123} {"type":"lag","note":"client too slow; reattach"} ``` +A `pong` echoes the `ping.id` and carries `pos`: the absolute byte offset the +server has **actually streamed to that socket**, not `total_written`. The pong +is formed and sent by the same outbound task that streams output, after +flushing anything already queued, so it is ordered on the wire after those +chunks and its `pos` can never exceed what the client has received. A client +uses it purely as a liveness probe — a `pos` ahead of what it has rendered is a +suspect to confirm with a second probe, not an immediate reconnect. + `snapshot_tail_bytes` is the replay **budget**, and it bounds every reply — a client's terminal keeps only a fixed scrollback, so the server never ships more than the client can hold: diff --git a/engine/server/src/ws.rs b/engine/server/src/ws.rs index 86c1f49a..89b16f4a 100644 --- a/engine/server/src/ws.rs +++ b/engine/server/src/ws.rs @@ -99,6 +99,47 @@ fn coalesce(snap_pos: u64, chunks: &[OutputChunk]) -> (Vec, u64) { (out, end) } +/// Send whatever broadcast chunks are already queued for this socket, without +/// blocking, advancing `sent_pos` past them. Used before answering a liveness +/// probe so the pong's position reflects output the client has actually been +/// sent. A `Lagged`/`Closed` on the non-blocking drain is left for the main +/// `rx.recv()` loop to handle (resync or teardown); flushing what was already +/// pulled is still correct and gap-free. +async fn flush_available( + sink: &mut S, + rx: &mut tokio::sync::broadcast::Receiver, + snap_pos: u64, + sent_pos: &mut u64, +) -> Result<(), ()> +where + S: SinkExt + Unpin, +{ + let mut drained: Vec = Vec::new(); + let mut queued = 0usize; + while queued < OUTBOUND_COALESCE_CAP { + match rx.try_recv() { + Ok(next) => { + queued += next.data.len(); + drained.push(next); + } + Err(_) => break, + } + } + if drained.is_empty() { + return Ok(()); + } + let (frame, end) = coalesce(snap_pos, &drained); + if end > *sent_pos { + *sent_pos = end; + } + if !frame.is_empty() { + sink.send(Message::Binary(frame.into())) + .await + .map_err(|_| ())?; + } + Ok(()) +} + /// Re-synchronise a lagging client in-band, on the same socket, instead of /// dropping it and forcing a reconnect + replay (a cascade under load). /// @@ -315,7 +356,11 @@ async fn handle_socket( }; let (mut sink, mut stream) = socket.split(); - let (control_tx, mut control_rx) = mpsc::unbounded_channel::(); + // Carries liveness-probe ids from the inbound task to the outbound task. + // The pong is *formed* by the outbound task with the position it has + // actually streamed, not by the inbound task with `total_written` — see the + // ping branch of the outbound select (WI-126). + let (ping_tx, mut ping_rx) = mpsc::unbounded_channel::(); // Subscribe BEFORE snapshotting so no broadcast chunks are missed in the gap. let mut rx = session.subscribe(); @@ -386,10 +431,11 @@ async fn handle_socket( let _ = writer_session.resize(cols, rows); } Ok(ClientControl::Ping { id }) => { - let _ = control_tx.send(ServerControl::Pong { - id, - pos: writer_session.scrollback_position(), - }); + // Hand the probe to the outbound task; it answers + // with the position it has actually sent this + // socket, after flushing queued output, so the pong + // can never report the server ahead of the client. + let _ = ping_tx.send(id); } Ok(ClientControl::Auth { .. }) => { // Already authenticated; ignore further auth frames. @@ -427,9 +473,23 @@ async fn handle_socket( loop { tokio::select! { - Some(control) = control_rx.recv() => { + Some(ping_id) = ping_rx.recv() => { + // Flush anything already queued for this socket, then answer + // the probe with `sent_pos` — the byte offset actually + // written here. Because the pong is ordered AFTER those + // chunks on the wire and carries only what has been sent, it + // can never report the server ahead of what the client has + // received, so a bursty session no longer trips a spurious + // recycle (WI-126). + if flush_available(&mut sink, &mut rx, snap_pos, &mut sent_pos) + .await + .is_err() + { + break; + } + let pong = ServerControl::Pong { id: ping_id, pos: sent_pos }; if sink - .send(Message::Text(serde_json::to_string(&control).unwrap().into())) + .send(Message::Text(serde_json::to_string(&pong).unwrap().into())) .await .is_err() { diff --git a/engine/server/tests/integration.rs b/engine/server/tests/integration.rs index cc569707..535e4507 100644 --- a/engine/server/tests/integration.rs +++ b/engine/server/tests/integration.rs @@ -3877,6 +3877,94 @@ async fn lagging_subscriber_recovers_in_band_bounded_and_without_duplicates() { kill_session(&client, &base, &id).await; } +#[tokio::test] +async fn pong_never_reports_the_server_ahead_of_this_socket() { + // F2 (WI-126). Under load the liveness pong must carry the position actually + // streamed to THIS socket, never `total_written` (which includes queued but + // unsent output). Before the fix the inbound task answered with + // scrollback_position() and the pong could be delivered ahead of the chunks + // it referenced, so the client saw the server "ahead" and recycled the + // socket. Now the outbound task answers with sent_pos after flushing, so the + // pong's pos is never greater than what this socket has received. + let (base, _h) = boot_with(4 * 1024 * 1024).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + let id = create_command_session(&client, &base, "pong", load_session_command(50_000)).await; + + let mut ws = ws_attach(&base, &id).await; + + // Drain the initial snapshot, recording the absolute position it ended at. + let mut snap_end: u64 = 0; + let mut in_snapshot = false; + loop { + let m = tokio::time::timeout(Duration::from_secs(5), ws.next()) + .await + .expect("snapshot frame arrives") + .unwrap() + .unwrap(); + match m { + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + match v["type"].as_str() { + Some("snapshot-start") => { + in_snapshot = true; + snap_end = v["scrollback_pos"].as_u64().unwrap(); + } + Some("snapshot-done") => break, + _ => {} + } + } + Message::Binary(_) if in_snapshot => {} + _ => {} + } + } + + // Probe while output is still flowing. + ws.send(Message::Text( + json!({ "type": "ping", "id": 1 }).to_string().into(), + )) + .await + .unwrap(); + + // Count live bytes received on this socket until the pong arrives. The pong + // is ordered after any flushed chunks, so by the time we read it we have + // received every byte up to its position. + let mut live: u64 = 0; + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + let m = tokio::time::timeout(remaining, ws.next()) + .await + .expect("a pong should arrive within the deadline") + .unwrap() + .unwrap(); + match m { + Message::Binary(b) => live += b.len() as u64, + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + if v["type"] == "pong" { + assert_eq!(v["id"].as_u64(), Some(1)); + let pos = v["pos"].as_u64().unwrap(); + let received = snap_end + live; + assert!( + pos <= received, + "pong pos {pos} is ahead of what this socket received \ + ({received} = snap_end {snap_end} + live {live})" + ); + break; + } + // A resync snapshot-start would only add to `received`; keep going. + } + _ => {} + } + } + + ws.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + #[tokio::test] async fn ws_attach_echoes_input_and_replays_on_reattach() { let (base, _h) = boot().await; diff --git a/web/src/Terminal.tsx b/web/src/Terminal.tsx index 90da21bf..fa1baa19 100644 --- a/web/src/Terminal.tsx +++ b/web/src/Terminal.tsx @@ -1025,7 +1025,14 @@ const TerminalView: Component = (props) => { ws?.close(); scheduleReconnect(100); } else if (ctrl.type === "pong") { - if (watchdog.notePong(ctrl.id, ctrl.pos, outputPosition ?? 0) === "recycle") { + if ( + watchdog.notePong( + ctrl.id, + ctrl.pos, + outputPosition ?? 0, + Date.now(), + ) === "recycle" + ) { recycleSocket("server output is ahead of the rendered cursor"); } } diff --git a/web/src/__tests__/terminalWatchdog.test.ts b/web/src/__tests__/terminalWatchdog.test.ts index 4ef00c1f..da05490b 100644 --- a/web/src/__tests__/terminalWatchdog.test.ts +++ b/web/src/__tests__/terminalWatchdog.test.ts @@ -21,16 +21,60 @@ describe("terminal socket watchdog", () => { expect(watchdog.check(100 + WATCHDOG_TIMEOUT_MS)).toBe("healthy"); }); - it("recycles when pong proves that output was missed", () => { + it("does not recycle on a single behind pong; it arms a confirm probe", () => { + // F2: one behind pong is a suspect, not a verdict — the pong can momentarily + // precede the very chunks it is ahead of. No recycle, and check() issues a + // prompt confirm probe rather than waiting a full interval. const watchdog = new SocketWatchdog(); watchdog.notePingSent(7, 1_000); - expect(watchdog.notePong(7, 42, 41)).toBe("recycle"); + expect(watchdog.notePong(7, 42, 41, 1_010)).toBe("healthy"); + expect(watchdog.check(1_020, false)).toBe("probe"); + }); + + it("recycles on a second behind pong within the window with no output", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // first: suspect + watchdog.notePingSent(2, 1_020); // the confirm probe + // Second behind pong, within WATCHDOG_TIMEOUT_MS of the first, nothing in + // between: a real stall. + expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("recycle"); + }); + + it("clears the suspicion when output arrives between the two probes", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect + watchdog.noteOutput(1_015); // the stream is live after all + watchdog.notePingSent(2, 1_020); + // The next behind pong starts a fresh suspicion, it does not recycle. + expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("healthy"); + }); + + it("clears the suspicion when the confirm pong shows the client caught up", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect + watchdog.notePingSent(2, 1_020); + expect(watchdog.notePong(2, 42, 42, 1_030)).toBe("healthy"); // caught up + // Suspicion cleared: the next lone behind pong is again only a suspect. + watchdog.notePingSent(3, 1_040); + expect(watchdog.notePong(3, 50, 41, 1_050)).toBe("healthy"); + }); + + it("drops a stale suspect the confirm never resolved in time", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect at 1010 + // No confirm pong arrives; past the timeout the suspicion is dropped and the + // watchdog falls back to its ordinary interval cadence. + expect(watchdog.check(1_010 + WATCHDOG_TIMEOUT_MS, false)).toBe("healthy"); }); it("does not probe again before the periodic interval", () => { const watchdog = new SocketWatchdog(); watchdog.notePingSent(1, 0); - watchdog.notePong(1, 0, 0); + watchdog.notePong(1, 0, 0, 0); expect(watchdog.check(WATCHDOG_INTERVAL_MS - 1, false)).toBe("healthy"); }); }); diff --git a/web/src/terminalWatchdog.ts b/web/src/terminalWatchdog.ts index 993c6b3d..75deed0a 100644 --- a/web/src/terminalWatchdog.ts +++ b/web/src/terminalWatchdog.ts @@ -14,16 +14,43 @@ export class SocketWatchdog { private pending: PendingPing | null = null; private lastProbeAt = Number.NEGATIVE_INFINITY; private lastOutputAt = Number.NEGATIVE_INFINITY; + // A single "behind" pong (server position ahead of what the client has + // rendered) is a suspect, not a verdict: even with the engine answering a + // probe with its actual sent position, a pong can momentarily precede the + // very chunks it is ahead of. So the first behind pong arms a prompt confirm + // probe instead of recycling; only a second behind pong within the timeout, + // with no output in between, is treated as a real stall. `suspectAt` is when + // the first behind pong was seen, or null when there is no live suspicion. + private suspectAt: number | null = null; notePingSent(id: number, at: number): void { this.pending = { id, at }; this.lastProbeAt = at; } - notePong(id: number, serverPos: number, localPos: number): WatchdogResult { + notePong( + id: number, + serverPos: number, + localPos: number, + at: number, + ): WatchdogResult { if (!this.pending || this.pending.id !== id) return "healthy"; this.pending = null; - return serverPos > localPos ? "recycle" : "healthy"; + if (serverPos <= localPos) { + // Caught up (or the engine reported its true sent position): any prior + // suspicion is cleared. + this.suspectAt = null; + return "healthy"; + } + // Behind. Recycle only on a second behind pong within the timeout window + // that no output cleared in between (a genuine stall); otherwise arm a + // suspect and let `check` issue a prompt confirm probe. + if (this.suspectAt !== null && at - this.suspectAt < WATCHDOG_TIMEOUT_MS) { + this.suspectAt = null; + return "recycle"; + } + this.suspectAt = at; + return "healthy"; } noteOutput(at: number): void { @@ -31,6 +58,9 @@ export class SocketWatchdog { // Old engines do not answer the probe. Output after it is still proof of // life, so an idle shell does not reconnect in a loop against one. if (this.pending && at >= this.pending.at) this.pending = null; + // Output flowing is proof the stream is live, so a behind-pong suspicion is + // no longer credible. + this.suspectAt = null; } check(now: number, forceProbe = false): WatchdogResult { @@ -42,6 +72,16 @@ export class SocketWatchdog { } return "recycle"; } + // A live suspect confirms promptly — one more probe now, not a full + // interval later — so a real stall is caught within the timeout. A suspect + // the confirm never resolved in time is dropped as stale. + if (this.suspectAt !== null) { + if (now - this.suspectAt >= WATCHDOG_TIMEOUT_MS) { + this.suspectAt = null; + } else { + return "probe"; + } + } return forceProbe || now - this.lastProbeAt >= WATCHDOG_INTERVAL_MS ? "probe" : "healthy"; @@ -51,5 +91,6 @@ export class SocketWatchdog { this.pending = null; this.lastProbeAt = Number.NEGATIVE_INFINITY; this.lastOutputAt = Number.NEGATIVE_INFINITY; + this.suspectAt = null; } }