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:
 

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<S>(SplitSink<WebSocketStream<S>, Message>);
+type Sink<S> = Arc<Mutex<SplitSink<WebSocketStream<S>, Message>>>;
+
+struct WsTx<S>(Sink<S>);
 struct WsRx<S>(SplitStream<WebSocketStream<S>>);
 
 #[async_trait]
 impl<S: AsyncRead + AsyncWrite + Unpin + Send + 'static> PipeTx for WsTx<S> {
     async fn send(&mut self, msg: Vec<u8>) -> 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<S: AsyncRead + AsyncWrite + Unpin + Send + 'static> PipeRx for WsRx<S> {
     }
 }
 
-/// A MOQT session over an upgraded WebSocket.
+/// A MOQT session over an upgraded WebSocket, pinging the peer every [`PING_EVERY`].
 pub fn session<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(ws: WebSocketStream<S>, 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<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
+    ws: WebSocketStream<S>,
+    client: bool,
+    remote: String,
+    cfg: MuxConfig,
+    every: Duration,
+) -> MuxConnection {
     let (tx, rx) = ws.split();
+    let tx: Sink<S> = 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)))
 }
 

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");
+}