GameCult/Muninn
Watch
9
0
Fork
You've already forked Muninn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge Cut 2: bounded sender, one deadline owner, a drivable seam

Browse source 
Audio priority per payload, deadlines per payload, every sender-side drop
consumes a frame_id, the latency budget clamped to 2000 ms, a departed
receiver drops out without stopping the others. Soul pass 3: merge with
follow-ups (mid-group leaver order unpinned; per-receiver I/O errors still
stop the loop; raw budget persists in the command record).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 21:31:34 +02:00
commit bd69598503
7 changed files with 3094 additions and 1125 deletions
Download patch file Download diff file Expand all files Collapse all files

1280
crates/muninn-daemon/src/main.rs
View file

File diff suppressed because it is too large Load diff

323
crates/muninn-daemon/src/media_packetizer.rs
View file

@ -213,11 +213,20 @@ pub struct MuninnMediaSendPayload {
pub payload: Vec<u8>,
}
/// Everything one video access unit puts on the wire: its chunks and, until
/// the FEC cut replaces it, their parity. A group is sent whole or not at
/// all, because parity for a frame that never completes is wasted bandwidth.
pub type MuninnMediaPayloadGroup = Vec<MuninnMediaSendPayload>;
pub struct VideoAnnexBStreamSendState {
config: VideoAnnexBStreamSendConfig,
pending: Vec<u8>,
next_frame_id: u64,
next_pts_ticks: i64,
dropped_access_units: u64,
/// Set by an overflow drop: the slices that follow belong to the dropped
/// access unit and are discarded up to the next access-unit boundary.
discarding_tail: bool,
}
impl VideoAnnexBStreamSendState {
@ -260,26 +269,43 @@ impl VideoAnnexBStreamSendState {
next_pts_ticks: config.first_pts_ticks,
config,
pending: Vec::new(),
dropped_access_units: 0,
discarding_tail: false,
})
}
pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result<Vec<MuninnMediaSendPayload>> {
/// One group per completed access unit. Whatever is still pending after
/// that is a single unfinished access unit; when it outgrows
/// `max_pending_bytes` it is dropped and counted, never an error that
/// would restart the stream. The dropped unit consumes its frame id, so a
/// receiver sees the gap, and its remaining slices are discarded up to the
/// next access-unit boundary instead of being emitted as a frame of their
/// own.
pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result<Vec<MuninnMediaPayloadGroup>> {
if !bytes.is_empty() {
self.pending.extend_from_slice(bytes);
}
let groups = self.emit_available(stored_at, false)?;
if self.pending.len() > self.config.max_pending_bytes {
return Err(anyhow!(
"Annex B stream sender pending buffer exceeded {} bytes without a complete frame",
self.config.max_pending_bytes
));
self.pending.clear();
if !self.discarding_tail {
self.dropped_access_units += 1;
self.advance_frame_clock()?;
self.discarding_tail = true;
}
}
self.emit_available(stored_at, false)
Ok(groups)
}
pub fn finish(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaSendPayload>> {
pub fn finish(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaPayloadGroup>> {
self.emit_available(stored_at, true)
}
/// Access units dropped for outgrowing `max_pending_bytes`, cumulative.
pub fn dropped_access_units(&self) -> u64 {
self.dropped_access_units
}
pub fn pending_bytes(&self) -> usize {
self.pending.len()
}
@ -292,10 +318,23 @@ impl VideoAnnexBStreamSendState {
&mut self,
stored_at: &str,
flush_last: bool,
) -> Result<Vec<MuninnMediaSendPayload>> {
) -> Result<Vec<MuninnMediaPayloadGroup>> {
if stored_at.is_empty() {
return Err(anyhow!("stored_at must be non-empty"));
}
// After a drop the tail of the dropped access unit arrives with no
// start code in front of it; skip to the next one.
if let Some((start, _)) = find_annex_b_start_code(&self.pending, 0)
&& start > 0
{
self.pending.drain(..start);
}
if self.pending.is_empty() {
return Ok(Vec::new());
}
if self.discarding_tail && !self.discard_leftover_slices(flush_last)? {
return Ok(Vec::new());
}
if self.pending.is_empty() {
return Ok(Vec::new());
}
@ -319,7 +358,7 @@ impl VideoAnnexBStreamSendState {
return Ok(Vec::new());
}
let mut send_payloads = Vec::new();
let mut groups = Vec::with_capacity(emit_count);
let mut emitted_bytes = 0_usize;
for access_unit in access_units.iter().take(emit_count) {
emitted_bytes = emitted_bytes
@ -345,7 +384,7 @@ impl VideoAnnexBStreamSendState {
access_unit,
)?;
let wire_records = video_wire_records_with_parity(&records)?;
send_payloads.extend(wire_payloads_to_media_send(encode_media_wire_records(
groups.push(wire_payloads_to_media_send(encode_media_wire_records(
&wire_records,
stored_at,
&self.config.source_runtime_id,
@ -355,7 +394,49 @@ impl VideoAnnexBStreamSendState {
}
self.pending.drain(..emitted_bytes);
Ok(send_payloads)
Ok(groups)
}
/// While the tail of a dropped access unit is being discarded: if the
/// pending bytes open with a continuation slice, drops that leftover access
/// unit (up to the next boundary) and stops discarding; if they open on a
/// boundary, stops discarding and leaves them alone. `false` means it
/// cannot tell yet (the leftover is not complete) and the caller must wait
/// for more bytes.
fn discard_leftover_slices(&mut self, flush_last: bool) -> Result<bool> {
let codec = self.config.codec.as_str();
let nal_units = match normalized_video_codec(codec) {
Some("h265") => annex_b_nal_units(&self.pending, "H.265", h265_nal_type),
_ => annex_b_nal_units(&self.pending, "H.264", h264_nal_type),
};
let Ok(nal_units) = nal_units else {
return Ok(false);
};
let first = &nal_units[0];
let continuation = match normalized_video_codec(codec) {
Some("h265") => {
is_h265_vcl_nal(first.nal_type)
&& !h265_first_slice_segment_in_pic(first.payload).unwrap_or(false)
}
_ => {
is_h264_vcl_nal(first.nal_type)
&& h264_first_mb_in_slice(first.payload).unwrap_or(1) != 0
}
};
if !continuation {
self.discarding_tail = false;
return Ok(true);
}
let access_units = video_annex_b_access_units(codec, &self.pending)?;
if access_units.len() >= 2 {
self.pending.drain(..access_units[0].bytes.len());
} else if flush_last {
self.pending.clear();
} else {
return Ok(false);
}
self.discarding_tail = false;
Ok(true)
}
fn advance_frame_clock(&mut self) -> Result<()> {
@ -1525,8 +1606,8 @@ mod tests {
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[second_slice_header, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?;
payloads.extend(sender.finish("2026-06-18T00:00:00Z")?);
let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?);
payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?));
let mut records = Vec::new();
for payload in &payloads {
if let GameCultMediaWireRecord::Video(record) =
@ -1595,8 +1676,8 @@ mod tests {
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[0x41, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?;
let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?;
payloads.extend(sender.finish("2026-06-18T00:00:00Z")?);
let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?);
payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?));
assert!(payloads.len() >= 2);
let wire_records = payloads
.iter()
@ -1636,9 +1717,10 @@ mod tests {
let second_push = sender.push("2026-06-18T00:00:00Z", &second_frame)?;
assert_eq!(second_push.len(), 1);
assert_eq!(second_push[0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL);
assert_eq!(second_push[0].len(), 1);
assert_eq!(second_push[0][0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL);
let GameCultMediaWireRecord::Video(first) =
decode_media_wire_record(&second_push[0].payload)?
decode_media_wire_record(&second_push[0][0].payload)?
else {
panic!("expected video media record");
};
@ -1651,7 +1733,8 @@ mod tests {
let tail = sender.finish("2026-06-18T00:00:00Z")?;
assert_eq!(tail.len(), 1);
let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0].payload)?
assert_eq!(tail[0].len(), 1);
let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0][0].payload)?
else {
panic!("expected video media record");
};
@ -1680,21 +1763,205 @@ mod tests {
);
}
/// `(frame_id, whole access unit bytes)` for each group, chunks rejoined.
fn frames_of(groups: &[MuninnMediaPayloadGroup]) -> Result<Vec<(u64, Vec<u8>)>> {
groups
.iter()
.map(|group| {
let mut frame_id = None;
let mut bytes = Vec::new();
for payload in group {
if let GameCultMediaWireRecord::Video(record) =
decode_media_wire_record(&payload.payload)?
{
frame_id = Some(record.frame_id);
bytes.extend_from_slice(&record.payload);
}
}
Ok((frame_id.expect("a group carries video chunks"), bytes))
})
.collect()
}
#[test]
fn annex_b_stream_send_state_rejects_unbounded_pending_bytes() -> Result<()> {
fn an_overflow_drop_consumes_its_frame_id_and_discards_the_leftover_slices() -> Result<()> {
let mut config = stream_send_config();
config.max_pending_bytes = 4;
config.max_pending_bytes = 16;
config.max_payload_bytes = 64;
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let stored_at = "2026-06-18T00:00:00Z";
let nal = |header: [u8; 2], filler: &[u8]| {
[&start_code()[..], &header[..], filler].concat()
};
// Frame 9 whole, then the head of a frame that outgrows the ceiling.
let mut first = nal([0x65, 0x80], &[0xA1]);
first.extend(nal([0x41, 0x80], &[0xB2; 30]));
let groups = frames_of(&sender.push(stored_at, &first)?)?;
assert_eq!(groups, vec![(9, nal([0x65, 0x80], &[0xA1]))]);
assert_eq!(sender.dropped_access_units(), 1);
assert_eq!(sender.next_frame_id(), 11, "the dropped frame 10 is spent");
// Its second slice (first_mb_in_slice != 0) arrives alone: nothing can
// be decided yet, and nothing is emitted.
assert!(sender.push(stored_at, &nal([0x41, 0x40], &[0xC3]))?.is_empty());
assert_eq!(sender.dropped_access_units(), 1);
// Then two whole frames follow: the leftover slice goes with frame 10.
let mut next = nal([0x41, 0x80], &[0xD4]);
next.extend(nal([0x41, 0x80], &[0xE5]));
let mut groups = frames_of(&sender.push(stored_at, &next)?)?;
groups.extend(frames_of(&sender.finish(stored_at)?)?);
assert_eq!(
groups,
vec![
(11, nal([0x41, 0x80], &[0xD4])),
(12, nal([0x41, 0x80], &[0xE5])),
]
);
assert_eq!(sender.dropped_access_units(), 1);
Ok(())
}
#[test]
fn a_leftover_spanning_two_pushes_is_discarded_whole() -> Result<()> {
let mut config = stream_send_config();
config.max_pending_bytes = 16;
config.max_payload_bytes = 64;
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let stored_at = "2026-06-18T00:00:00Z";
let nal = |header: [u8; 2], filler: &[u8]| {
[&start_code()[..], &header[..], filler].concat()
};
let mut first = nal([0x65, 0x80], &[0xA1]);
first.extend(nal([0x41, 0x80], &[0xB2; 30]));
assert_eq!(sender.push(stored_at, &first)?.len(), 1);
assert_eq!(sender.dropped_access_units(), 1);
// The dropped frame's second and third slices (first_mb_in_slice 1
// and 3) arrive in separate pushes. Neither is a frame, and neither
// may come out as one.
assert!(sender.push(stored_at, &nal([0x41, 0x40], &[0xC3]))?.is_empty());
assert!(sender.push(stored_at, &nal([0x41, 0x20], &[0xC4]))?.is_empty());
assert_eq!(sender.next_frame_id(), 11);
let mut next = nal([0x41, 0x80], &[0xD4]);
next.extend(nal([0x41, 0x80], &[0xE5]));
let mut groups = frames_of(&sender.push(stored_at, &next)?)?;
groups.extend(frames_of(&sender.finish(stored_at)?)?);
assert_eq!(
groups,
vec![
(11, nal([0x41, 0x80], &[0xD4])),
(12, nal([0x41, 0x80], &[0xE5])),
],
"both leftover slices went with the dropped frame"
);
assert_eq!(sender.dropped_access_units(), 1);
Ok(())
}
#[test]
fn a_stream_that_resumes_on_a_frame_boundary_after_an_overflow_loses_nothing_more() -> Result<()> {
let mut config = stream_send_config();
config.max_pending_bytes = 16;
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let stored_at = "2026-06-18T00:00:00Z";
let nal = |header: [u8; 2], filler: &[u8]| {
[&start_code()[..], &header[..], filler].concat()
};
assert!(sender.push(stored_at, &nal([0x65, 0x80], &[0xB2; 30]))?.is_empty());
assert_eq!(sender.next_frame_id(), 10);
// The next bytes open on a new frame, not on a leftover slice.
let mut next = nal([0x41, 0x80], &[0xD4]);
next.extend(nal([0x41, 0x80], &[0xE5]));
let groups = frames_of(&sender.push(stored_at, &next)?)?;
assert_eq!(groups.len(), 1);
assert_eq!(groups[0].0, 10);
Ok(())
}
fn flat(groups: Vec<MuninnMediaPayloadGroup>) -> Vec<MuninnMediaSendPayload> {
groups.into_iter().flatten().collect()
}
#[test]
fn annex_b_stream_send_state_drops_and_counts_an_oversize_access_unit_instead_of_erroring()
-> Result<()> {
let mut config = stream_send_config();
config.max_pending_bytes = 8;
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let error = sender
.push("2026-06-18T00:00:00Z", &[0x47, 0x40, 0x00, 0x10, 0x00])
.unwrap_err();
// One access unit that never completes and is larger than the ceiling.
let mut oversize = start_code().to_vec();
oversize.extend_from_slice(&[0x65, 0x80, 1, 2, 3, 4, 5, 6, 7, 8]);
let groups = sender.push("2026-06-18T00:00:00Z", &oversize)?;
assert!(
error
.to_string()
.contains("pending buffer exceeded 4 bytes")
);
assert!(groups.is_empty());
assert_eq!(sender.dropped_access_units(), 1);
assert_eq!(sender.pending_bytes(), 0);
// The tail of the dropped unit has no start code and is skipped; the
// stream then carries on with the next access units.
let mut next = vec![9, 9, 9];
next.extend_from_slice(&start_code());
next.extend_from_slice(&[0x65, 0x80]);
next.extend_from_slice(&start_code());
next.extend_from_slice(&[0x41, 0x80]);
let groups = sender.push("2026-06-18T00:00:00Z", &next)?;
assert_eq!(groups.len(), 1);
assert_eq!(sender.dropped_access_units(), 1);
let GameCultMediaWireRecord::Video(record) =
decode_media_wire_record(&groups[0][0].payload)?
else {
panic!("expected video media record");
};
assert!(record.keyframe);
Ok(())
}
#[test]
fn annex_b_stream_send_state_returns_one_group_per_access_unit_with_all_its_chunks_and_parity()
-> Result<()> {
// Three chunks of 16 bytes plus a short fourth: one access unit.
let mut stream = start_code().to_vec();
stream.extend_from_slice(&[0x65, 0x80]);
stream.resize(52, 0xAB);
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[0x41, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?;
let mut groups = sender.push("2026-06-18T00:00:00Z", &stream)?;
groups.extend(sender.finish("2026-06-18T00:00:00Z")?);
assert_eq!(groups.len(), 2);
let mut frame_ids_per_group = Vec::new();
for group in &groups {
let mut ids = std::collections::BTreeSet::new();
let mut data = 0;
let mut parity = 0;
for payload in group {
match decode_media_wire_record(&payload.payload)? {
GameCultMediaWireRecord::Video(record) => {
ids.insert(record.frame_id);
data += 1;
}
GameCultMediaWireRecord::VideoParity(record) => {
ids.insert(record.frame_id);
parity += 1;
}
other => panic!("unexpected record {other:?}"),
}
}
assert_eq!(ids.len(), 1, "a group carries exactly one frame");
frame_ids_per_group.push((ids.into_iter().next().unwrap(), data, parity));
}
assert_eq!(frame_ids_per_group[0], (9, 4, 4));
assert_eq!(frame_ids_per_group[1], (10, 1, 0));
Ok(())
}

1085
crates/muninn-daemon/src/media_send.rs Normal file
View file

File diff suppressed because it is too large Load diff

1523
crates/muninn-daemon/src/media_send/tests.rs Normal file
View file

File diff suppressed because it is too large Load diff

BIN
crates/muninn-daemon/testdata/clip.h264 vendored Normal file
View file

Binary file not shown.

7
crates/muninn-daemon/testdata/clip.h264.provenance.txt vendored Normal file
View file

@ -0,0 +1,7 @@
clip.h264: 15 frames of the ffmpeg testsrc2 pattern, 160x120 at 30 fps, H.264 Constrained
Baseline in Annex B with access unit delimiters, one IDR every 8 frames, no B-frames.
Generated once and committed so the media send harness needs no ffmpeg.
Command (ffmpeg 8.1.1 full build, libx264, 2026-09-29):
ffmpeg -f lavfi -i "testsrc2=size=160x120:rate=30:duration=0.5" -c:v libx264 -profile:v baseline -preset veryslow -crf 28 -bf 0 -g 8 -x264-params "threads=1:aud=1:repeat-headers=1:keyint=8:min-keyint=8:scenecut=0" -f h264 clip.h264

5
docs/muninn-media-streaming.md
View file

@ -237,7 +237,7 @@ a temporary PCM/AAC compatibility bridge.
Recommended first LAN profile:
- Profile id: `muninn.rudp.low_latency_h264_lan.v1`.
- Video: H.264 NVENC, `p5` (`main.rs:10279`), ultra-low-latency tune, CBR high enough that LAN
- Video: H.264 NVENC, `p5` (`main.rs:9504`), ultra-low-latency tune, CBR high enough that LAN
bandwidth is not the constraint, no B-frames, no lookahead, short GOP or
intra-refresh, keyframe actuation on feedback pressure (not built: keyframe requests are only
counted and logged today; Cut 6 builds it).
@ -257,7 +257,8 @@ Recommended first LAN profile:
in CultNet deliberately instead of falling through to accidental IP
fragmentation.
- Resend cadence: sender RUDP media resends are scheduled every `30` ms (`main.rs:94`) while
the packet is still inside its expiry, which is the request's latency budget (default `2,000` ms, `main.rs:95`).
the packet is still inside its expiry, which is the request's latency budget (a request that names none gets `250` ms, `main.rs:98`, and the advertisement offers the same as `default_latency_budget_ms`).
- Send queue: `media_send.rs` owns it. The unit is a group, one access unit with its parity or one audio packet, stamped `produced_at + latency budget` when the packetizer emits it. A request's latency budget is clamped to 1..=2000 ms. The send loop moves one payload per step and picks it afresh each time: audio, then repairs, then video, so audio never waits behind more than the payload in progress, full socket or not. The group deadline is checked before every payload; a group that outlives it mid-send is abandoned and counted `groups_cut_short`. Queues are capped per kind (audio 256, video 512 groups) and drop the oldest group, counted as `groups_dropped_audio` and `groups_dropped_video`. Every access unit the sender drops (queue cap, expiry, packetizer overflow) has already consumed its frame id, so the receiver sees a gap; an overflow also discards the dropped frame's remaining slices instead of emitting them as a frame. A repair keeps the deadline of the frame it repairs, goes only to the receiver that asked, and is not sent after that deadline.
- Control: sender may adapt bitrate, keyframe cadence, chunk size, and playout
budget from receiver feedback; the receiver must not silently stretch latency
to preserve visual perfection.