fix(slice-1): F1+F3 — feed Input::Timeout + honor drive() return in sleep

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<Duration> = 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

@@ -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<Duration> {
+    // === 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 {

crates/rutster-media/src/rtc_session.rs

@@ -119,6 +119,14 @@ pub struct RtcSession {
     /// `From<Duration>` 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};

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<Duration>`. 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));
+            // First tick fires immediately so a freshly-created session
-            interval.tick().await;
+            // 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<ChannelId> = 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
 }