From 0f3ee9935491538e3681a328c04e13007bab7b29 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Sun, 28 Jun 2026 20:23:48 -0400 Subject: [PATCH] =?UTF-8?q?fix(slice-1):=20F1+F3=20=E2=80=94=20feed=20Inpu?= =?UTF-8?q?t::Timeout=20+=20honor=20drive()=20return=20in=20sleep?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review F1 (High) + F3 (Med): per the user-locked Option B direction (slice-1 spec §3.4), one atomic change closing both findings. F1 (input): feed Input::Timeout(now) at the top of every drive() tick. str0m's internal clock (DTLS retransmit, ICE consent freshness RFC 7675, RTCP scheduling) only advances via Input::Timeout — the previous loop only fed Input::Receive, so sustained inbound silence (muted mic, hold, brain thinking mid-step-3) froze the clock; consent checks stopped; the browser tore the call down (~15-30 s) before the 60 s idle timeout fired. F3 (output): drive() returns Option = str0m's next-deadline - now. The return was previously discarded via `let _ =` in session_map::drive_all_sessions. Now the poll task sleeps `min(per-session deadline, 10 ms)` between ticks: honoring the deadline saves wakeups when str0m has nothing pending; the 10 ms cap guarantees the outbound 20 ms encode tick fires on time. TDD via the PacketIo harness + a new observability counter: - session.timeouts_fed: u64 — increments each drive() that feeds Input::Timeout. Doubles as the F1 regression test's seam and the embryo of a real metric per AGENTS.md's OTel aspirations. - RED: f1_drive_feeds_input_timeout_to_advance_str0m_clock asserted timeouts_fed > 0; pre-fix the counter stayed at 0. - GREEN: added Step 0 in drive() (Input::Timeout feed + counter increment). Test passes. - Refactor: session_map::drive_all_sessions returns Duration (was ()); poll task uses tokio::time::sleep instead of fixed tokio::time::interval(10ms). --- crates/rutster-media/src/loop_driver.rs | 16 +++++++++ crates/rutster-media/src/rtc_session.rs | 45 +++++++++++++++++++++++++ crates/rutster/src/session_map.rs | 41 ++++++++++++++++++---- 3 files changed, 96 insertions(+), 6 deletions(-) diff --git a/crates/rutster-media/src/loop_driver.rs b/crates/rutster-media/src/loop_driver.rs index d647db8..d6b9af6 100644 --- a/crates/rutster-media/src/loop_driver.rs +++ b/crates/rutster-media/src/loop_driver.rs @@ -53,6 +53,22 @@ const OUTBOUND_TICK: Duration = Duration::from_millis(20); /// to `Closed`. /// 5. Return the `Duration` to the next `Timeout`. pub fn drive(session: &mut RtcSession, now: Instant) -> Option { + // === Step 0: feed Input::Timeout to advance str0m's internal clock. === + // F1 (adversarial review, High): str0m's internal clock — which + // schedules DTLS retransmit, ICE consent freshness (RFC 7675), and + // RTCP — only advances via `Input::Timeout`. The previous loop only + // ever fed `Input::Receive`; sustained inbound silence (muted mic, + // hold, brain thinking mid-step-3) froze the clock; consent checks + // stopped; a browser would tear the call down (~15-30 s) before the + // 60 s idle timeout fired. Feeding `Input::Timeout` at the top of + // every tick — independent of inbound traffic — keeps the clock + // live. The returned `Result` is logged+dropped on Err per the + // hot-path "drop + observe" policy (§3.8). + if session.rtc.handle_input(Input::Timeout(now)).is_err() { + tracing::warn!("str0m rejected Input::Timeout; continuing"); + } + session.timeouts_fed = session.timeouts_fed.saturating_add(1); + // === Step 1: drain our UDP socket non-blocking, feed str0m. === let mut buf = [0u8; 2000]; loop { diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs index 7cbd5da..a18bdbb 100644 --- a/crates/rutster-media/src/rtc_session.rs +++ b/crates/rutster-media/src/rtc_session.rs @@ -119,6 +119,14 @@ pub struct RtcSession { /// `From` impl interprets the duration against the /// underlying clock frequency. pub(crate) next_media_time: str0m::media::MediaTime, + /// Observability counter: number of times `loop_driver::drive()` has + /// fed `Input::Timeout(now)` to str0m. Doubles as the F1 regression + /// test's seam (slice-1 review "test-coverage gap" finding): the + /// counter stays at 0 pre-fix (drive() only fed `Input::Receive`), + /// and increments once per `drive()` call post-fix. Per AGENTS.md's + /// OTel aspirations, this is the embryo of a real metric — not a + /// pure test spy. + pub(crate) timeouts_fed: u64, } impl RtcSession { @@ -168,6 +176,7 @@ impl RtcSession { last_rx: Instant::now(), last_outbound_at: Instant::now(), next_media_time: str0m::media::MediaTime::ZERO, + timeouts_fed: 0, }) } @@ -396,6 +405,42 @@ a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r /// so a peer that kept ICE alive but sent no audio was never reaped /// within 60 s. Post-fix the bump lives only in /// `handle_event::Event::MediaData`. + /// Adversarial review F1 (High): str0m's internal clock — which + /// drives DTLS retransmit, ICE consent freshness (RFC 7675), and + /// RTCP scheduling — can only advance via `Input::Timeout(now)`. + /// Pre-fix `drive()` only ever fed `Input::Receive`, so sustained + /// inbound silence (muted mic, brain thinking mid-step-3, hold) + /// froze str0m's clock; consent checks stopped; the browser tore + /// the call down (~15–30 s) before the 60 s idle timeout fired. + /// + /// The fix feeds `Input::Timeout(now)` at the top of each `drive()` + /// tick. The `timeouts_fed` counter on `RtcSession` is the test + /// seam (and the embryo of a real observability metric per AGENTS.md). + #[test] + fn f1_drive_feeds_input_timeout_to_advance_str0m_clock() { + use crate::packet_io::test_support::FakePackets; + use std::net::SocketAddr; + use std::time::Duration; + + let local_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let mut session = RtcSession::new_with_socket(Box::new(FakePackets::new()), local_addr); + let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("offer"); + + let initial_timeouts_fed = session.timeouts_fed; + // Drive once at a `now` ~10 s after construction. The fake has + // no queued datagrams, so Step 1's `recv_from` returns + // WouldBlock; pre-fix nothing else touched str0m's clock. + let now = Instant::now() + Duration::from_secs(10); + let _ = crate::loop_driver::drive(&mut session, now); + + assert!( + session.timeouts_fed > initial_timeouts_fed, + "F1: drive() must feed Input::Timeout(now) at the top so str0m's \ + internal clock advances during inbound silence. Pre-fix only \ + Input::Receive advanced it; sustained silence froze str0m." + ); + } + #[test] fn f4_non_rtp_datagram_does_not_reset_idle_timer() { use crate::packet_io::test_support::{FakePackets, RTP_HEADER_NO_PAYLOAD}; diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 4345c92..dfc7f3b 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -198,6 +198,15 @@ impl AppState { } /// Spawn the single poll task for all sessions (idempotent). + /// + /// F3 (adversarial review): the previous loop used a fixed + /// `tokio::time::interval(10ms)` and discarded `drive()`'s returned + /// `Option`. Per the user-locked Option B direction + /// (slice-1 spec §3.4), the poll task now sleeps `min(deadline, 10 ms)` + /// between ticks — `drive()`'s return value (str0m's next-deadline − + /// now) is honored when it's sooner than the 10 ms cap, saving + /// wakeups when str0m has nothing pending. The 10 ms cap guarantees + /// the outbound 20 ms encode tick fires on time regardless. pub async fn spawn_poll_task(self) { let mut running = self.poll_running.lock().await; if *running { @@ -208,12 +217,12 @@ impl AppState { let state = self.clone(); tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_millis(10)); - interval.tick().await; + // First tick fires immediately so a freshly-created session + // gets driven without waiting 10 ms. loop { - interval.tick().await; let now = Instant::now(); - drive_all_sessions(&state, now).await; + let sleep_d = drive_all_sessions(&state, now).await; + tokio::time::sleep(sleep_d).await; } }); } @@ -234,6 +243,13 @@ impl Default for AppState { /// - `Connected && tap.is_none()` → spawn TapEngine → wire TapAudioPipe /// into RtcSession via `with_pipe` → set `channel.tap = Some(TapHandle)`. /// +/// F1+F3 (slice-1 review): returns the `Duration` to sleep before the +/// next iteration — `min(per-session drive() returns, POLL_TICK_CAP)`. +/// Honoring str0m's next-deadline saves wakeups when str0m has nothing +/// pending; the 10 ms cap guarantees the outbound 20 ms encode tick +/// (loop_driver::OUTBOUND_TICK, spec §3.4) fires on time. A session +/// returning `None` (closed/empty) is dropped from the min reduction. +/// /// Teardown is NOT a poll-task branch — see `AppState::close` and the /// module-level docs for why. /// @@ -243,10 +259,16 @@ impl Default for AppState { /// locally — imperceptible). On the NEXT cycle the TapAudioPipe is wired. /// Acceptable for slice-2 — do NOT swap mid-cycle (the spec specifically /// says the spawn happens here, not in `loop_driver.rs`). -async fn drive_all_sessions(state: &AppState, now: Instant) { +async fn drive_all_sessions(state: &AppState, now: Instant) -> Duration { + /// 10 ms cap — see `loop_driver::OUTBOUND_TICK` (spec §3.4: the + /// outbound encode tick fires every 20 ms; sleeping past 10 ms + /// risks missing the boundary). + const POLL_TICK_CAP: Duration = Duration::from_millis(10); + // Collect ids first to avoid holding the DashMap shard during the // async poll (which would block other handlers mutating the same shard). let ids: Vec = state.sessions.iter().map(|r| *r.key()).collect(); + let mut min_sleep = POLL_TICK_CAP; for id in ids { // === slice-2 §5.3 step 4: drain the per-session flush side-channel // BEFORE the poll cycle. The engine task signals a flush via @@ -277,7 +299,13 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { // frames the brain queues during this poll go into an empty ring. s.clear_playout_ring(); } - let _ = s.run_poll_once(now); // hot-path match-and-continue inside + // F1+F3: capture str0m's next-deadline return value (was `let _ =`). + let returned = s.run_poll_once(now); + if let Some(d) = returned { + if d < min_sleep { + min_sleep = d; + } + } // === slice-2: TapEngine spawn seam (spec §5.1 step 3, §8.5 #6). === // Observe the state AFTER `loop_driver::drive` mutated `channel.state` @@ -318,4 +346,5 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { debug!(channel_id = %id, "session evicted after close"); } } + min_sleep }