From f4f83202cbe09307a70aacda52a9a9c5487c84c6 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 14:07:46 +0200 Subject: [PATCH] feat(core/client): an AU's prefix reaches the decoder while its tail is on the wire Delivery used to be all-or-nothing: the decoder saw byte 0 only after the last packet of the AU landed, so the whole transmit time sat in front of decode. With the slice-streamed wire (previous commit) blocks now arrive addressable, and a client can opt in (connect's new frame_parts) to receive each AU's newly-contiguous prefix as Frame::part pieces - offset tiling, first/last marked, the completing push carrying only the suffix. A PARTIAL_FRAME-capable decoder then chews slices concurrently with the remaining network transfer. The reassembler walks a per-frame cursor over successfully-completed blocks (failed FEC reconstructs don't advance it), coalesces blocks that finished out of order into one part, keeps probe filler whole, and stops short of the final block so the zero-padded tail still trims at completion. Whole-frame consumers see byte-identical behavior - parts never flow without the opt-in, and never on PyroWave (its newest-wins draining assumes whole AUs). Per-AU accounting keeps its units: OWD/ABR feeds, the inter-arrival series and the clock-based jump-to-live detector only count completing deliveries, and FrameChannel::depth() counts AUs so a part-rich queue can't trip jump-to-live at a fraction of the real backlog. The consumer contract (gap or orphan part = AU lost: abandon, flush, resync on the next first) is documented on FramePart; the C ABI keeps parts off until PunktfunkFrame can express them. Co-Authored-By: Claude Fable 5 --- clients/android/native/src/session/connect.rs | 3 + clients/cli/src/main.rs | 17 +- clients/linux/src/app.rs | 7 +- clients/windows/src/probe.rs | 9 +- crates/pf-client-core/src/session.rs | 3 + crates/punktfunk-core/src/abi.rs | 5 + .../src/client/frame_channel.rs | 37 +++- crates/punktfunk-core/src/client/mod.rs | 6 + crates/punktfunk-core/src/client/pump/data.rs | 11 +- .../src/client/pump/handshake.rs | 6 + crates/punktfunk-core/src/client/worker.rs | 4 + .../punktfunk-core/src/packet/reassemble.rs | 85 +++++++- crates/punktfunk-core/src/packet/tests.rs | 182 ++++++++++++++++++ crates/punktfunk-core/src/session.rs | 41 +++- crates/punktfunk-host/src/native.rs | 76 ++++---- 15 files changed, 433 insertions(+), 59 deletions(-) diff --git a/clients/android/native/src/session/connect.rs b/clients/android/native/src/session/connect.rs index 839e0835..eefbe48e 100644 --- a/clients/android/native/src/session/connect.rs +++ b/clients/android/native/src/session/connect.rs @@ -225,6 +225,9 @@ pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeConnect<'lo // No non-video caps: this client does not render the host cursor locally (no shape/state // planes in the jni surface), so advertising CLIENT_CAP_CURSOR would stream cursor-less. 0, + // Slice-progressive delivery: off until the decode loop feeds MediaCodec with + // BUFFER_FLAG_PARTIAL_FRAME (P2d flips this behind a FEATURE_PartialFrame probe). + false, launch, // a store-qualified library id to boot into a game, or None for the desktop device_name, // Kotlin's Build.MODEL — the host's approval-list / trust-store label pin, // Some → Crypto on host-fp mismatch diff --git a/clients/cli/src/main.rs b/clients/cli/src/main.rs index 011dc883..b359f7b1 100644 --- a/clients/cli/src/main.rs +++ b/clients/cli/src/main.rs @@ -798,14 +798,15 @@ from the config directory for a true factory reset." }, punktfunk_core::config::CompositorPref::Auto, punktfunk_core::config::GamepadPref::Auto, - 0, // bitrate_kbps: the host's default; this connect never presents - 0, // video_caps: nothing decodes here - 2, // audio_channels - 0, // video_codecs: the probe carries no video - 0, // preferred_codec - None, // display_hdr - 0, // client_caps: nothing renders a cursor - None, // launch + 0, // bitrate_kbps: the host's default; this connect never presents + 0, // video_caps: nothing decodes here + 2, // audio_channels + 0, // video_codecs: the probe carries no video + 0, // preferred_codec + None, // display_hdr + 0, // client_caps: nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch Some(punktfunk_core::client::device_name()), Some(pin), Some(identity), diff --git a/clients/linux/src/app.rs b/clients/linux/src/app.rs index 5fef2feb..eb3cb4f2 100644 --- a/clients/linux/src/app.rs +++ b/clients/linux/src/app.rs @@ -696,9 +696,10 @@ impl AppModel { 2, // audio_channels: stereo crate::video::decodable_codecs(), // codecs (unused by the probe, but honest) 0, // preferred_codec: no preference - None, // display_hdr: probe connect, nothing presents - 0, // client_caps: probe connect, nothing renders a cursor - None, // launch: probe connect, no game + None, // display_hdr: probe connect, nothing presents + 0, // client_caps: probe connect, nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch: probe connect, no game // Knock under this device's name, not a fingerprint placeholder, when the // probed host doesn't know us yet. Some(pf_client_core::trust::device_name()), diff --git a/clients/windows/src/probe.rs b/clients/windows/src/probe.rs index e159c5fc..b574bb37 100644 --- a/clients/windows/src/probe.rs +++ b/clients/windows/src/probe.rs @@ -54,10 +54,11 @@ pub fn run_speed_probe( 0, // video_caps: probe connect, nothing is decoded 2, // audio_channels: stereo baseline decodable_codecs(), - 0, // preferred_codec: no preference - None, // display_hdr: probe connect, nothing presents - 0, // client_caps: probe connect, nothing renders a cursor - None, // launch: no game + 0, // preferred_codec: no preference + None, // display_hdr: probe connect, nothing presents + 0, // client_caps: probe connect, nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch: no game // Same label a real session sends — a speed test against a host that doesn't know us yet // should knock under this device's name, not a fingerprint placeholder. Some(punktfunk_core::client::device_name()), diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 65746854..93c6dc38 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -272,6 +272,9 @@ fn pump( } else { 0 }, + // Slice-progressive delivery: off — this presenter feeds FFmpeg whole AUs; a partial + // avcodec feed path can flip it later. + false, params.launch.clone(), // The host's approval-list / trust-store label for this client. Without it every no-PIN // "request access" knock showed up as the fingerprint placeholder "device abcd1234". diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index 54495f87..adadd089 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -1871,6 +1871,11 @@ unsafe fn connect_ex_impl( // compositing the pointer — only an embedder that renders the cursor planes // ([`punktfunk_connection_next_cursor_shape`]/`_state`) may set it. ex7/ex8 pass 0. client_caps, + // The C ABI cannot carry slice-progressive parts yet — `PunktfunkFrame` has no + // part/completeness fields, so a part would be indistinguishable from a whole AU. + // An `ex10` variant adds the opt-in together with those fields when an ABI embedder + // (Apple) grows a partial-feed decode path. + false, launch, // The C ABI has no device-name parameter (only `punktfunk_pair` takes one), so every // embedder gets the OS hostname default — this is what the host's pending-approval diff --git a/crates/punktfunk-core/src/client/frame_channel.rs b/crates/punktfunk-core/src/client/frame_channel.rs index c011d225..4d5c3dad 100644 --- a/crates/punktfunk-core/src/client/frame_channel.rs +++ b/crates/punktfunk-core/src/client/frame_channel.rs @@ -321,9 +321,18 @@ impl FrameChannel { self.ready.notify_one(); } - /// Pump side: current queued depth — the clock-free standing-queue signal. + /// Pump side: current queued depth in ACCESS UNITS — the clock-free standing-queue + /// signal. Slice-progressive parts of a still-open AU don't count (the detector's + /// thresholds are in frames; counting parts would trip it at a fraction of the real + /// backlog), and a queue that stops draining still accumulates completed AUs. pub(crate) fn depth(&self) -> usize { - self.inner.lock().unwrap().q.len() + self.inner + .lock() + .unwrap() + .q + .iter() + .filter(|f| f.complete) + .count() } /// Pump side: discard the whole backlog (the jump-to-live path); returns how many were dropped. @@ -378,10 +387,34 @@ mod frame_channel_tests { pts_ns: i as u64, flags: 0, complete: true, + part: None, received_ns: 0, } } + /// `depth()` is the standing-queue detector's signal, in AU units: parts of a + /// still-open AU must not count (the thresholds are frames — parts would trip + /// jump-to-live at a fraction of the real backlog), while completed AUs always do. + #[test] + fn depth_counts_aus_not_parts() { + let ch = FrameChannel::new(); + let mut p = frame(1); + p.complete = false; + p.part = Some(crate::session::FramePart { + offset: 0, + first: true, + last: false, + }); + ch.push(p); + assert_eq!( + ch.depth(), + 0, + "an open AU's prefix part is not a queued frame" + ); + ch.push(frame(2)); + assert_eq!(ch.depth(), 1); + } + fn popped(ch: &FrameChannel) -> Option { match ch.pop(Duration::from_millis(0)) { FramePop::Frame(f) => Some(f.frame_index), diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index ba36f9f7..59472a0e 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -354,6 +354,11 @@ impl NativeClient { // the video for a session that advertises it, so a non-rendering embedder that sets it // streams with NO visible cursor at all. `0` = today's composited behavior. client_caps: u8, + // Slice-progressive delivery opt-in: AU prefixes arrive as [`Frame`]s with + // [`crate::session::Frame::part`]` = Some` while the rest is still on the wire. Set it + // ONLY when this embedder's decode path understands parts (e.g. feeds MediaCodec with + // BUFFER_FLAG_PARTIAL_FRAME); with it false every AU arrives whole, exactly as before. + frame_parts: bool, launch: Option, // This device's display name, carried in [`crate::quic::Hello::name`]: what the host's // pending-approval list shows when an unpaired client knocks, and what its trust store @@ -438,6 +443,7 @@ impl NativeClient { preferred_codec, display_hdr, client_caps, + frame_parts, launch, name, pin, diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index ade6c30c..6b5527dc 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -520,7 +520,12 @@ impl DataPump { if frame.flags & FLAG_PROBE as u32 != 0 { continue; // speed-test filler, not video — measured via the counters above } - if pump_perf_on { + // A prefix part is not an AU arrival: the inter-arrival series, the OWD + // window and the clock-based staleness detector below all measure per-AU + // signals, so only the delivery that completes an AU feeds them (parts + // would bias OWD low and constantly reset the staleness run). + let is_au = frame.complete; + if pump_perf_on && is_au { let now = Instant::now(); if let Some(prev) = last_arrival.replace(now) { // 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded. @@ -552,7 +557,7 @@ impl DataPump { stale_since = None; standing_since = None; } else { - let lat_ns = if clock_offset_ns != 0 { + let lat_ns = if clock_offset_ns != 0 && is_au { now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128 } else { 0 @@ -576,7 +581,7 @@ impl DataPump { && lat_ns > FLUSH_LATENCY.as_nanos() as i128 { stale_since.get_or_insert_with(Instant::now); - } else { + } else if is_au { stale_since = None; } let depth = frames.depth(); diff --git a/crates/punktfunk-core/src/client/pump/handshake.rs b/crates/punktfunk-core/src/client/pump/handshake.rs index 981a48a4..c8f79b04 100644 --- a/crates/punktfunk-core/src/client/pump/handshake.rs +++ b/crates/punktfunk-core/src/client/pump/handshake.rs @@ -221,6 +221,12 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result(( session, send, diff --git a/crates/punktfunk-core/src/client/worker.rs b/crates/punktfunk-core/src/client/worker.rs index 18100b83..07b1f961 100644 --- a/crates/punktfunk-core/src/client/worker.rs +++ b/crates/punktfunk-core/src/client/worker.rs @@ -23,6 +23,10 @@ pub(crate) struct WorkerArgs { pub(crate) preferred_codec: u8, pub(crate) display_hdr: Option, pub(crate) client_caps: u8, + /// Slice-progressive delivery opt-in ([`crate::session::Session::set_deliver_frame_parts`]): + /// only an embedder whose decode path understands [`crate::session::Frame::part`] may set + /// it. Ignored on all-intra (PyroWave) sessions — their newest-wins draining needs whole AUs. + pub(crate) frame_parts: bool, pub(crate) launch: Option, /// This device's display name, sent in `Hello` (the host's approval list / trust store label). pub(crate) name: Option, diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 4ac5dbfb..26507e62 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -5,7 +5,7 @@ use super::*; use crate::config::Config; use crate::error::Result; use crate::fec::ErasureCoder; -use crate::session::Frame; +use crate::session::{Frame, FramePart}; use crate::stats::StatsCounters; use std::collections::HashMap; use zerocopy::FromBytes; @@ -85,6 +85,11 @@ struct FrameBuf { block_count: usize, pts_ns: u64, user_flags: u32, + /// Slice-progressive delivery cursor ([`Reassembler::deliver_parts`]): count of leading + /// blocks already handed up as prefix parts... + next_part_block: u16, + /// ...and their total data shards — the AU shard offset where the next part starts. + delivered_shards: usize, /// The whole frame's data region — `total_data_shards × shard_bytes` zeroed bytes. Data /// shards are copied to their final offset on arrival; FEC reconstruction writes only the /// missing shards' ranges. On completion this Vec IS [`Frame::data`] (truncated to @@ -187,6 +192,11 @@ pub struct Reassembler { deliver_partial: bool, /// The newest such partial awaiting pickup (newest-wins: partials are a lossy byproduct). pending_partial: Option, + /// Deliver each video AU's newly-contiguous PREFIX as [`Frame`]s with + /// [`Frame::part`]`= Some` while the rest is still in flight (client opt-in — the + /// slice-progressive decode path). Works for any multi-block frame — both streamed shapes + /// and legacy uniform frames tile their blocks contiguously (`BlockState::base_shard`). + deliver_parts: bool, /// The video stream's window — its aged-out incomplete frames count into `frames_dropped` /// (the client's loss-recovery trigger). video: ReassemblyWindow, @@ -210,6 +220,7 @@ impl Reassembler { limits, deliver_partial: false, pending_partial: None, + deliver_parts: false, video: ReassemblyWindow::default(), probe: ReassemblyWindow::default(), recovery_pool: Vec::new(), @@ -225,6 +236,11 @@ impl Reassembler { } } + /// Opt into slice-progressive prefix delivery (see [`Reassembler::deliver_parts`]). + pub fn set_deliver_parts(&mut self, on: bool) { + self.deliver_parts = on; + } + /// Take the newest aged-out partial frame, if one is pending (see `set_deliver_partial`). pub fn take_partial(&mut self) -> Option { self.pending_partial.take() @@ -260,11 +276,13 @@ impl Reassembler { limits, deliver_partial, pending_partial, + deliver_parts, video, probe, recovery_pool, in_flight_bytes, } = self; + let deliver_parts = *deliver_parts; let lim = *limits; let shard_bytes = hdr.shard_bytes as usize; let data_shards = hdr.data_shards as usize; @@ -454,6 +472,8 @@ impl Reassembler { block_count, pts_ns: hdr.pts_ns, user_flags: hdr.user_flags, + next_part_block: 0, + delivered_shards: 0, buf: vec![0; buf_len], blocks: HashMap::new(), blocks_ok: 0, @@ -566,8 +586,13 @@ impl Reassembler { blocks, blocks_ok, block_count: frame_block_count, + pts_ns: frame_pts_ns, + user_flags: frame_user_flags, + next_part_block, + delivered_shards, .. } = frame; + let (frame_pts_ns, frame_user_flags) = (*frame_pts_ns, *frame_user_flags); // First packet of a block sizes its state; `data_shards` is already pinned by the // derived geometry above, but `recovery_shards` is per-block wire input (adaptive FEC @@ -713,15 +738,69 @@ impl Reassembler { ); *in_flight_bytes -= done.buf.len(); done.buf.truncate(done.frame_bytes); // trim trailing-shard zero padding + // Slice-progressive consumers already hold the delivered prefix — the completing + // packet hands up only the SUFFIX (with `last`), or the degenerate whole-AU part + // when nothing was delivered early. Probe filler stays whole either way — the + // speed test accounts AUs, not slices. + let (data, part) = if deliver_parts && !is_probe { + let lo = (done.delivered_shards * shard_bytes).min(done.frame_bytes); + let part = FramePart { + offset: lo as u32, + first: lo == 0, + last: true, + }; + if lo == 0 { + (done.buf, Some(part)) + } else { + (done.buf[lo..].to_vec(), Some(part)) + } + } else { + (done.buf, None) + }; return Ok(Some(Frame { - data: done.buf, + data, frame_index: hdr.frame_index, pts_ns: done.pts_ns, flags: done.user_flags, complete: true, + part, received_ns: 0, // stamped by Session::poll_frame at the session boundary })); } + // Slice-progressive delivery: when this packet completed a block that extends the AU's + // contiguous prefix (possibly unlocking blocks that finished out of order behind it), + // hand the newly-contiguous bytes up as ONE part. Only successfully-completed blocks + // count (`done` alone includes failed reconstructs), and the cursor stops short of the + // FINAL block — its zero-padded tail is only trimmed at completion above. + if deliver_parts && !is_probe { + let start = *delivered_shards; + while let Some(b) = blocks.get(&*next_part_block) { + if !(b.done && (b.reconstructed || b.data_received == b.data_shards)) { + break; + } + if block_count != 0 && (*next_part_block as usize) + 1 >= block_count { + break; + } + *delivered_shards = b.base_shard + b.data_shards; + *next_part_block += 1; + } + if *delivered_shards > start { + let (lo, hi) = (start * shard_bytes, *delivered_shards * shard_bytes); + return Ok(Some(Frame { + data: buf[lo..hi].to_vec(), + frame_index: hdr.frame_index, + pts_ns: frame_pts_ns, + flags: frame_user_flags, + complete: false, + part: Some(FramePart { + offset: lo as u32, + first: start == 0, + last: false, + }), + received_ns: 0, // stamped by Session::poll_frame at the session boundary + })); + } + } Ok(None) } @@ -830,6 +909,7 @@ impl ReassemblyWindow { pts_ns: f.pts_ns, flags: f.user_flags, complete: false, + part: None, received_ns: 0, // stamped by Session::poll_frame at the session boundary }); } @@ -894,6 +974,7 @@ mod reset_tests { pts_ns: 1, flags: 0, complete: false, + part: None, received_ns: 0, }); r.reset(); diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 3097eafb..b0bb3c68 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -1296,6 +1296,188 @@ fn slice_streamed_mixed_flag_packet_dropped() { assert_eq!(stats.snapshot().frames_dropped, 0, "dropped, not killed"); } +// --------------------------------------------------------------------------- +// Slice-progressive prefix delivery (Frame::part — P2c) +// --------------------------------------------------------------------------- + +/// Push packets collecting EVERY delivery (parts and completions), returning them in order. +fn push_collect( + r: &mut Reassembler, + coder: &dyn crate::fec::ErasureCoder, + stats: &StatsCounters, + delivery: &[Vec], +) -> Vec { + let mut out = Vec::new(); + for p in delivery { + if let Some(f) = r.push(p, coder, stats).unwrap() { + out.push(f); + } + } + out +} + +/// In-order slice delivery streams one part per completed block, offsets tiling exactly, the +/// final delivery carrying only the suffix with `last` + `complete`. +#[test] +fn parts_stream_in_order() { + let (pkts, src) = slice_streamed_packets(); + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert_eq!( + got.len(), + 4, + "three sentinel-block parts + the final suffix" + ); + let mut rebuilt = Vec::new(); + for (i, f) in got.iter().enumerate() { + let part = f + .part + .expect("parts mode: every delivery carries part meta"); + assert_eq!( + part.offset as usize, + rebuilt.len(), + "parts tile with no gaps" + ); + assert_eq!(part.first, i == 0); + assert_eq!(part.last, i + 1 == got.len()); + assert_eq!( + f.complete, part.last, + "complete rides exactly the last part" + ); + rebuilt.extend_from_slice(&f.data); + } + assert_eq!(rebuilt, src, "concatenated parts must be the byte-exact AU"); + assert_eq!( + stats.snapshot().frames_completed, + 0, + "the reassembler leaves the completion count to the session boundary" + ); +} + +/// A block completing BEHIND the prefix emits nothing; the block that closes the gap emits +/// ONE coalesced part spanning everything unlocked. +#[test] +fn parts_coalesce_across_reordered_blocks() { + let (pkts, src) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + // Blocks 1 and 2 fully first, then block 0, then the final block. + let mut delivery: Vec> = Vec::new(); + for want in [1u16, 2, 0] { + delivery.extend( + pkts.iter() + .filter(|p| { + let h = hdr_of(p); + h.block_count == 0 && h.block_index == want + }) + .cloned(), + ); + } + delivery.extend(pkts.iter().filter(|p| hdr_of(p).block_count != 0).cloned()); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery); + assert_eq!(got.len(), 2, "one coalesced prefix part + the final suffix"); + let p0 = got[0].part.unwrap(); + assert_eq!((p0.offset, p0.first, p0.last), (0, true, false)); + assert_eq!(got[0].data.len(), 720 + 288, "blocks 0-2 in one part"); + let p1 = got[1].part.unwrap(); + assert!(p1.last && got[1].complete); + let mut rebuilt = got[0].data.clone(); + rebuilt.extend_from_slice(&got[1].data); + assert_eq!(rebuilt, src); +} + +/// Loss inside a block delays its part until FEC reconstructs it — the part then carries the +/// recovered bytes, still byte-exact. +#[test] +fn parts_wait_for_fec_reconstruction() { + let (pkts, src) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + // Kill two data shards of block 0 — within the 50% parity budget. + let delivery: Vec> = pkts + .iter() + .filter(|p| { + let h = hdr_of(p); + !(h.block_count == 0 && h.block_index == 0 && [1, 7].contains(&h.shard_index)) + }) + .cloned() + .collect(); + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery); + let mut rebuilt = Vec::new(); + for f in &got { + rebuilt.extend_from_slice(&f.data); + } + assert_eq!( + rebuilt, src, + "reconstructed prefix parts must be byte-exact" + ); + assert!(got.last().unwrap().complete); +} + +/// With parts on, a legacy single-block frame degenerates to ONE whole-AU delivery carrying +/// `{offset 0, first, last}` — the consumer's feed logic stays uniform. +#[test] +fn parts_degenerate_whole_frame() { + let cfg = e2e_config(FecScheme::Gf16, 50); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let src: Vec = (0..40).map(|i| i as u8).collect(); + let pkts = pk.packetize(&src, 7, 0, coder.as_ref()).unwrap(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert_eq!(got.len(), 1); + let f = &got[0]; + assert_eq!( + f.part, + Some(crate::session::FramePart { + offset: 0, + first: true, + last: true + }) + ); + assert!(f.complete); + assert_eq!(f.data, src); +} + +/// Parts also flow for the LEGACY streamed shape (uniform full-K sentinels) — the prefix +/// cursor rides `base_shard`, which both wire shapes maintain. +#[test] +fn parts_flow_for_legacy_streamed_frames() { + let chunks: Vec> = (0..3) + .map(|c| (0..50).map(|i| (c * 57 + i * 131 + 7) as u8).collect()) + .collect(); + let chunk_refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect(); + let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &chunk_refs); + let cfg = e2e_config(FecScheme::Gf16, 50); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert!(got.len() > 1, "sentinel blocks must deliver early parts"); + let mut rebuilt = Vec::new(); + for f in &got { + rebuilt.extend_from_slice(&f.data); + } + assert_eq!(rebuilt, src); + assert!(got.last().unwrap().complete); +} + /// A sentinel first-packet commits a MAX-sized frame buffer, so the in-flight budget must /// bite after IN_FLIGHT_BUF_FACTOR frames — the amplification bound for one-datagram opens. #[test] diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 544c0962..fd15e672 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -20,6 +20,27 @@ use crate::stats::{Stats, StatsCounters}; use crate::transport::Transport; use zerocopy::IntoBytes; +/// Slice-progressive delivery metadata ([`Session::set_deliver_frame_parts`]): this [`Frame`] +/// carries one contiguous piece of an access unit, handed up while the rest is still on the +/// wire so a `PARTIAL_FRAME`-capable decoder can start ahead of the last packet. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FramePart { + /// Byte offset of `data` within the whole AU. The reassembler emits parts in order with + /// no gaps — but the pre-decode hand-off may drop entries under memory pressure or a + /// jump-to-live clear, so a consumer must still verify `offset` equals its open AU's next + /// expected byte and treat a mismatch as the AU lost (abandon + flush the decoder's + /// partial input, then discard until the next `first`). The same discard applies to a + /// non-`first` part arriving with no AU open. + pub offset: u32, + /// The AU's first part. A `first` part for a NEW `frame_index` while an earlier AU is + /// still open means that AU died mid-flight (aged out or was cleared) — abandon it and + /// flush the decoder's partial input; no explicit abort part is ever sent. + pub first: bool, + /// The AU's final part — the whole AU has now been delivered ([`Frame::complete`] is set + /// on exactly this part) and the input may be submitted to the decoder as finished. + pub last: bool, +} + /// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder. pub struct Frame { pub data: Vec, @@ -32,6 +53,10 @@ pub struct Frame { /// ([`crate::packet::USER_FLAG_CHUNK_ALIGNED`]) are ever delivered partial; missing /// shard ranges are zero-filled at their exact offsets. pub complete: bool, + /// `Some` = slice-progressive delivery is on and this is ONE PIECE of an AU (see + /// [`FramePart`]); `data` holds only the part's bytes. `None` = a whole AU (or an + /// aged-out chunk-aligned partial — `complete` distinguishes). + pub part: Option, /// Wall-clock instant (ns since the Unix epoch, CLOCK_REALTIME basis — the same clock the /// skew handshake compares and the host stamps `pts_ns` with) at which this AU finished /// reassembly, stamped by [`Session::poll_frame`] as the frame leaves the session. Embedders @@ -611,6 +636,16 @@ impl Session { self.reassembler.set_deliver_partial(on); } + /// Client opt-in: deliver each AU's newly-contiguous prefix as [`Frame`]s with + /// [`Frame::part`]` = Some` while the rest is still on the wire, instead of one whole-AU + /// delivery (the slice-progressive decode path — [`crate::packet::USER_FLAG_SLICE_STREAM`]). + /// With it on, EVERY video frame delivery carries `part: Some` (a frame with no early + /// parts arrives as the degenerate `{offset: 0, first, last}` whole). Do not combine with + /// an all-intra (PyroWave) stream: its newest-wins draining assumes whole AUs. + pub fn set_deliver_frame_parts(&mut self, on: bool) { + self.reassembler.set_deliver_parts(on); + } + /// The session's negotiated wire shard payload size (bytes of AU per datagram) — /// the window size for chunk-aligned AUs (`USER_FLAG_CHUNK_ALIGNED`). pub fn shard_payload(&self) -> usize { @@ -710,7 +745,11 @@ impl Session { p.packets += 1; } if let Some(frame) = pushed { - StatsCounters::add(&self.stats.frames_completed, 1); + // A prefix part is not a completed frame — only the delivery that closes the + // AU counts, or parts would multiply the completion rate. + if frame.complete { + StatsCounters::add(&self.stats.frames_completed, 1); + } return Ok(stamp_received(frame)); } // A push that completed nothing may still have aged a partial out — deliver it diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 8bff2202..05807e7c 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -2282,17 +2282,18 @@ mod tests { mode, CompositorPref::Auto, GamepadPref::Auto, - 0, // bitrate_kbps - 0, // video_caps - 2, // audio_channels - 0, // video_codecs (HEVC-only) - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name - None, // pin (TOFU) - None, // identity (host doesn't require pairing) + 0, // bitrate_kbps + 0, // video_caps + 2, // audio_channels + 0, // video_codecs (HEVC-only) + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name + None, // pin (TOFU) + None, // identity (host doesn't require pairing) std::time::Duration::from_secs(10), ) .expect("client connects to synthetic host"); @@ -2455,15 +2456,16 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs (0 → HEVC-only) - 0, // preferred_codec (auto) - None, // display_hdr - 0, // client_caps - None, // launch - None, // name: absent on purpose — this test asserts the fingerprint-derived label - None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs (0 → HEVC-only) + 0, // preferred_codec (auto) + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name: absent on purpose — this test asserts the fingerprint-derived label + None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client Some((cert, key)), std::time::Duration::from_secs(15), ) @@ -2524,14 +2526,15 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name None, None, timeout @@ -2555,14 +2558,15 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name Some(host_fp), Some((cert.clone(), key.clone())), timeout,