Datagram packet loss due to "last-write-wins" pattern in DatagramsWriter

The current implementation of DatagramsWriter and DatagramsReader in moq-transport/src/serve/datagram.rs uses a "last-write-wins" pattern that causes severe packet loss
when datagrams arrive faster than they can be consumed.

Problem

The DatagramsState struct only stores a single datagram:

struct DatagramsState {
latest: Option, // ← Only ONE datagram!
epoch: u64,
closed: Result<(), ServeError>,
}

When write() is called, it overwrites the previous datagram:

pub fn write(&mut self, datagram: Datagram) -> Result<(), ServeError> {
let mut state = self.state.lock_mut().ok_or(ServeError::Cancel)?;
state.latest = Some(datagram); // ← Overwrites previous!
state.epoch += 1;
Ok(())
}

Impact: If 4 datagrams arrive before the reader processes one, 3 are silently lost. In real-world video streaming scenarios (30fps H.264 with multiple RTP packets per
frame), this causes ~75% packet loss.

Reproduction

1. Use moq-relay-ietf as a relay
2. Publish video using datagrams at 30fps
3. Subscribe from a browser player
4. Observe that only ~25% of datagrams are received

Solution

Replace the single-slot latest: Option with a bounded FIFO queue (VecDeque).

Key changes:

1. Writer: Appends to queue instead of overwriting
2. Reader: Tracks absolute position via consumed_count
3. Buffer overflow: When queue reaches max size (1024), drop oldest datagram
4. Multi-reader support: Each reader has independent position tracking

Proposed Fix

use std::{collections::VecDeque, fmt, sync::Arc};
use crate::watch::State;
use super::{ServeError, Track};

const MAX_DATAGRAM_BUFFER: usize = 1024;

struct DatagramsState {
queue: VecDeque, // FIFO queue instead of single slot
write_count: u64, // Total datagrams written
dropped_count: u64, // Datagrams dropped due to buffer overflow
closed: Result<(), ServeError>,
}

impl Default for DatagramsState {
fn default() -> Self {
Self {
queue: VecDeque::with_capacity(256),
write_count: 0,
dropped_count: 0,
closed: Ok(()),
}
}
}

impl DatagramsWriter {
pub fn write(&mut self, datagram: Datagram) -> Result<(), ServeError> {
let mut state = self.state.lock_mut().ok_or(ServeError::Cancel)?;

// Drop oldest if buffer full
if state.queue.len() >= MAX_DATAGRAM_BUFFER {
state.queue.pop_front();
state.dropped_count += 1;
}

state.queue.push_back(datagram);
state.write_count += 1;
Ok(())
}
}

#[derive(Clone)]
pub struct DatagramsReader {
state: State,
pub track: Arc,
consumed_count: u64, // Absolute position (handles multi-reader)
}

impl DatagramsReader {
fn new(state: State, track: Arc) -> Self {
let initial_dropped = {
let state = state.lock();
state.dropped_count
};
Self {
state,
track,
consumed_count: initial_dropped,
}
}

pub async fn read(&mut self) -> Result, ServeError> {
loop {
{
let state = self.state.lock();

// Calculate queue index from absolute position
let queue_index = if self.consumed_count >= state.dropped_count {
(self.consumed_count - state.dropped_count) as usize
} else {
// Reader fell behind, skip to front
self.consumed_count = state.dropped_count;
0
};

if queue_index < state.queue.len() {
let datagram = state.queue.get(queue_index).cloned();
self.consumed_count += 1;
return Ok(datagram);
}

state.closed.clone()?;
match state.modified() {
Some(notify) => notify,
None => return Ok(None),
}
}
.await;
}
}

pub fn latest(&self) -> Option<(u64, u64)> {
let state = self.state.lock();
state.queue.back().map(|d| (d.group_id, d.object_id))
}
}

Start with moq-transport/src/serve/datagram.rs and inspect DatagramsWriter, DatagramsReader, and DatagramsState, then reproduce the loss using the relay and browser-player scenario described. Done means readers receive datagrams in FIFO order without last-write-wins loss, overflow is bounded with oldest-drop behavior, and independent readers remain supported.

Written by the indexing model from the issue text.