diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 8e5a6c7..4345c92 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -135,8 +135,10 @@ impl AppState { /// 2. If a TapEngine was attached, fire `close_tx` — this triggers /// `run_tap_client`'s close arm, which sends `session_end` over the /// WS, awaits brain `bye` (bounded 500 ms), then closes the WS. - /// We bounded-await the engine task for that same 500 ms so the - /// teardown handshake actually completes before we proceed. + /// We bounded-await the engine task for 750 ms (strictly larger + /// than the inner 500 ms bound) so the inner has room to finish its + /// post-timeout `ws.close(None).await` and cleanly exit before our + /// abort fallback fires. /// 3. Fall back to `JoinHandle::abort()` if the bounded-await times out /// (the brain didn't ack, or the pump was stuck — abort as safety net). /// 4. Clear `channel.tap = None` BEFORE state advances to `Closed` per @@ -156,12 +158,17 @@ impl AppState { // — taking it by value would prevent the fallback abort. === if let Some(mut conn) = entry.tap_conn { let _ = conn.close_tx.send(()); - // Match the 500 ms bye-wait bound inside `run_tap_client`'s - // close arm (spec §5.2) — gives the engine time to send - // `session_end`, await `bye`, and close the WS before we - // consider the task done. + // Outer bound STRICTLY LARGER than the inner close-arm bound + // (500 ms in `tap_client.rs::run_tap_client`'s close arm): + // if the brain doesn't `bye`-ack, the inner arm times out at + // ~500 ms and then calls `ws.close(None).await` (which can + // take additional ms) before exiting via `Err(Closed)` → + // `return;`. The 750 ms outer bound gives the inner room to + // finish that post-timeout `ws.close(None).await` and cleanly + // exit before this outer-bound abort fallback fires. + // (final-fixes re-review Important #1.) let teardown = - tokio::time::timeout(Duration::from_millis(500), &mut conn.join).await; + tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await; match teardown { Ok(Ok(())) => { info!(channel_id = %id, "tap engine torn down via DELETE (graceful handshake)"); @@ -170,12 +177,12 @@ impl AppState { warn!(error = ?e, channel_id = %id, "tap engine task failed during teardown"); } Err(_) => { - // Engine didn't finish in 500 ms — abort as fallback. + // Engine didn't finish in 750 ms — abort as fallback. // Spec §5.2: a brain that doesn't bye in time just // gets a WS close; we cap our own wait to keep the // DELETE handler responsive. conn.join.abort(); - info!(channel_id = %id, "tap engine torn down via DELETE (abort after 500ms timeout)"); + info!(channel_id = %id, "tap engine torn down via DELETE (abort after 750ms timeout)"); } } } diff --git a/crates/rutster/tests/tap_integration.rs b/crates/rutster/tests/tap_integration.rs index bbc3e2b..e73140c 100644 --- a/crates/rutster/tests/tap_integration.rs +++ b/crates/rutster/tests/tap_integration.rs @@ -78,17 +78,24 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() { let session_id = ChannelId::new(); let (mut pipe, mut conn) = spawn_tap_engine(session_id, url); - // 3. Push a frame; wait for the engine dial → handshake → pump cycle, - // then for the brain to echo it back as `audio_out`. ~200 ms covers - // a loopback dial + the WS handshake + the JSON echo round-trip. + // 3. Push TWO frames with the same marker `samples[0] = 7` back-to-back + // before the kill so the playout ring has buffered content that the + // read loop has NOT yet drained. This makes the step-7 "silence after + // kill + flush" assertion NON-VACUOUS — there's actually something in + // the ring to flush. Spec §5.3 step 4 contract: the flush side-channel + // must wipe these stale frames so they don't bleed through post-restart. + // Wait long enough first for the engine dial → handshake → pump cycle; + // ~200 ms covers a loopback dial + the WS handshake. tokio::time::sleep(Duration::from_millis(200)).await; let mut frame = PcmFrame::zeroed(); frame.samples[0] = 7; - pipe.on_pcm_frame(frame); + pipe.on_pcm_frame(frame.clone()); + pipe.on_pcm_frame(frame.clone()); - // Drain the engine's audio_out through `next_pcm_frame` until some - // frame arrives (echo round-trip). Bounded to ~400 ms to fail fast on - // a broken pump rather than hanging the test. + // Drain the engine's audio_out through `next_pcm_frame` until SOME frame + // arrives (echo round-trip) — but only drain ONE; we want at least one + // echo frame still buffered in the playout ring at kill time so the + // step-7 flush has something to wipe. Bounded to ~400 ms to fail fast. let mut got_echo = false; for _ in 0..20 { tokio::time::sleep(Duration::from_millis(20)).await; @@ -102,6 +109,10 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() { got_echo, "brain should echo back our PCM within ~400 ms of pump start" ); + // Push one more stale frame so the ring definitely has buffered content + // (the read above may have drained both); this guarantees the §5.3 step-4 + // flush has something to wipe. + pipe.on_pcm_frame(frame); // 4. Kill the EchoServer. `start_echo_server`'s accept_loop aborts // each spawned per-connection task on shutdown — this forces the @@ -176,22 +187,53 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() { // attempt during the outage; in the worst case it's at 1 s or 2 s // by the time we restart. 6 s total budget > 5 s cap, so we'll // definitely observe the resume within the loop. + // + // __STRENGTHENED ASSERTION__ (final-fixes re-review Important #2): + // the original test only checked `next_pcm_frame().is_some()` — a + // future refactor that silently broke the §5.3 playout-ring flush + // would still pass, because the stale-frame bleed-through would + // still surface as Some. We now explicitly assert the returned frame + // carries the FRESH marker (`samples[0] == 9`) and NOT the stale + // marker (`samples[0] == 7`) — actively witnessing the §5.3 step 4 + // "no stale bleed-through" contract. let mut resumed = false; let restart_at = std::time::Instant::now(); while restart_at.elapsed() < Duration::from_secs(6) { - // Push a fresh PCM; if the engine has reconnected, it'll echo - // through to the (flushed) playout ring. + // Push a fresh PCM marker `samples[0] = 9`; if the engine has + // reconnected, it'll echo through to the (flushed) playout ring. let mut f = PcmFrame::zeroed(); f.samples[0] = 9; pipe.on_pcm_frame(f); tokio::time::sleep(Duration::from_millis(50)).await; - if pipe.next_pcm_frame().is_some() { + if let Some(returned) = pipe.next_pcm_frame() { + // Fresh-marker: witnesses reconnect + audio resumption. + assert_eq!( + returned.samples[0], 9, + "resumed frame should carry the FRESH marker (samples[0] == 9), \ + not the stale samples[0] == 7 — catches a broken §5.3 flush \ + side-channel that would bleed stale frames through." + ); resumed = true; break; } } assert!(resumed, "audio should resume after EchoServer restart"); + // 9b. Belt-and-suspenders: walk a few more frames and assert NONE of + // them carry the stale marker `samples[0] == 7`. The fresh-marker + // check above is sufficient to catch the failure mode, but walking + // the next few frames actively witnesses that no stale frame is + // queued behind the fresh one in the playout ring. + for _ in 0..5 { + if let Some(returned) = pipe.next_pcm_frame() { + assert_ne!( + returned.samples[0], 7, + "no stale bleed-through (samples[0] == 7) permitted after the \ + §5.3 step-4 flush + reconnect" + ); + } + } + // 10. Tear down the engine task to avoid leaking. let _ = conn.close_tx.send(()); conn.join.abort();