seyd_transport/
session.rs

1// Copyright 2026 Anton Gravestam
2// SPDX-License-Identifier: Apache-2.0
3use crate::relay::Relay;
4use bytes::{Buf, Bytes, BytesMut};
5use std::net::SocketAddr;
6use std::sync::Mutex;
7use std::time::Instant;
8use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
9use tokio::sync::{mpsc, oneshot};
10
11pub type ControlReader = BufReader<Box<dyn tokio::io::AsyncRead + Send + Unpin>>;
12pub type ControlWriter = Box<dyn tokio::io::AsyncWrite + Send + Unpin>;
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
15pub enum SendError {
16    /// Not enough room in the datagram send buffer. Nothing was queued, so the
17    /// caller can skip the whole frame cleanly.
18    #[error("datagram send would block")]
19    Blocked,
20    #[error("datagram exceeds the current path MTU")]
21    TooLarge,
22    #[error("session closed")]
23    Closed,
24}
25
26#[derive(Debug, Clone, Copy, Default, PartialEq)]
27pub struct PathStats {
28    pub rtt_ms: f64,
29    /// Lowest RTT observed by this session's sampling of `rtt`.
30    pub min_rtt_ms: f64,
31    pub cwnd: u64,
32    /// UDP bytes transmitted per second over the last `stats()` interval.
33    pub delivery_kbps: u64,
34    pub lost_packets: u64,
35    pub current_mtu: u16,
36}
37
38/// How the pilot is reached.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum SessionKind {
41    /// A direct WebTransport session over QUIC — the product.
42    Direct,
43    /// Bytes carried through the cloud relay (ADR 0010): the last resort when
44    /// no candidate connected, and always shown to the pilot as such.
45    Relay,
46}
47
48struct RateSample {
49    at: Instant,
50    bytes: u64,
51    min_rtt_ms: f64,
52}
53
54struct Quic {
55    conn: quinn::Connection,
56    /// Quarter-stream-id prefix identifying this session in datagrams.
57    prefix: Bytes,
58    rate: Mutex<RateSample>,
59    /// `TransportConfig::datagram_send_buffer` for this connection, so the
60    /// queued backlog can be recovered from quinn's free-space figure.
61    send_buffer_capacity: usize,
62}
63
64enum Backend {
65    Quic(Quic),
66    Relay(Relay),
67}
68
69/// One session with one pilot: a WebTransport session over QUIC, or, when no
70/// direct path connected, the same datagrams and control stream carried through
71/// the cloud relay. The engine sees one type either way; only [`Session::kind`]
72/// tells them apart, and it is reported to the pilot rather than hidden.
73pub struct Session {
74    id: u64,
75    backend: Backend,
76    datagrams: Mutex<Option<mpsc::Receiver<Bytes>>>,
77    control: Mutex<Option<oneshot::Receiver<(ControlReader, ControlWriter)>>>,
78}
79
80impl Session {
81    pub(crate) fn new(
82        id: u64,
83        conn: quinn::Connection,
84        prefix: Bytes,
85        datagrams: mpsc::Receiver<Bytes>,
86        control: oneshot::Receiver<(ControlReader, ControlWriter)>,
87        send_buffer_capacity: usize,
88    ) -> Self {
89        let bytes = conn.stats().udp_tx.bytes;
90        Self {
91            id,
92            backend: Backend::Quic(Quic {
93                conn,
94                prefix,
95                rate: Mutex::new(RateSample {
96                    at: Instant::now(),
97                    bytes,
98                    min_rtt_ms: f64::MAX,
99                }),
100                send_buffer_capacity,
101            }),
102            datagrams: Mutex::new(Some(datagrams)),
103            control: Mutex::new(Some(control)),
104        }
105    }
106
107    pub(crate) fn from_relay(
108        id: u64,
109        relay: Relay,
110        datagrams: mpsc::Receiver<Bytes>,
111        control: oneshot::Receiver<(ControlReader, ControlWriter)>,
112    ) -> Self {
113        Self {
114            id,
115            backend: Backend::Relay(relay),
116            datagrams: Mutex::new(Some(datagrams)),
117            control: Mutex::new(Some(control)),
118        }
119    }
120
121    /// Open a relayed session through the cloud (ADR 0010). `url`, `session_id`
122    /// and `token` come from the signal server's `relay-open`. Resolves once
123    /// the relay has paired this socket with the pilot's, or fails.
124    pub async fn relay(
125        id: u64,
126        url: &str,
127        session_id: &str,
128        token: &str,
129        send_buffer_capacity: usize,
130    ) -> anyhow::Result<Session> {
131        crate::relay::connect(id, url, session_id, token, send_buffer_capacity).await
132    }
133
134    pub fn session_id(&self) -> u64 {
135        self.id
136    }
137
138    pub fn kind(&self) -> SessionKind {
139        match &self.backend {
140            Backend::Quic(_) => SessionKind::Direct,
141            Backend::Relay(_) => SessionKind::Relay,
142        }
143    }
144
145    /// The pilot's address for a direct session; `None` through the relay,
146    /// where the only peer address is the cloud's.
147    pub fn remote_addr(&self) -> Option<SocketAddr> {
148        match &self.backend {
149            Backend::Quic(q) => Some(q.conn.remote_address()),
150            Backend::Relay(_) => None,
151        }
152    }
153
154    /// Largest payload `send_datagram` currently accepts (path MTU minus QUIC
155    /// and WebTransport framing). Changes as DPLPMTUD converges.
156    pub fn max_datagram_size(&self) -> usize {
157        match &self.backend {
158            Backend::Quic(q) => q
159                .conn
160                .max_datagram_size()
161                .unwrap_or(0)
162                .saturating_sub(q.prefix.len()),
163            Backend::Relay(r) => r.max_datagram_size(),
164        }
165    }
166
167    /// Bytes the send buffer can take right now without dropping anything.
168    /// Compare a whole frame's size against this *before* sending its first chunk.
169    pub fn send_buffer_space(&self) -> usize {
170        match &self.backend {
171            Backend::Quic(q) => q
172                .conn
173                .datagram_send_buffer_space()
174                .saturating_sub(q.prefix.len()),
175            Backend::Relay(r) => r.send_buffer_space(),
176        }
177    }
178
179    /// Payload bytes queued in the datagram send buffer and not yet packetised:
180    /// the backlog that admission control measures against the QoS profile's
181    /// threshold. Everything counted here is latency the pilot has not seen yet.
182    ///
183    /// quinn exposes only the free space (capacity minus queued payload, minus
184    /// one bookkeeping constant), so this is the complement; the constant makes
185    /// it read a few dozen bytes high, which is nothing against a frame.
186    pub fn send_buffer_queued(&self) -> usize {
187        match &self.backend {
188            Backend::Quic(q) => q
189                .send_buffer_capacity
190                .saturating_sub(q.conn.datagram_send_buffer_space()),
191            Backend::Relay(r) => r.send_buffer_queued(),
192        }
193    }
194
195    /// Queue one datagram. Never blocks, never evicts queued datagrams.
196    pub fn send_datagram(&self, payload: Bytes) -> Result<(), SendError> {
197        match &self.backend {
198            Backend::Quic(q) => {
199                let total = payload.len() + q.prefix.len();
200                if q.conn.datagram_send_buffer_space() < total {
201                    return Err(SendError::Blocked);
202                }
203                let mut buf = BytesMut::with_capacity(total);
204                buf.extend_from_slice(&q.prefix);
205                buf.extend_from_slice(&payload);
206                q.conn.send_datagram(buf.freeze()).map_err(|e| match e {
207                    quinn::SendDatagramError::TooLarge => SendError::TooLarge,
208                    quinn::SendDatagramError::ConnectionLost(_) => SendError::Closed,
209                    quinn::SendDatagramError::UnsupportedByPeer
210                    | quinn::SendDatagramError::Disabled => SendError::Closed,
211                })
212            }
213            Backend::Relay(r) => r.send_datagram(payload),
214        }
215    }
216
217    /// Inbound datagrams (WebTransport prefix stripped). Can be taken once.
218    pub fn take_datagrams(&self) -> Option<mpsc::Receiver<Bytes>> {
219        self.datagrams.lock().unwrap().take()
220    }
221
222    /// The first bidirectional stream the pilot opens — the control stream.
223    /// Resolves once the pilot has opened it; `None` if the session ends first.
224    pub async fn control(&self) -> Option<(ControlReader, ControlWriter)> {
225        let rx = self.control.lock().unwrap().take()?;
226        rx.await.ok()
227    }
228
229    pub fn stats(&self) -> PathStats {
230        match &self.backend {
231            Backend::Quic(q) => {
232                let s = q.conn.stats();
233                let now = Instant::now();
234                let mut rate = q.rate.lock().unwrap();
235                let rtt_ms = s.path.rtt.as_secs_f64() * 1000.0;
236                if rtt_ms > 0.0 && rtt_ms < rate.min_rtt_ms {
237                    rate.min_rtt_ms = rtt_ms;
238                }
239                let dt = now.duration_since(rate.at).as_secs_f64();
240                let delivery_kbps = if dt > 0.05 {
241                    ((s.udp_tx.bytes.saturating_sub(rate.bytes)) as f64 * 8.0 / dt / 1000.0) as u64
242                } else {
243                    0
244                };
245                if dt > 0.05 {
246                    rate.at = now;
247                    rate.bytes = s.udp_tx.bytes;
248                }
249                PathStats {
250                    rtt_ms,
251                    min_rtt_ms: if rate.min_rtt_ms == f64::MAX {
252                        rtt_ms
253                    } else {
254                        rate.min_rtt_ms
255                    },
256                    cwnd: s.path.cwnd,
257                    delivery_kbps,
258                    lost_packets: s.path.lost_packets,
259                    current_mtu: s.path.current_mtu,
260                }
261            }
262            Backend::Relay(r) => r.stats(),
263        }
264    }
265
266    pub fn close(&self, code: u32, reason: &str) {
267        match &self.backend {
268            Backend::Quic(q) => q.conn.close(code.into(), reason.as_bytes()),
269            Backend::Relay(r) => r.close(reason),
270        }
271    }
272
273    /// Resolves when the connection is gone.
274    pub async fn closed(&self) {
275        match &self.backend {
276            Backend::Quic(q) => {
277                let _ = q.conn.closed().await;
278            }
279            Backend::Relay(r) => r.closed().await,
280        }
281    }
282}
283
284/// NDJSON helpers on the control stream halves.
285pub async fn read_line(reader: &mut ControlReader) -> Option<String> {
286    let mut line = String::new();
287    loop {
288        line.clear();
289        match reader.read_line(&mut line).await {
290            Ok(0) | Err(_) => return None,
291            Ok(_) => {
292                let t = line.trim_end_matches(['\n', '\r']);
293                if !t.is_empty() {
294                    return Some(t.to_string());
295                }
296            }
297        }
298    }
299}
300
301pub async fn write_line(writer: &mut ControlWriter, line: &str) -> std::io::Result<()> {
302    writer.write_all(line.as_bytes()).await?;
303    writer.write_all(b"\n").await?;
304    writer.flush().await
305}
306
307/// Split a QUIC varint-prefixed WebTransport datagram into (quarter stream id, payload).
308pub(crate) fn split_prefix(mut buf: Bytes) -> Option<(u64, Bytes)> {
309    if buf.is_empty() {
310        return None;
311    }
312    let first = buf[0];
313    let len = 1usize << (first >> 6);
314    if buf.len() < len {
315        return None;
316    }
317    let mut v: u64 = (first & 0x3f) as u64;
318    for i in 1..len {
319        v = (v << 8) | buf[i] as u64;
320    }
321    buf.advance(len);
322    Some((v, buf))
323}
324
325/// Encode a QUIC varint.
326pub(crate) fn varint(v: u64) -> Bytes {
327    let mut out = BytesMut::new();
328    if v < 1 << 6 {
329        out.extend_from_slice(&[v as u8]);
330    } else if v < 1 << 14 {
331        out.extend_from_slice(&((v as u16) | 0x4000).to_be_bytes());
332    } else if v < 1 << 30 {
333        out.extend_from_slice(&((v as u32) | 0x8000_0000).to_be_bytes());
334    } else {
335        out.extend_from_slice(&(v | 0xc000_0000_0000_0000).to_be_bytes());
336    }
337    out.freeze()
338}
339
340#[cfg(test)]
341mod tests {
342    use super::*;
343
344    #[test]
345    fn varint_roundtrip() {
346        for v in [0u64, 1, 63, 64, 16383, 16384, 1 << 29, 1 << 30, 1 << 40] {
347            let mut b = BytesMut::from(&varint(v)[..]);
348            b.extend_from_slice(b"xyz");
349            let (got, rest) = split_prefix(b.freeze()).unwrap();
350            assert_eq!(got, v);
351            assert_eq!(&rest[..], b"xyz");
352        }
353    }
354}
seyd_transport/session.rs

seyd_transport/
session.rs
