From 84fc30591e031da37824d34e2e09f74449101c93 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Sun, 28 Jun 2026 17:46:50 -0400 Subject: [PATCH] fix(binary): spec-exact backoff staircase + remove dead Closing branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tap_engine.rs: Backoff::next_delay now matches spec §4.3 exactly: 250ms -> 500ms -> 1s -> 2s -> 5s cap (no intermediate 4s step from naive doubling). Test asserts the spec enumeration including the "stays at 5s forever" cap-after-step-5 invariant. - session_map.rs: removed the dead 'Closing && tap.is_some()' branch from drive_all_sessions. AppState::close already handles teardown inline (fires close_tx, aborts task, clears field, advances state, removes entry). The branch was never exercised and pre-paved a "future peer-initiated close" path -- both AGENTS.md anti-patterns ("don't pre-pave the wrong pattern"). Spec ref: 2026-06-28-slice-2-agent-tap-design.md §4.3 step 4 + §5.1 step 3. --- crates/rutster/src/session_map.rs | 79 +++++++++++++++---------------- crates/rutster/src/tap_engine.rs | 38 ++++++++++----- 2 files changed, 65 insertions(+), 52 deletions(-) diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 25bd4e1..64fad92 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -14,16 +14,20 @@ //! //! # slice-2: TapEngine wiring seam (spec §5.1 step 3) //! -//! `drive_all_sessions` is the spawn/teardown boundary for the per-session -//! TapEngine task. After each `RtcSession::run_poll_once`, we observe the +//! `drive_all_sessions` is the spawn boundary for the per-session TapEngine +//! task. After each `RtcSession::run_poll_once`, we observe the //! `channel.state` transition: //! - `Connected && tap.is_none()` → spawn TapEngine, wire `TapAudioPipe` //! into `RtcSession.pipe` via `with_pipe`, set `channel.tap = Some(...)`. -//! - `Closing && tap.is_some()` → fire the close oneshot, abort the -//! engine task, clear `channel.tap` BEFORE the state advances to Closed. //! -//! `loop_driver.rs` is NOT modified — the spawn/teardown happens in this -//! poll-task layer per spec §8.5 #6. +//! Teardown is NOT a poll-task branch here — all slice-2 teardown happens +//! inline in `AppState::close` (removes the entry, fires `close_tx`, aborts +//! the task, clears `channel.tap`). The future peer-initiated close path +//! (browser `peerconnectionclose`) WILL observe the `Closing` transition +//! here; deferred per slice-2 §1.2. +//! +//! `loop_driver.rs` is NOT modified — the spawn happens in this poll-task +//! layer per spec §8.5 #6. use std::sync::Arc; use std::time::{Duration, Instant}; @@ -123,12 +127,12 @@ impl AppState { self.sessions.get(&id).map(|r| r.rtc.clone()) } - /// Transition to Closing (triggers TapEngine teardown via the poll - /// task on the next cycle), then advance to Closed. The fire-close/ - /// abort happens in `drive_all_sessions` on the `Closing && tap.is_some()` - /// transition — NOT here — because `close` is called from - /// `routes::delete_session` and could race with `drive_all_sessions` - /// (we let the poll task own the TapConn lifecycle consistently). + /// Transition to Closing then Closed. The TapEngine teardown (fire + /// `close_tx`, abort the task, clear `channel.tap`) happens inline + /// here — NOT in `drive_all_sessions`'s poll loop — because `close` + /// is called from `routes::delete_session` and could race with the + /// poll task; doing teardown inline on entry removal keeps the + /// TapConn lifecycle owned by exactly one call site. pub async fn close(&self, id: ChannelId) { if let Some((_id, entry)) = self.sessions.remove(&id) { // If a TapEngine was running, fire the close oneshot + abort @@ -175,14 +179,15 @@ impl Default for AppState { /// One iteration of "drive every active session." Removes closed entries. /// -/// Per spec §5.1 step 3 (slice-2): the poll task is the spawn/teardown -/// boundary for the per-session TapEngine. We observe the channel state -/// AFTER `run_poll_once` returns; the `loop_driver` already wrote the new -/// state. Specifically: +/// Per spec §5.1 step 3 (slice-2): the poll task is the spawn boundary for +/// the per-session TapEngine. We observe the channel state AFTER +/// `run_poll_once` returns; the `loop_driver` already wrote the new state. +/// Specifically: /// - `Connected && tap.is_none()` → spawn TapEngine → wire TapAudioPipe /// into RtcSession via `with_pipe` → set `channel.tap = Some(TapHandle)`. -/// - `Closing && tap.is_some()` → fire close oneshot, abort engine task, -/// clear `channel.tap = None` BEFORE state advances to `Closed`. +/// +/// Teardown is NOT a poll-task branch — see `AppState::close` and the +/// module-level docs for why. /// /// The wiring race (spec §5.1 race note): in the poll cycle that observes /// the `Connected` transition, `MediaData` frames in the same cycle go @@ -204,12 +209,24 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { let mut s = rtc.lock().await; let _ = s.run_poll_once(now); // hot-path match-and-continue inside - // === slice-2: TapEngine spawn/teardown (spec §5.1 step 3, §8.5 #6). === + // === slice-2: TapEngine spawn seam (spec §5.1 step 3, §8.5 #6). === // Observe the state AFTER `loop_driver::drive` mutated `channel.state` - // — we make the engine-control decisions here, NOT inside loop_driver + // — we make the engine-control decision here, NOT inside loop_driver // (the seam test preserves `loop_driver.rs` byte-identical call sites). - match s.channel.state { - ChannelState::Connected if s.channel.tap.is_none() => { + // + // Teardown is intentionally NOT a poll-task branch here. All + // slice-2 teardown happens inline in `AppState::close` (fired from + // `routes::delete_session`): it removes the DashMap entry, fires + // `close_tx`, aborts the engine `JoinHandle`, and clears + // `channel.tap`. By the time this poll task iterates again the + // entry is GONE — a `Closing && tap.is_some()` branch would be + // dead code. The future peer-initiated close path (when slice-2 + // eventually handles a `peerconnectionclose` event from the + // browser) WILL live here, observing the `Closing` transition to + // tear down the engine before the entry is removed — deferred + // per slice-2 §1.2 (no browser-driven close events this slice). + if let ChannelState::Connected = s.channel.state { + if s.channel.tap.is_none() { // First connect: spawn the TapEngine, wire the TapAudioPipe. let (pipe, conn) = spawn_tap_engine(id, tap_url); s.set_pipe(pipe); @@ -223,24 +240,6 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { } continue; } - ChannelState::Closing if s.channel.tap.is_some() => { - // Teardown: fire close + abort + clear tap BEFORE Closed. - // Spec §5.1 step 5 — we send `bye` (TapClient's close arm) - // rather than `session_end`; the brain closes cleanly either - // way per Task 5's echo brain impl. Documented deviation in - // the task report. - s.channel.tap = None; - drop(s); - if let Some(mut entry) = state.sessions.get_mut(&id) { - if let Some(conn) = entry.tap_conn.take() { - let _ = conn.close_tx.send(()); - conn.join.abort(); - } - } - info!(channel_id = %id, "tap engine torn down on Closing"); - continue; - } - _ => {} } if s.is_closed() { diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index f59f9be..85d5a57 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -300,12 +300,22 @@ impl Default for Backoff { impl Backoff { /// Return the current delay and advance the backoff state. - /// Doubles up to the 5 s cap and stays there — infinite retries. + /// + /// Matches the spec §4.3 step 4 enumeration exactly: + /// 250ms → 500ms → 1s → 2s → 5s cap (forever). The "cap at 5s" line + /// means jump directly to 5s after the 2s step — NOT doubling to 4s + /// first and then clamping (that would inject a 4s step the spec + /// doesn't enumerate). The conditional below skips the intermediate + /// 4s value that naïve `(current * 2).min(5s)` would produce. fn next_delay(&mut self) -> Duration { let d = self.current; - // Saturating `min` here guarantees we never exceed the 5 s cap. - // `(current * 2).min(cap)` is the standard "double-up-to-cap" shape. - self.current = (self.current * 2).min(Duration::from_secs(5)); + // Below the 2s step: keep doubling (250 → 500 → 1s → 2s). + // At or above 2s: jump straight to the 5s cap (no 4s step). + self.current = if self.current < Duration::from_secs(2) { + self.current * 2 + } else { + Duration::from_secs(5) + }; self.count += 1; d } @@ -324,14 +334,18 @@ mod tests { #[test] fn backoff_doubles_until_cap() { let mut b = Backoff::default(); - assert_eq!(b.next_delay(), Duration::from_millis(250)); - assert_eq!(b.next_delay(), Duration::from_millis(500)); - assert_eq!(b.next_delay(), Duration::from_secs(1)); - assert_eq!(b.next_delay(), Duration::from_secs(2)); - assert_eq!(b.next_delay(), Duration::from_secs(4)); - // Cap reached: stays at 5s thereafter. - assert_eq!(b.next_delay(), Duration::from_secs(5)); - assert_eq!(b.next_delay(), Duration::from_secs(5)); + // Spec §4.3 step 4 + §5.2 enumeration: + // 250ms → 500ms → 1s → 2s → 5s cap (no intermediate 4s step from + // naïve doubling). Infinite retries at the cap. + assert_eq!(b.next_delay(), Duration::from_millis(250)); // step 1 + assert_eq!(b.next_delay(), Duration::from_millis(500)); // step 2 + assert_eq!(b.next_delay(), Duration::from_secs(1)); // step 3 + assert_eq!(b.next_delay(), Duration::from_secs(2)); // step 4 + assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 5: cap + // Cap invariant: stays at 5s forever (infinite retries, per spec §5.2). + assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 6 + assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 7 + assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 8+ } #[test]