diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/README.md b/joule-code/joule-code-rs/crates/joule-moq-relay/README.md index 9bfe22f8b..93260a8cf 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/README.md +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/README.md @@ -164,6 +164,11 @@ CONNECT stream, so that a client written for one is most of the way to the other with `0x045D4487`. 7. **Closing.** `WT_CLOSE_SESSION` with a MOQT session termination code (§12.2), then the WebSocket's close. A WebSocket that closes without one is a session closed with code 0. +8. **Keeping alive.** Each end the relay runs, accepting or dialling, sends a WebSocket + ping (RFC 6455 §5.5.2) every 30 seconds, and the peer's WebSocket answers each with a + pong. A front between the two that closes a connection it sees idle (a tunnel, a proxy) + then keeps a session that has nothing to carry, such as a relay's link to its upstream + while no one watches. The first capsules a relay with the default limits sends are its credit: diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/websocket.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/websocket.rs index 8454255c4..00b30dfdc 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/websocket.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/websocket.rs @@ -9,32 +9,44 @@ //! - every binary message is exactly one capsule of `draft-ietf-webtrans-http2-15` §6, and //! the session is what a WebTransport session over HTTP/2 would carry ([`super::mux`]); //! - MOQT's control, request and data streams ride those streams unchanged: the same -//! bytes a QUIC stream would carry. +//! bytes a QUIC stream would carry; +//! - each end pings the other every [`PING_EVERY`] (RFC 6455 §5.5.2), so a front that closes +//! a connection it sees idle keeps a session with nothing to carry. use super::mux::{MuxConfig, MuxConnection, PipeRx, PipeTx}; use super::{TResult, TransportError, TransportKind}; use async_trait::async_trait; use futures::stream::{SplitSink, SplitStream}; use futures::{SinkExt, StreamExt}; +use std::sync::Arc; +use std::time::Duration; use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::sync::Mutex; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::WebSocketStream; +/// How often each end of a session pings the other. A front that closes idle connections +/// (a tunnel closed a relay's idle upstream link every two minutes) sees a ping as traffic, +/// and the peer's WebSocket answers each with a pong. +pub const PING_EVERY: Duration = Duration::from_secs(30); + /// The subprotocol that selects this draft. pub fn subprotocol() -> &'static str { std::str::from_utf8(joule_moq::draft22::ALPN).expect("the ALPN is ASCII") } -struct WsTx(SplitSink, Message>); +type Sink = Arc, Message>>>; + +struct WsTx(Sink); struct WsRx(SplitStream>); #[async_trait] impl PipeTx for WsTx { async fn send(&mut self, msg: Vec) -> TResult<()> { - self.0.send(Message::Binary(msg.into())).await.map_err(TransportError::other) + self.0.lock().await.send(Message::Binary(msg.into())).await.map_err(TransportError::other) } async fn close(&mut self) { - let _ = self.0.close().await; + let _ = self.0.lock().await.close().await; } } @@ -54,9 +66,34 @@ impl PipeRx for WsRx { } } -/// A MOQT session over an upgraded WebSocket. +/// A MOQT session over an upgraded WebSocket, pinging the peer every [`PING_EVERY`]. pub fn session(ws: WebSocketStream, client: bool, remote: String, cfg: MuxConfig) -> MuxConnection { + session_pinging(ws, client, remote, cfg, PING_EVERY) +} + +/// [`session`], pinging the peer every `every`. The pings stop when the session's sender is +/// dropped or a ping cannot be sent. +pub fn session_pinging( + ws: WebSocketStream, + client: bool, + remote: String, + cfg: MuxConfig, + every: Duration, +) -> MuxConnection { let (tx, rx) = ws.split(); + let tx: Sink = Arc::new(Mutex::new(tx)); + let pinger = Arc::downgrade(&tx); + tokio::spawn(async move { + let mut tick = tokio::time::interval_at(tokio::time::Instant::now() + every, every); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let Some(tx) = pinger.upgrade() else { break }; + if tx.lock().await.send(Message::Ping(Default::default())).await.is_err() { + break; + } + } + }); MuxConnection::new(client, TransportKind::WebSocket, remote, cfg, Box::new(WsTx(tx)), Box::new(WsRx(rx))) } diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/relay_websocket.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/relay_websocket.rs index 0e463b211..96729f5be 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/relay_websocket.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/relay_websocket.rs @@ -9,3 +9,34 @@ async fn two_publishers_three_subscribers_over_websocket() { .await .expect("the scenario finishes in time"); } + +/// A session with nothing to carry pings its peer, so a front that closes idle connections +/// keeps it; the peer's pongs come back and the session stays open. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn an_idle_session_pings_its_peer() { + use futures::StreamExt; + use joule_moq_relay::transport::mux::MuxConfig; + use joule_moq_relay::transport::websocket::session_pinging; + use std::time::Duration; + use tokio_tungstenite::tungstenite::protocol::Role; + use tokio_tungstenite::tungstenite::Message; + use tokio_tungstenite::WebSocketStream; + + let (relay_end, peer_end) = tokio::io::duplex(1 << 16); + let ws = WebSocketStream::from_raw_socket(relay_end, Role::Server, None).await; + let mut peer = WebSocketStream::from_raw_socket(peer_end, Role::Client, None).await; + let _session = session_pinging(ws, false, "peer".into(), MuxConfig::default(), Duration::from_millis(50)); + + let mut pings = 0; + let read = tokio::time::timeout(Duration::from_secs(10), async { + while pings < 3 { + match peer.next().await { + Some(Ok(Message::Ping(_))) => pings += 1, + Some(Ok(_)) => {} + other => panic!("the session ended before its pings: {other:?}"), + } + } + }) + .await; + assert!(read.is_ok(), "three pings at a 50 ms interval within 10 s; {pings} arrived"); +}