fix(core,android): networking-audit small follow-ups — bounds, oversized AUs, probe flag

Networking-audit deferred plan §6:

- 6.1 client reassembler ceiling derived from the negotiated rate:
  Welcome::session_config (client role) now sets max_frame_bytes to
  clamp(4 × bitrate_kbps×125 / refresh_hz, 8 MiB, 64 MiB) instead of the
  blanket 64 MiB p1_defaults bound — the hostile-header memory ceiling was
  ~10× larger than any real access unit. Local only (the host never
  reassembles video; the wire is self-describing); a bitrate-0 (older)
  host keeps the old bound. Unit-tested floor/derived/host/old-host cases.
- 6.2 ProbeState.active is cleared when the host's ProbeResult lands, so
  the pump stops mirroring receive counters once the burst is over.
- 6.3 Android: an AU larger than the codec input buffer is DROPPED with a
  recovery-keyframe request and a counter, on both the sync (feed) and
  async (feed_ready) paths — a truncated AU is corrupt input the decoder
  chews on silently, poisoning the reference chain until the next IDR. The
  async path recycles the never-queued input slot; the sync path returns
  the dequeued slot with zero valid bytes.
- 6.4 bounded uplink channels: mic_tx at 64 (~320 ms of 5 ms frames;
  overflow sheds the fresh frame with a debug log — a tokio mpsc can't
  shed from the head, and past 320 ms of backlog the mic is broken either
  way; the bound is about memory) and ctrl_tx at 32 (sparse requests; a
  full queue means a wedged control task, reported as Closed). input_tx
  stays unbounded per the plan: keyboard/mouse events must never silently
  drop, and gamepad state is snapshot-healed.
- 6.5 (wire version byte says P1 while streaming Gf16): record-only,
  resolves with the P2 packet revision.

include/punktfunk_core.h: cbindgen re-emitted in the new module order
after the quic/ split (item 3) — no semantic change beyond the reorder.

cargo ndk check (arm64-v8a), workspace clippy, core+host tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-10 16:07:42 +02:00
parent 9afcbcd307
commit ddb93c533c
5 changed files with 189 additions and 91 deletions
+43 -18
View File
@@ -99,7 +99,8 @@ struct Negotiated {
/// completes, so the old AU-based count cliffed to zero even though most bytes still arrived.
#[derive(Default)]
struct ProbeState {
/// A probe is in progress (set by `request_probe`, cleared by nothing — the latest one wins).
/// A probe is in progress: set by `request_probe`, cleared when the host's [`ProbeResult`]
/// lands (a re-probe just overwrites the whole state — the latest one wins).
active: bool,
/// `session.stats()` receive counters at the burst's start (snapshotted by the pump on its first
/// tick while active) and latest, mirrored every pump iteration.
@@ -215,6 +216,17 @@ const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2;
/// FIRST no-op clock flush — the moment a step is actually suspected.
const CLOCK_RESYNC_INTERVAL: Duration = Duration::from_secs(60);
/// Outbound mic uplink queue depth: 5 ms Opus frames, so 64 is ~320 ms of audio — far beyond
/// any worker stall a live mic session survives anyway. On overflow the FRESH frame is dropped
/// (a tokio mpsc can't shed from the head; by the time 320 ms are queued the stream is broken
/// either way, and the bound is about memory, not audio quality) and logged at debug.
const MIC_QUEUE: usize = 64;
/// Outbound control-request queue depth. The requests are sparse (mode switches, keyframe
/// requests, ~1.3 loss reports/s, clock re-syncs) — 32 is hours of headroom; a full queue means
/// the control task is wedged, which callers treat as a closed session.
const CTRL_QUEUE: usize = 32;
/// The pre-decode video hand-off from the data-plane pump to the embedder. Unlike the side planes
/// (self-contained samples that drop the newest on overflow), video AUs are reference-chained under the
/// host's infinite GOP: dropping ANY frame mid-stream corrupts every dependent frame until the next
@@ -353,11 +365,15 @@ pub struct NativeClient {
host_timing: Mutex<Receiver<crate::quic::HostTiming>>,
input_tx: tokio::sync::mpsc::UnboundedSender<InputEvent>,
/// Outbound mic frames `(seq, pts_ns, opus)` → encoded as 0xCB datagrams by the worker.
mic_tx: tokio::sync::mpsc::UnboundedSender<(u32, u64, Vec<u8>)>,
/// Bounded ([`MIC_QUEUE`]): a wedged worker drops fresh frames (logged) instead of queueing
/// audio-latency (and memory) without limit — mic is best-effort end to end.
mic_tx: tokio::sync::mpsc::Sender<(u32, u64, Vec<u8>)>,
/// Outbound rich input (DualSense touchpad / motion) → 0xCC datagrams by the worker.
rich_input_tx: tokio::sync::mpsc::UnboundedSender<RichInput>,
/// Outbound control-stream requests (mode switch, speed test) → the worker's control task.
ctrl_tx: tokio::sync::mpsc::UnboundedSender<CtrlRequest>,
/// Bounded ([`CTRL_QUEUE`]) — the requests are sparse; a full queue means the control task
/// is wedged/dead, and callers treat it like a closed session.
ctrl_tx: tokio::sync::mpsc::Sender<CtrlRequest>,
/// Speed-test accumulator, shared with the data-plane pump + control task.
probe: Arc<Mutex<ProbeState>>,
shutdown: Arc<AtomicBool>,
@@ -549,9 +565,9 @@ impl NativeClient {
let (host_timing_tx, host_timing_rx) =
std::sync::mpsc::sync_channel::<crate::quic::HostTiming>(HOST_TIMING_QUEUE);
let (input_tx, input_rx) = tokio::sync::mpsc::unbounded_channel::<InputEvent>();
let (mic_tx, mic_rx) = tokio::sync::mpsc::unbounded_channel::<(u32, u64, Vec<u8>)>();
let (mic_tx, mic_rx) = tokio::sync::mpsc::channel::<(u32, u64, Vec<u8>)>(MIC_QUEUE);
let (rich_input_tx, rich_input_rx) = tokio::sync::mpsc::unbounded_channel::<RichInput>();
let (ctrl_tx, ctrl_rx) = tokio::sync::mpsc::unbounded_channel::<CtrlRequest>();
let (ctrl_tx, ctrl_rx) = tokio::sync::mpsc::channel::<CtrlRequest>(CTRL_QUEUE);
let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<Negotiated>>();
let shutdown = Arc::new(AtomicBool::new(false));
let quit = Arc::new(AtomicBool::new(false));
@@ -825,7 +841,7 @@ impl NativeClient {
/// reflects it. A rejected request leaves the session unchanged.
pub fn request_mode(&self, mode: Mode) -> Result<()> {
self.ctrl_tx
.send(CtrlRequest::Mode(mode))
.try_send(CtrlRequest::Mode(mode))
.map_err(|_| PunktfunkError::Closed)
}
@@ -835,7 +851,7 @@ impl NativeClient {
/// lands, so requesting on every frame would flood the control stream).
pub fn request_keyframe(&self) -> Result<()> {
self.ctrl_tx
.send(CtrlRequest::Keyframe)
.try_send(CtrlRequest::Keyframe)
.map_err(|_| PunktfunkError::Closed)
}
@@ -915,7 +931,7 @@ impl NativeClient {
..Default::default()
};
self.ctrl_tx
.send(CtrlRequest::Probe(ProbeRequest {
.try_send(CtrlRequest::Probe(ProbeRequest {
target_kbps,
duration_ms,
}))
@@ -1061,9 +1077,17 @@ impl NativeClient {
/// uses them only for diagnostics). The host decodes it into a virtual microphone source.
/// Best-effort — like every datagram, it's dropped under loss; no retransmit.
pub fn send_mic(&self, seq: u32, pts_ns: u64, opus: Vec<u8>) -> Result<()> {
self.mic_tx
.send((seq, pts_ns, opus))
.map_err(|_| PunktfunkError::Closed)
use tokio::sync::mpsc::error::TrySendError;
match self.mic_tx.try_send((seq, pts_ns, opus)) {
Ok(()) => Ok(()),
Err(TrySendError::Full(_)) => {
// Bounded queue full = the worker stalled for ~MIC_QUEUE x 5 ms. Shed this
// frame (mic is best-effort end to end) instead of queueing latency/memory.
tracing::debug!("mic uplink queue full — dropping frame");
Ok(())
}
Err(TrySendError::Closed(_)) => Err(PunktfunkError::Closed),
}
}
/// Queue one rich input event (DualSense touchpad contact or motion sample) for delivery as a
@@ -1115,10 +1139,10 @@ struct WorkerArgs {
hdr_meta_tx: SyncSender<HdrMeta>,
host_timing_tx: SyncSender<crate::quic::HostTiming>,
input_rx: tokio::sync::mpsc::UnboundedReceiver<InputEvent>,
mic_rx: tokio::sync::mpsc::UnboundedReceiver<(u32, u64, Vec<u8>)>,
mic_rx: tokio::sync::mpsc::Receiver<(u32, u64, Vec<u8>)>,
rich_input_rx: tokio::sync::mpsc::UnboundedReceiver<RichInput>,
ctrl_rx: tokio::sync::mpsc::UnboundedReceiver<CtrlRequest>,
ctrl_tx: tokio::sync::mpsc::UnboundedSender<CtrlRequest>,
ctrl_rx: tokio::sync::mpsc::Receiver<CtrlRequest>,
ctrl_tx: tokio::sync::mpsc::Sender<CtrlRequest>,
ready_tx: std::sync::mpsc::Sender<Result<Negotiated>>,
shutdown: Arc<AtomicBool>,
/// Deliberate-quit flag (see [`NativeClient::quit`]): the worker closes with the quit code if set.
@@ -1472,6 +1496,7 @@ async fn worker_main(args: WorkerArgs) {
p.host_send_dropped = result.send_dropped;
p.host_duration_ms = result.duration_ms;
p.done = true;
p.active = false; // burst over — the pump stops mirroring counters
tracing::info!(
host_goodput_bytes = result.bytes_sent,
wire_packets_sent = result.wire_packets_sent,
@@ -1664,7 +1689,7 @@ async fn worker_main(args: WorkerArgs) {
// the mid-stream re-sync now (once — the 60 s periodic covers everything else).
if resync_wanted {
resync_wanted = false;
let _ = ctrl_tx.send(CtrlRequest::ClockResync);
let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
}
let window_dropped = st.frames_dropped.wrapping_sub(last_dropped);
let loss_ppm = window_loss_ppm(
@@ -1672,7 +1697,7 @@ async fn worker_main(args: WorkerArgs) {
st.packets_received.wrapping_sub(last_received),
window_dropped,
);
let _ = ctrl_tx.send(CtrlRequest::Loss(LossReport { loss_ppm }));
let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm }));
// Adaptive bitrate: drain any host ack first (its clamp is authoritative), then
// feed the controller this window's congestion signals; a decision becomes a
// SetBitrate on the control stream.
@@ -1690,7 +1715,7 @@ async fn worker_main(args: WorkerArgs) {
flush_in_window,
) {
tracing::info!(kbps, "adaptive bitrate: requesting encoder re-target");
let _ = ctrl_tx.send(CtrlRequest::SetBitrate(kbps));
let _ = ctrl_tx.try_send(CtrlRequest::SetBitrate(kbps));
}
flush_in_window = false;
last_report = Instant::now();
@@ -1762,7 +1787,7 @@ async fn worker_main(args: WorkerArgs) {
flush_in_window = true; // strongest "link can't hold the rate" signal
let flushed = session.flush_backlog().unwrap_or(0);
let dropped = frames.clear();
let _ = ctrl_tx.send(CtrlRequest::Keyframe);
let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
tracing::warn!(
behind_ms = if clock_behind { lat_ns / 1_000_000 } else { -1 },
queue_depth = depth,
+11
View File
@@ -902,6 +902,17 @@ impl Welcome {
c.encrypt = self.encrypt;
c.key = self.key;
c.salt = self.salt;
// Client-side reassembler ceiling: p1_defaults' 64 MiB hostile-header memory bound is
// ~10x larger than any real access unit. Derive it from the negotiated rate instead:
// 4x the average frame size at the resolved bitrate (IDR headroom), floored at 8 MiB,
// capped at the old 64 MiB. Purely local — the host never reassembles video and the
// wire is self-describing, so old hosts are unaffected; a host that reports bitrate 0
// (pre-negotiation) keeps the old bound.
if role == Role::Client && self.bitrate_kbps > 0 {
let per_frame = (self.bitrate_kbps as usize).saturating_mul(125)
/ self.mode.refresh_hz.max(1) as usize;
c.max_frame_bytes = per_frame.saturating_mul(4).clamp(8 << 20, 64 << 20);
}
c
}
}
+30 -1
View File
@@ -1,5 +1,5 @@
use super::*;
use crate::config::{CompositorPref, FecConfig, FecScheme, GamepadPref, Mode};
use crate::config::{CompositorPref, FecConfig, FecScheme, GamepadPref, Mode, Role};
#[test]
fn welcome_roundtrip() {
@@ -32,6 +32,35 @@ fn welcome_roundtrip() {
host_caps: HOST_CAP_GAMEPAD_STATE,
};
assert_eq!(Welcome::decode(&w.encode()).unwrap(), w);
// Client-side reassembler ceiling derives from the negotiated rate: 4x the average frame at
// 50 Mbps/240 Hz is ~104 KB, so the 8 MiB floor governs. The host keeps the p1_defaults
// bound (it never reassembles video), as does a client of a bitrate-0 (older) host.
let cc = w.session_config(Role::Client);
assert_eq!(cc.max_frame_bytes, 8 << 20);
cc.validate().expect("derived client config validates");
assert_eq!(w.session_config(Role::Host).max_frame_bytes, 64 << 20);
let old_host = Welcome {
bitrate_kbps: 0,
..w.clone()
};
assert_eq!(
old_host.session_config(Role::Client).max_frame_bytes,
64 << 20
);
// A high-rate mode scales past the floor: 1.5 Gbps at 60 Hz = 4 x 3.125 MB = 12.5 MB.
let fat = Welcome {
bitrate_kbps: 1_500_000,
mode: Mode {
width: 5120,
height: 1440,
refresh_hz: 60,
},
..w.clone()
};
let derived = fat.session_config(Role::Client).max_frame_bytes;
assert_eq!(derived, 4 * 1_500_000 * 125 / 60);
assert!(derived > (8 << 20) && derived < (64 << 20));
}
#[test]