alee/rutster
0
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(trunk): TrunkSession + trunk_driver::drive -- FOB-side trunk-leg tick (slice-5 T4)

Browse Source 
The trunk leg's tick function. Parallels crates/rutster-media/src/loop_driver.rs
minus the str0m/Opus/RTP machinery -- there is no RTP to decode, no Opus to
encode, no str0m poll loop to drain. Caller->FOB direction is a pure mpsc
drain; FOB->caller direction is a mpsc push.

Slice-4's Reflex<P> + LocalVadReflex stack compose identically around the
trunk leg's session.pipe -- proving the FOB reflex loop is ingress-agnostic
(spec §2.3 -- the architecture's load-bearing claim). TrunkSession is generic
over P: AudioPipe + Send so unit tests substitute EchoAudioPipe without
constructing a full TapEngine wiring harness; production uses TapAudioPipe.

The seam gate holds: crates/rutster-media/src/{loop_driver.rs,rtc_session.rs}
stay byte-identical because the trunk leg NEVER enters that code path. The
MediaThread dispatches via the new MediaLeg enum (T5).

Signed-off-by: Aaron D. Lee <himself@adlee.work>
This commit is contained in:
A.D.Lee
2026-07-05 03:31:22 -04:00
parent c7d48559a5
commit caf4c5ba91
3 changed files with 366 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
crates/rutster-trunk/src/lib.rs
View File

@@ -8,61 +8,43 @@
//! It owns **no SIP stack**: SIP lives outside the trust boundary, in the rented transport
//! (green zone — [ADR-0008](../../../docs/adr/0008-fob-and-green-zone.md)).
//!
//! ## Spearhead step 5 — what's here vs what's deferred
//! ## Spearhead step 5 — what's here
//!
//! The provider call-control seam ([`provider`]) lands in T2: the
//! `CallControlClient` trait, `MockCallControlClient`, and `TwilioCredentials` (with
//! redacted `Debug`). The live `TwilioCallControlClient` (T6) is feature-gated behind
//! `twilio-live`. The WSS ingress (`g711`, `twilio_media_streams`, `session`,
//! `loop_driver`) lands in the parallel FOB-side tasks (T1/T3/T4) — see the slice-5 spec.
//! FOB-side (dev-c, T1/T3/T4/T5):
//! * `g711` — G711Codec (µ-law encode/decode + 8kHz↔24kHz resampling).
//! * `twilio_media_streams` — TwilioMediaStreamsServer (axum WSS handler).
//! * `session` — TrunkSession (per-trunk-leg struct).
//! * `loop_driver` — trunk_driver::drive (per-tick function).
//!
//! Green-zone (dev-b, T2/T6):
//! * `provider/{mod,mock,twilio}` — CallControlClient trait + Mock + live
//! TwilioCallControlClient (behind `twilio-live` feature).
//!
//! ## Dependency direction
//!
//! `rutster-trunk` → `rutster-call-model` + `rutster-media` (once the FOB-side media path
//! lands): PSTN audio is resampled to the canonical tap format and handed to the media plane
//! `rutster-trunk` → `rutster-call-model` + `rutster-media` + `rutster-tap`:
//! PSTN audio is resampled to the canonical tap format and handed to the media plane
//! as a media-leg ingress, parallel to WebRTC. The provider (call-control) side is
//! green-zone and depends on no FOB media types — only on `url` + `async-trait` (T2).
pub mod provider;
// slice-5 fills in the FOB-side rented-transport ingress. Module map:
//
// * g711 -- G711Codec (µ-law encode/decode + 8kHz<->24kHz
// linear-interpolated resampling). T1.
// * mulaw_decode_table -- 256-entry compile-time-generated ITU-T G.711
// µ-law decode table. T1.
// * mulaw_encode_table -- 65536-entry compile-time-generated ITU-T G.711
// µ-law encode table. T1.
//
// Future modules landing in subsequent tasks (slice-5 dev-c chain):
// * twilio_media_streams -- TwilioMediaStreamsServer (axum WSS handler). T3.
// * session -- TrunkSession (per-trunk-leg session struct). T4.
// * loop_driver -- trunk_driver::drive (per-tick function). T4.
//
// Green-zone modules (slice-5 dev-b branch):
// * provider/{mod,mock,twilio} -- CallControlClient trait + Mock + live
// TwilioCallControlClient (behind `twilio-live` feature). T2/T6 -- lands
// in dev-b's branch, rebase-merges here at PR time.
pub mod g711;
pub mod loop_driver;
mod mulaw_decode_table;
mod mulaw_encode_table;
pub mod provider;
pub mod session;
pub mod twilio_media_streams;
#[cfg(test)]
mod tests {
use crate::provider::MockCallControlClient;
use crate::g711::G711Codec;
use crate::provider::MockCallControlClient;
/// Stub crates lock boundaries; the compile-test is the lock. Now that
/// the provider module (T2, dev-b) + FOB modules (T1, dev-c) both land,
/// this test guards the crate-level re-exports compile against the
/// combined public API surface.
/// Stub boundary smoke test, retained after the FOB modules land. Also
/// exercises the public codec + provider API surface once.
#[test]
fn crate_compiles() {
// Touch the provider public API (T2).
let _mock = MockCallControlClient::new();
// Touch the codec public API (T1).
let frame = G711Codec::decode_mulaw_to_pcm(&[0xFF; 160]);
assert_eq!(frame.samples.len(), 480);
}

193
crates/rutster-trunk/src/loop_driver.rs Normal file
View File

@@ -0,0 +1,193 @@
//! # trunk_driver::drive -- the FOB-side trunk-leg tick function (spec §3.3)
//!
//! Parallels `crates/rutster-media/src/loop_driver.rs` minus the str0m/Opus/
//! RTP machinery. There is no RTP to decode, no Opus to encode, no str0m poll
//! loop to drain. The caller->FOB direction is a pure mpsc drain; the
//! FOB->caller direction is a mpsc push.
//!
//! # The seam (sacred -- slice-4 Task 10 pinned-blob CI gate)
//!
//! This function is a SEPARATE file in a SEPARATE crate. `loop_driver.rs`
//! in `rutster-media` stays byte-identical because the trunk leg never
//! enters that code path. The MediaThread's tick (T5) dispatches via the
//! `MediaLeg` enum:
//!
//! ```ignore
//! match leg {
//! MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now), // unchanged slice-4
//! MediaLeg::Trunk(s) => rutster_trunk::loop_driver::drive(s, now), // NEW this slice
//! }
//! ```
use std::time::{Duration, Instant};
use tracing::{info, warn};
use crate::session::{TrunkSession, IDLE_TIMEOUT, OUTBOUND_TICK};
/// One iteration of the trunk-leg driver. Mirrors `rutster_media::loop_driver::drive`
/// minus str0m. Returns the sleep-until-next-tick budget.
///
/// # Hot-path policy (AGENTS.md "Error handling")
///
/// Never `?`-propagate. `try_recv` / `try_send` return `Err` on empty/full,
/// neither of which halts the tick -- log + move on. A dropped caller PCM
/// frame is recoverable; a crashed std thread is not.
pub fn drive<P: rutster_media::AudioPipe + Send>(
session: &mut TrunkSession<P>,
now: Instant,
) -> Option<Duration> {
// === Step 1: drain inbound caller PCM from Twilio ===
// (The WebRTC equivalent reads UDP + str0m::Input::Receive; the trunk
// equivalent drains a tokio mpsc the WSS pump task fills.)
while let Ok(frame) = session.inbound_from_twilio_rx.try_recv() {
// Push the caller's PCM through the wrapped AudioPipe stack:
// LocalVadReflex::on_pcm_frame inspects for VAD (slice-4 §3.4 --
// primary trigger path), then delegates inward -> Reflex::
// on_pcm_frame -> the inner P (TapAudioPipe in production) ->
// tx_pcm_in -> TapEngine -> brain WS as `audio_in`.
session.pipe.on_pcm_frame(frame);
session.last_idle_rx = now;
}
// === Step 2: outbound encode tick (every 20 ms) ===
if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK {
if let Some(reply_frame) = session.pipe.next_pcm_frame() {
// Brain reply PCM; push to the WSS pump task for µ-law encode
// + Twilio Media Streams JSON "media" frame emission. try_send;
// drop + observe on full (hot-path policy).
if session
.outbound_to_twilio_tx
.try_send(reply_frame)
.is_err()
{
warn!(
channel_id = %session.channel.id,
"outbound_to_twilio_tx full; dropping brain reply frame"
);
}
}
session.last_outbound_at = now;
}
// === Step 3: idle timeout (mirrors slice-1 §4.5) ===
if now.duration_since(session.last_idle_rx) > IDLE_TIMEOUT {
info!(
channel_id = %session.channel.id,
"trunk idle timeout (60s no RX); closing session"
);
session.channel.state = rutster_call_model::ChannelState::Closed;
return None;
}
// === Step 4: sleep budget ===
// The std thread's meta-tick is 10 ms; this duration tells the next
// tick's worth of sleep (mirrors loop_driver::drive's contract).
session.next_timeout = Some(now + OUTBOUND_TICK);
Some(OUTBOUND_TICK)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::session::tests::trunk_session_with_echo_pipe_and_mpsc;
use rutster_media::PcmFrame;
#[tokio::test]
async fn tick_drains_inbound_mpsc_and_pushes_to_pipe_sink() {
// Given: a TrunkSession with an EchoAudioPipe inner + an inbound
// mpsc carrying one PCM frame.
let now = Instant::now();
let (mut session, inbound_tx, _out_rx) =
trunk_session_with_echo_pipe_and_mpsc(now);
let mut frame = PcmFrame::zeroed();
frame.samples[0] = 4242;
inbound_tx.send(frame).await.unwrap();
// When: drive() runs one tick.
let _ = drive(&mut session, now);
// Then: the EchoAudioPipe's internal queue holds the frame (and
// the VAD path's RMS threshold for a constant 4242 amplitude is
// below the barge-in trigger threshold, so the frame passes
// through unmuted). We verify by reading it back via the same
// pipe's `next_pcm_frame` -- but `next_pcm_frame` is downstream
// of the LocalVadReflex's muted state, so we pull the inner
// pipe's queue contents indirectly by sending a fresh formatted
// frame. (Slice-4 §3.4 covers the unmuted drain identity held
// when caller energy is below VAD_RMS_THRESHOLD.)
let out_frame = session.pipe.next_pcm_frame();
// The EchoAudioPipe holds exactly one frame (the one we pushed).
// If the tick route worked, we get Some back; if the inbound
// mpsc was never drained, we get None (the echopipe is empty).
assert!(out_frame.is_some(), "tick must drain inbound mpsc to the pipe");
}
#[tokio::test]
async fn tick_pulls_from_pipe_source_and_pushes_to_outbound() {
// Given: a TrunkSession with an EchoAudioPipe inner that already
// holds a frame (pushed via on_pcm_frame), and the outbound mpsc
// ready to receive.
let now = Instant::now();
let (mut session, _in_tx, mut out_rx) =
trunk_session_with_echo_pipe_and_mpsc(now);
// Prime the inner echo pipe with a frame.
let mut frame = PcmFrame::zeroed();
frame.samples[0] = 9001;
session.pipe.on_pcm_frame(frame);
// Force the outbound tick to fire (last_outbound_at = now - 100ms).
session.last_outbound_at = now - Duration::from_millis(100);
// When: drive() runs one tick.
let _ = drive(&mut session, now);
// Then: outbound mpsc has the frame (try_recv returns it).
let drained = out_rx.try_recv();
assert!(drained.is_ok(), "tick must push brain reply PCM to outbound mpsc");
}
#[tokio::test]
async fn idle_timeout_closes_session_after_60s_silence() {
// Given: a session whose last_idle_rx is 90s in the past.
let now = Instant::now();
let (mut session, _in_tx, _out_rx) =
trunk_session_with_echo_pipe_and_mpsc(now);
session.last_idle_rx = now - Duration::from_secs(90);
// When: drive() runs one tick.
let next = drive(&mut session, now);
// Then: the channel is Closed and the tick returns None (halting
// the std thread's tick loop for this leg).
assert!(matches!(
session.channel.state,
rutster_call_model::ChannelState::Closed
));
assert_eq!(next, None, "drive returns None on close");
}
#[tokio::test]
async fn drive_advances_last_idle_rx_on_inbound_frame() {
// Given: a TrunkSession with an inbound frame ready; last_idle_rx
// is currently in the past.
let now = Instant::now();
let (mut session, inbound_tx, _out_rx) =
trunk_session_with_echo_pipe_and_mpsc(now);
session.last_idle_rx = now - Duration::from_secs(30);
inbound_tx.send(PcmFrame::zeroed()).await.unwrap();
// When: drive() runs one tick.
let _ = drive(&mut session, now);
// Then: last_idle_rx advanced to now (idle timeout didn't fire).
assert_eq!(session.last_idle_rx, now);
assert!(matches!(
session.channel.state,
rutster_call_model::ChannelState::New
));
}
}

155
crates/rutster-trunk/src/session.rs Normal file
View File

@@ -0,0 +1,155 @@
//! # TrunkSession -- the per-trunk-leg session (spec §3.3)
//!
//! Parallels slice-1's `RtcSession` minus the str0m/Opus/UDP-socket footprint.
//! The wrapped pipe stack is REUSED from slice-4 verbatim:
//! `LocalVadReflex<Reflex<P>>`, where `P: AudioPipe + Send` is the inner pipe
//! (production: `TapAudioPipe`; tests: `EchoAudioPipe`). The composition site
//! (MediaThread's `RegisterTrunk` handler -- T5) constructs the full stack.
use std::time::{Duration, Instant};
use rutster_call_model::Channel;
use rutster_media::PcmFrame;
use tokio::sync::mpsc;
/// 60 s of silence (no inbound PCM from Twilio's "media" frames) means
/// "the call leg is dead" -- the WSS pump task has likely crashed or the
/// Twilio Media Streams fork has been silently closed by the cloud. Same
/// threshold as slice-1's RtcSession idle timeout (spec §4.5).
pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
/// 20 ms outbound encode tick (matches slice-1's PCM frame size: 480 samples
/// @ 24 kHz = 20 ms). The trunk driver pulls a single PCM frame from the
/// pipe's `next_pcm_frame` source on each tick onset + pushes to the
/// outbound WSS mpsc. Mirrors `rutster_media::loop_driver::OUTBOUND_TICK`.
pub const OUTBOUND_TICK: Duration = Duration::from_millis(20);
/// The per-trunk-leg session. Generic over `P: AudioPipe + Send` so tests
/// substitute a stub inner pipe (e.g. `EchoAudioPipe`) without constructing
/// the full TapEngine wiring (TapConn, WS handshake, mpsc pairs).
///
/// # Production type instantiation
///
/// `TrunkSession<TapAudioPipe>` -- MediaThread::RegisterTrunk (T5)
/// composes:
/// 1. `let (tap_pipe, tap_conn) = spawn_tap_engine(...)` (slice-2).
/// 2. `let reflex = Reflex::new(tap_pipe, advisory_rx, metrics)` (slice-4).
/// 3. `let wrapped = LocalVadReflex::new(reflex, advisory_tx)` (slice-4).
/// 4. `TrunkSession { pipe: wrapped, ... }`.
///
/// The `(P: AudioPipe + Send)` bound mirrors the slice-4 trait constraints:
/// the std thread holds the session behind `&mut TrunkSession` exclusively,
/// but Send is required when the binary moves sessions across thread-task
/// boundaries for teardown or migration.
pub struct TrunkSession<P: rutster_media::AudioPipe + Send> {
/// The call-channel the FOB created for this trunk leg.
pub channel: Channel,
/// The wrapped reflex stack: `LocalVadReflex<Reflex<P>>`. Constructed
/// by MediaThread::RegisterTrunk (T5), REUSED slice-4 task-6 composition.
pub pipe: rutster_media::LocalVadReflex<rutster_media::Reflex<P>>,
/// Caller PCM from Twilio (decoded at 24 kHz by the WSS pump task).
pub inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
/// FOB outbound PCM for Twilio (the WSS pump encodes µ-law + sends back).
pub outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
/// Last Instant a caller PCM frame was received. Used to detect the
/// 60 s idle timeout (WSS pump gone silent -> close).
pub last_idle_rx: Instant,
/// Cached next-deadline Instant (mirrors `RtcSession::next_timeout`).
pub next_timeout: Option<Instant>,
/// Last Instant we wrote an outbound PCM frame (mirrors
/// `RtcSession::last_outbound_at`). Paces the 20 ms outbound tick.
pub last_outbound_at: Instant,
}
impl<P: rutster_media::AudioPipe + Send> TrunkSession<P> {
/// Construct a session with the provided wrapped pipe + mpsc pair. The
/// MediaThread::RegisterTrunk (T5) handler owns this construction site
/// in production; tests call this directly with stub pipes.
pub fn new(
channel: Channel,
pipe: rutster_media::LocalVadReflex<rutster_media::Reflex<P>>,
inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
now: Instant,
) -> Self {
Self {
channel,
pipe,
inbound_from_twilio_rx,
outbound_to_twilio_tx,
last_idle_rx: now,
next_timeout: None,
last_outbound_at: now,
}
}
/// Slice-4's `ChannelState::Closed` predicate (mirrors the slice-1
/// `RtcSession::is_closed` API). Used by the std thread's tick loop to
/// evict closed sessions from the session map.
pub fn is_closed(&self) -> bool {
matches!(self.channel.state, rutster_call_model::ChannelState::Closed)
}
}
#[cfg(test)]
pub mod tests {
use super::*;
use rutster_media::{EchoAudioPipe, Reflex, ReflexMetrics, LocalVadReflex};
use tokio::sync::mpsc;
/// Build a `TrunkSession<EchoAudioPipe>` with no real brain wiring
/// (no TapEngine, no WS endpoint). The materialized wrapped pipe is the
/// same composition the FOB constructs in production:
/// `LocalVadReflex<Reflex<EchoAudioPipe>>`. The advisory mpsc pair is
/// fresh and never receives events in these tests.
pub fn trunk_session_with_echo_pipe(now: Instant) -> TrunkSession<EchoAudioPipe> {
let (advisory_tx, advisory_rx) = mpsc::channel(16);
let metrics = ReflexMetrics::new();
let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics);
let wrapped = LocalVadReflex::new(reflex, advisory_tx);
let (inbound_tx, inbound_rx) = mpsc::channel(16);
let (outbound_tx, outbound_rx) = mpsc::channel(16);
let channel = Channel::new_inbound();
let mut session = TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now);
// Expose the tx side to the test via a leak into a multi-call test.
session.last_idle_rx = now;
session
}
/// EchoAudioPipe ownership partner of [`trunk_session_with_echo_pipe`]:
/// returns the sender + receiver ends separately for test scenarios.
pub fn trunk_session_with_echo_pipe_and_mpsc(
now: Instant,
) -> (
TrunkSession<EchoAudioPipe>,
mpsc::Sender<PcmFrame>,
mpsc::Receiver<PcmFrame>,
) {
let (advisory_tx, advisory_rx) = mpsc::channel(16);
let metrics = ReflexMetrics::new();
let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics);
let wrapped = LocalVadReflex::new(reflex, advisory_tx);
let (inbound_tx, inbound_rx) = mpsc::channel(16);
let (outbound_tx, outbound_rx) = mpsc::channel(16);
let channel = Channel::new_inbound();
let session = TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now);
(session, inbound_tx, outbound_rx)
}
#[test]
fn new_session_starts_in_new_state() {
let now = Instant::now();
let session = trunk_session_with_echo_pipe(now);
assert!(matches!(
session.channel.state,
rutster_call_model::ChannelState::New
));
}
#[test]
fn is_closed_false_for_fresh_session() {
let now = Instant::now();
let session = trunk_session_with_echo_pipe(now);
assert!(!session.is_closed());
}
}