bstein/lesavka
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

media: add mjpeg uvc quality telemetry

Browse Source
This commit is contained in:
Brad Stein 2026-05-14 05:18:36 -03:00
parent dec332ea40
commit 22dd45aa39
16 changed files with 748 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.lock generated
View File

@ -1652,7 +1652,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
[[package]]
name = "lesavka_client"
version = "0.22.29"
version = "0.22.30"
dependencies = [
"anyhow",
"async-stream",
@ -1686,7 +1686,7 @@ dependencies = [
[[package]]
name = "lesavka_common"
version = "0.22.29"
version = "0.22.30"
dependencies = [
"anyhow",
"base64",
@ -1698,7 +1698,7 @@ dependencies = [
[[package]]
name = "lesavka_server"
version = "0.22.29"
version = "0.22.30"
dependencies = [
"anyhow",
"base64",

2
client/Cargo.toml
View File

@ -4,7 +4,7 @@ path = "src/main.rs"
[package]
name = "lesavka_client"
version = "0.22.29"
version = "0.22.30"
edition = "2024"
[dependencies]

2
common/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "lesavka_common"
version = "0.22.29"
version = "0.22.30"
edition = "2024"
build = "build.rs"

4
docs/operational-env.md
View File

@ -330,8 +330,9 @@ from `LESAVKA_CLIENT_PKI_SSH_SOURCE` over SSH. Runtime clients require the insta
| `LESAVKA_UVC_FRAME_META_LOG_PATH` | UVC helper diagnostic override; when set with `LESAVKA_UVC_FRAME_META=1`, append every MJPEG spool timing record as JSONL for full-probe HEVC/RCT correlation; summarize with `scripts/manual/summarize_uvc_frame_meta_log.py` |
| `LESAVKA_UVC_FRAME_META_PATH` | UVC helper diagnostic override; explicit path for the optional MJPEG spool metadata sidecar |
| `LESAVKA_UVC_FRAME_MAX_AGE_MS` | UVC helper freshness override; stale spooled MJPEG frames older than this are not replayed, defaults to `1000`; `0` disables TTL |
| `LESAVKA_UVC_FRAME_MAX_BYTES` | UVC helper MJPEG frame-size guard; explicit maximum accepted frame bytes, where `0` disables the guard and otherwise oversized frames are frozen out |
| `LESAVKA_UVC_FRAME_MAX_BYTES` | UVC helper MJPEG frame-size guard; explicit maximum accepted frame bytes. Unset or `0` uses the live-call byte budget so oversized frames freeze instead of tearing on the host |
| `LESAVKA_UVC_FRAME_SIZE` | server hardware/device override |
| `LESAVKA_UVC_FRAME_SIZE_GUARD` | UVC helper MJPEG frame-size guard toggle; defaults to `1`; set `0` only for diagnostics when oversized MJPEG frames must be allowed through |
| `LESAVKA_UVC_HEIGHT` | server hardware/device override |
| `LESAVKA_UVC_HEVC_SPOOL_PULL_TIMEOUT_MS` | server HEVC decode-to-MJPEG freshness override; appsink pull wait for decoded MJPEG handoff before publishing newest frame to the UVC helper, defaults to `20` and is capped at `50` |
| `LESAVKA_UVC_HEVC_FRESHNESS_QUEUE_BUFFERS` | server HEVC decode-to-MJPEG branch queue depth; defaults to `2` and is capped at `4` so decode/JPEG scheduling jitter does not starve the UVC helper while stale frames still get dropped |
@ -344,6 +345,7 @@ from `LESAVKA_CLIENT_PKI_SSH_SOURCE` over SSH. Runtime clients require the insta
| `LESAVKA_UVC_MJPEG` | server hardware/device override |
| `LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC` | UVC helper MJPEG budget guard; derives a per-frame byte cap from target FPS when `LESAVKA_UVC_FRAME_MAX_BYTES` is unset |
| `LESAVKA_UVC_SKIP_UDEV` | server hardware/device override |
| `LESAVKA_UVC_STATS_INTERVAL_MS` | UVC helper telemetry interval for queued/reloaded/rejected MJPEG frame counters; defaults to `5000`, `0` disables |
| `LESAVKA_UVC_STREAMING_INTERVAL` | server hardware/device override |
| `LESAVKA_UVC_STREAM_INTF` | server hardware/device override |
| `LESAVKA_UVC_WIDTH` | server hardware/device override |

1
scripts/install/server.sh
View File

@ -355,6 +355,7 @@ LESAVKA_UVC_CODEC=${INSTALL_UVC_CODEC}
LESAVKA_UVC_BLOCKING=$(uvc_env_value LESAVKA_UVC_BLOCKING 1)
LESAVKA_UVC_CONTROL_READ_ONLY=$(uvc_env_value LESAVKA_UVC_CONTROL_READ_ONLY 0)
LESAVKA_UVC_MAXBURST=$(uvc_env_value LESAVKA_UVC_MAXBURST 0)
LESAVKA_UVC_FRAME_SIZE_GUARD=$(uvc_env_value LESAVKA_UVC_FRAME_SIZE_GUARD 1)
LESAVKA_UVC_FRAME_MAX_BYTES=$(uvc_env_value LESAVKA_UVC_FRAME_MAX_BYTES 0)
LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC=$(uvc_env_value LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC 10000000)
EOF

129
scripts/manual/analyze_uvc_mjpeg_frame.py Executable file
View File

@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""Inspect a Lesavka UVC MJPEG spool frame for size/profile corruption clues."""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
from typing import Optional
MAX_MJPEG_FRAME_BYTES = 8 * 1024 * 1024
DEFAULT_BUDGET_BYTES_PER_SEC = 10_000_000
def read_be16(data: bytes, offset: int) -> Optional[int]:
if offset + 2 > len(data):
return None
return int.from_bytes(data[offset : offset + 2], "big")
def marker_has_length(marker: int) -> bool:
return marker != 0x01 and not 0xD0 <= marker <= 0xD9
def is_sof(marker: int) -> bool:
return marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}
def inspect_jpeg(data: bytes) -> dict[str, object]:
complete = len(data) >= 4 and data.startswith(b"\xff\xd8") and data.endswith(b"\xff\xd9") and b"\xff\xda" in data
width = None
height = None
entropy_start = None
entropy_end = len(data) - 2 if len(data) >= 2 else len(data)
idx = 2
while idx + 4 <= len(data):
if data[idx] != 0xFF:
idx += 1
continue
while idx < len(data) and data[idx] == 0xFF:
idx += 1
if idx >= len(data):
break
marker = data[idx]
idx += 1
if marker == 0xDA:
seg_len = read_be16(data, idx)
if seg_len is not None and seg_len >= 2 and idx + seg_len <= len(data):
entropy_start = idx + seg_len
break
if marker == 0xD9:
break
if not marker_has_length(marker):
continue
seg_len = read_be16(data, idx)
if seg_len is None or seg_len < 2 or idx + seg_len > len(data):
break
if is_sof(marker) and seg_len >= 7:
height = read_be16(data, idx + 3)
width = read_be16(data, idx + 5)
idx += seg_len
payload = data[entropy_start:entropy_end] if entropy_start is not None and entropy_end > entropy_start else b""
counts = [0] * 256
max_run = 0
run = 0
prev = None
for byte in payload:
counts[byte] += 1
if byte == prev:
run += 1
else:
prev = byte
run = 1
max_run = max(max_run, run)
dominant = max(counts) if payload else 0
distinct = sum(1 for count in counts if count)
dominant_pct = round(dominant * 100 / len(payload), 2) if payload else 0.0
return {
"bytes": len(data),
"complete": complete,
"width": width,
"height": height,
"entropy_bytes": len(payload),
"entropy_distinct_bytes": distinct,
"entropy_dominant_pct": dominant_pct,
"entropy_max_run": max_run,
}
def derived_cap(fps: int, budget: int, explicit: Optional[int], guard: bool) -> int:
if not guard:
return MAX_MJPEG_FRAME_BYTES
if explicit and explicit > 0:
return min(explicit, MAX_MJPEG_FRAME_BYTES)
per_frame = max(budget // max(fps, 1), 64 * 1024)
return min(per_frame, MAX_MJPEG_FRAME_BYTES)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("frame", nargs="?", default=os.environ.get("LESAVKA_UVC_FRAME_PATH", "/run/lesavka-uvc-frame.mjpg"))
parser.add_argument("--fps", type=int, default=int(os.environ.get("LESAVKA_UVC_FPS", "30") or "30"))
parser.add_argument("--budget-bytes-per-sec", type=int, default=int(os.environ.get("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", str(DEFAULT_BUDGET_BYTES_PER_SEC)) or str(DEFAULT_BUDGET_BYTES_PER_SEC)))
parser.add_argument("--max-bytes", type=int, default=int(os.environ.get("LESAVKA_UVC_FRAME_MAX_BYTES", "0") or "0"))
parser.add_argument("--disable-size-guard", action="store_true")
args = parser.parse_args()
path = Path(args.frame)
data = path.read_bytes()
cap = derived_cap(args.fps, args.budget_bytes_per_sec, args.max_bytes, not args.disable_size_guard)
report = inspect_jpeg(data)
report.update(
{
"path": str(path),
"fps": args.fps,
"budget_bytes_per_sec": args.budget_bytes_per_sec,
"max_bytes": cap,
"over_budget": len(data) > cap,
}
)
print(json.dumps(report, indent=2, sort_keys=True))
return 1 if report["over_budget"] or not report["complete"] else 0
if __name__ == "__main__":
raise SystemExit(main())

2
server/Cargo.toml
View File

@ -10,7 +10,7 @@ bench = false
[package]
name = "lesavka_server"
version = "0.22.29"
version = "0.22.30"
edition = "2024"
autobins = false

99
server/src/bin/lesavka-uvc.real.inc
View File

@ -6,7 +6,7 @@ use std::fs::{File, OpenOptions};
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::io::{AsRawFd, RawFd};
use std::thread;
use std::time::{Duration, SystemTime};
use std::time::{Duration, Instant, SystemTime};
const STREAM_CTRL_SIZE_11: usize = 26;
const STREAM_CTRL_SIZE_15: usize = 34;
@ -52,6 +52,7 @@ const DEFAULT_UVC_BUFFER_COUNT: u32 = 2;
const DEFAULT_UVC_IDLE_PUMP_MS: u64 = 2;
const DEFAULT_UVC_FRAME_MAX_AGE_MS: u64 = 1_000;
const DEFAULT_UVC_MJPEG_BUDGET_BYTES_PER_SEC: u32 = 10_000_000;
const DEFAULT_UVC_STATS_INTERVAL_MS: u64 = 5_000;
#[repr(C)]
struct V4l2EventSubscription {
@ -243,6 +244,19 @@ struct UvcVideoStream {
latest_frame: Vec<u8>,
frame_max_bytes: usize,
streaming: bool,
stats: UvcVideoStats,
}
#[derive(Default)]
struct UvcVideoStats {
queued: u64,
reloaded: u64,
replayed_stale: u64,
rejected_oversize: u64,
rejected_invalid: u64,
fallback_idle: u64,
latest_bytes: usize,
last_report: Option<Instant>,
}
impl UvcVideoStream {
@ -254,6 +268,7 @@ impl UvcVideoStream {
latest_frame: IDLE_MJPEG_FRAME.to_vec(),
frame_max_bytes: MAX_MJPEG_FRAME_BYTES,
streaming: false,
stats: UvcVideoStats::default(),
}
}
@ -431,21 +446,34 @@ impl UvcVideoStream {
if rc < 0 {
return Err(std::io::Error::last_os_error()).context("VIDIOC_QBUF");
}
self.stats.queued += 1;
self.stats.latest_bytes = bytes;
self.report_stats_if_due();
Ok(())
}
fn refresh_latest_frame(&mut self) {
let stale = frame_spool_is_stale(&self.frame_path, frame_spool_max_age());
if stale && looks_like_mjpeg_frame(&self.latest_frame) {
self.stats.replayed_stale += 1;
return;
}
let max_frame_bytes = self.frame_payload_limit();
if let Ok(frame) = std::fs::read(&self.frame_path)
&& frame.len() <= max_frame_bytes
&& looks_like_mjpeg_frame(&frame)
{
self.latest_frame = frame;
} else if !looks_like_mjpeg_frame(&self.latest_frame) {
match std::fs::read(&self.frame_path) {
Ok(frame) if !looks_like_mjpeg_frame(&frame) => {
self.stats.rejected_invalid += 1;
}
Ok(frame) if frame.len() > max_frame_bytes => {
self.stats.rejected_oversize += 1;
}
Ok(frame) => {
self.stats.reloaded += 1;
self.latest_frame = frame;
}
Err(_) => {}
}
if !looks_like_mjpeg_frame(&self.latest_frame) {
self.stats.fallback_idle += 1;
self.latest_frame = IDLE_MJPEG_FRAME.to_vec();
}
}
@ -468,6 +496,32 @@ impl UvcVideoStream {
MINIMAL_MJPEG_FRAME
}
}
fn report_stats_if_due(&mut self) {
let Some(interval) = uvc_stats_interval() else {
return;
};
let now = Instant::now();
if self
.stats
.last_report
.is_some_and(|last| now.duration_since(last) < interval)
{
return;
}
self.stats.last_report = Some(now);
eprintln!(
"[lesavka-uvc] video stats queued={} reloaded={} stale_replay={} rejected_oversize={} rejected_invalid={} fallback_idle={} latest_bytes={} frame_cap={}",
self.stats.queued,
self.stats.reloaded,
self.stats.replayed_stale,
self.stats.rejected_oversize,
self.stats.rejected_invalid,
self.stats.fallback_idle,
self.stats.latest_bytes,
self.frame_payload_limit()
);
}
}
impl Drop for UvcVideoStream {
@ -536,14 +590,21 @@ fn uvc_idle_pump_sleep() -> Duration {
/// half-frame grey smears, and freezing the last good frame is preferable to
/// queueing a frame that is likely to arrive incomplete.
fn uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
if !uvc_frame_size_guard_enabled() {
return MAX_MJPEG_FRAME_BYTES;
}
if let Some(limit) = env_u32_opt("LESAVKA_UVC_FRAME_MAX_BYTES") {
return if limit == 0 {
MAX_MJPEG_FRAME_BYTES
derived_uvc_frame_max_bytes(cfg)
} else {
limit as usize
(limit as usize).min(MAX_MJPEG_FRAME_BYTES)
};
}
derived_uvc_frame_max_bytes(cfg)
}
fn derived_uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
let fps = cfg.fps.max(1);
let budget_per_sec = env_u32(
"LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC",
@ -554,6 +615,26 @@ fn uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
per_frame.min(MAX_MJPEG_FRAME_BYTES as u32) as usize
}
fn uvc_frame_size_guard_enabled() -> bool {
env::var("LESAVKA_UVC_FRAME_SIZE_GUARD")
.ok()
.map(|value| {
let trimmed = value.trim();
!(trimmed.eq_ignore_ascii_case("0")
|| trimmed.eq_ignore_ascii_case("false")
|| trimmed.eq_ignore_ascii_case("no")
|| trimmed.eq_ignore_ascii_case("off"))
})
.unwrap_or(true)
}
fn uvc_stats_interval() -> Option<Duration> {
match env_u64("LESAVKA_UVC_STATS_INTERVAL_MS", DEFAULT_UVC_STATS_INTERVAL_MS) {
0 => None,
value => Some(Duration::from_millis(value)),
}
}
fn frame_spool_max_age() -> Option<Duration> {
match env_u64(
"LESAVKA_UVC_FRAME_MAX_AGE_MS",

2
server/src/bin/lesavka_uvc/coverage_model.rs
View File

@ -56,6 +56,8 @@ const DEFAULT_UVC_FRAME_MAX_AGE_MS: u64 = 1_000;
#[cfg(coverage)]
const DEFAULT_UVC_MJPEG_BUDGET_BYTES_PER_SEC: u32 = 10_000_000;
#[cfg(coverage)]
const DEFAULT_UVC_STATS_INTERVAL_MS: u64 = 5_000;
#[cfg(coverage)]
const MAX_MJPEG_FRAME_BYTES: usize = 8 * 1024 * 1024;
#[cfg(coverage)]
const MINIMAL_MJPEG_FRAME: &[u8] = &[0xff, 0xd8, 0xff, 0xd9];

34
server/src/bin/lesavka_uvc/coverage_startup.rs
View File

@ -159,14 +159,22 @@ fn uvc_idle_pump_sleep() -> std::time::Duration {
/// byte length. Why: coverage tests should lock the artifact-prevention budget
/// that turns oversized UVC frames into freezes instead of grey half-frames.
fn uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
if !uvc_frame_size_guard_enabled() {
return MAX_MJPEG_FRAME_BYTES;
}
if let Some(limit) = env_u32_opt("LESAVKA_UVC_FRAME_MAX_BYTES") {
return if limit == 0 {
MAX_MJPEG_FRAME_BYTES
derived_uvc_frame_max_bytes(cfg)
} else {
limit as usize
(limit as usize).min(MAX_MJPEG_FRAME_BYTES)
};
}
derived_uvc_frame_max_bytes(cfg)
}
#[cfg(coverage)]
fn derived_uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
let fps = cfg.fps.max(1);
let budget_per_sec = env_u32(
"LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC",
@ -177,6 +185,28 @@ fn uvc_frame_max_bytes(cfg: UvcConfig) -> usize {
per_frame.min(MAX_MJPEG_FRAME_BYTES as u32) as usize
}
#[cfg(coverage)]
fn uvc_frame_size_guard_enabled() -> bool {
env::var("LESAVKA_UVC_FRAME_SIZE_GUARD")
.ok()
.map(|value| {
let trimmed = value.trim();
!(trimmed.eq_ignore_ascii_case("0")
|| trimmed.eq_ignore_ascii_case("false")
|| trimmed.eq_ignore_ascii_case("no")
|| trimmed.eq_ignore_ascii_case("off"))
})
.unwrap_or(true)
}
#[cfg(coverage)]
fn uvc_stats_interval() -> Option<std::time::Duration> {
match env_u64("LESAVKA_UVC_STATS_INTERVAL_MS", DEFAULT_UVC_STATS_INTERVAL_MS) {
0 => None,
value => Some(std::time::Duration::from_millis(value)),
}
}
#[cfg(coverage)]
/// Returns the optional maximum age for a spooled MJPEG frame.
///

18
server/src/bin/tests/lesavka_uvc.rs
View File

@ -73,6 +73,7 @@ fn io_helpers_cover_empty_and_missing_sources() {
fn uvc_frame_max_bytes_defaults_to_freshness_budget_and_allows_override() {
with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", None::<&str>),
("LESAVKA_UVC_FRAME_MAX_BYTES", None::<&str>),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", None::<&str>),
],
@ -84,6 +85,7 @@ fn uvc_frame_max_bytes_defaults_to_freshness_budget_and_allows_override() {
with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", Some("1")),
("LESAVKA_UVC_FRAME_MAX_BYTES", Some("123456")),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", Some("1")),
],
@ -93,7 +95,21 @@ fn uvc_frame_max_bytes_defaults_to_freshness_budget_and_allows_override() {
);
with_var("LESAVKA_UVC_FRAME_MAX_BYTES", Some("0"), || {
assert_eq!(uvc_frame_max_bytes(sample_cfg()), MAX_MJPEG_FRAME_BYTES);
assert_eq!(uvc_frame_max_bytes(sample_cfg()), 400_000);
});
with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", Some("0")),
("LESAVKA_UVC_FRAME_MAX_BYTES", Some("123456")),
],
|| {
assert_eq!(uvc_frame_max_bytes(sample_cfg()), MAX_MJPEG_FRAME_BYTES);
},
);
with_var("LESAVKA_UVC_STATS_INTERVAL_MS", Some("0"), || {
assert_eq!(uvc_stats_interval(), None);
});
}

319
server/src/video_sinks/hevc_mjpeg_guard.rs
View File

@ -7,6 +7,39 @@ const DEFAULT_HEVC_MIN_PAYLOAD_DISTINCT_BYTES: u32 = 12;
const DEFAULT_HEVC_DOMINANT_BYTE_PCT: u32 = 92;
const DEFAULT_DIRECT_MJPEG_SIZE_DROP_PCT: u32 = 18;
const DEFAULT_DIRECT_MJPEG_MIN_REFERENCE_BYTES: u32 = 48 * 1024;
const DEFAULT_DIRECT_MJPEG_PROFILE_MISMATCH_REJECT: bool = false;
/// Summarizes one compressed MJPEG frame without fully decoding pixels.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(super) struct MjpegFrameInspection {
pub bytes: usize,
pub complete: bool,
pub width: Option<u16>,
pub height: Option<u16>,
pub entropy_bytes: usize,
pub entropy_distinct_bytes: u16,
pub entropy_dominant_pct: u8,
pub entropy_max_run: usize,
}
/// Explains why a direct MJPEG frame was frozen before UVC handoff.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum DirectMjpegRejectReason {
Incomplete,
Oversized {
max_bytes: usize,
},
ProfileMismatch {
expected_width: u16,
expected_height: u16,
actual_width: u16,
actual_height: u16,
},
FlatPayload,
SizeCollapse {
threshold_bytes: u64,
},
}
/// Resolve the JPEG quality used after HEVC decode.
///
@ -133,6 +166,25 @@ pub(super) fn direct_mjpeg_min_reference_bytes() -> u32 {
.max(1)
}
/// Decide whether direct MJPEG frames must match the active UVC dimensions.
///
/// Inputs: optional `LESAVKA_UVC_DIRECT_MJPEG_REJECT_PROFILE_MISMATCH`.
/// Output: false unless explicitly enabled. Why: current field debugging needs
/// profile-mismatch telemetry first; rejecting mismatches by default could
/// accidentally freeze every frame on an attached gadget with stale descriptors.
pub(super) fn direct_mjpeg_reject_profile_mismatch_enabled() -> bool {
std::env::var("LESAVKA_UVC_DIRECT_MJPEG_REJECT_PROFILE_MISMATCH")
.ok()
.map(|value| {
let trimmed = value.trim();
trimmed.eq_ignore_ascii_case("1")
|| trimmed.eq_ignore_ascii_case("true")
|| trimmed.eq_ignore_ascii_case("yes")
|| trimmed.eq_ignore_ascii_case("on")
})
.unwrap_or(DEFAULT_DIRECT_MJPEG_PROFILE_MISMATCH_REJECT)
}
/// Return whether a decoded buffer looks like one complete JPEG image.
///
/// Inputs: decoded MJPEG bytes. Output: true when SOI, SOS, and EOI markers
@ -145,6 +197,149 @@ fn looks_like_complete_jpeg(bytes: &[u8]) -> bool {
&& bytes.windows(2).any(|pair| pair == [0xff, 0xda])
}
fn marker_has_length(marker: u8) -> bool {
!matches!(marker, 0x01 | 0xd0..=0xd9)
}
fn is_start_of_frame_marker(marker: u8) -> bool {
matches!(
marker,
0xc0 | 0xc1 | 0xc2 | 0xc3 | 0xc5 | 0xc6 | 0xc7 | 0xc9 | 0xca | 0xcb | 0xcd | 0xce
| 0xcf
)
}
fn read_be_u16(bytes: &[u8], offset: usize) -> Option<u16> {
bytes
.get(offset..offset + 2)
.map(|value| u16::from_be_bytes([value[0], value[1]]))
}
fn jpeg_entropy_range(bytes: &[u8]) -> Option<std::ops::Range<usize>> {
let mut idx = 2;
while idx + 4 <= bytes.len() {
if bytes[idx] != 0xff {
idx += 1;
continue;
}
while idx < bytes.len() && bytes[idx] == 0xff {
idx += 1;
}
let marker = *bytes.get(idx)?;
idx += 1;
if marker == 0xd9 {
return None;
}
if !marker_has_length(marker) {
continue;
}
let segment_len = usize::from(read_be_u16(bytes, idx)?);
if marker == 0xda && (segment_len < 2 || idx + segment_len > bytes.len()) {
let start = idx.min(bytes.len());
let end = bytes.len().saturating_sub(2);
return (end > start).then_some(start..end);
}
if segment_len < 2 || idx + segment_len > bytes.len() {
return None;
}
if marker == 0xda {
let start = idx + segment_len;
let end = bytes.len().saturating_sub(2);
return (end > start).then_some(start..end);
}
idx += segment_len;
}
None
}
/// Inspect one MJPEG payload without full pixel decode.
///
/// Inputs: compressed JPEG bytes. Output: dimensions and entropy-shape metrics.
/// Why: these metrics make the guard explainable and cheap enough to run on
/// every UVC-bound frame, while still catching black slabs, truncation, and
/// profile mismatches that byte-size checks alone cannot distinguish.
pub(super) fn inspect_mjpeg_frame(bytes: &[u8]) -> MjpegFrameInspection {
let complete = looks_like_complete_jpeg(bytes);
let mut width = None;
let mut height = None;
let mut idx = 2;
while idx + 4 <= bytes.len() {
if bytes[idx] != 0xff {
idx += 1;
continue;
}
while idx < bytes.len() && bytes[idx] == 0xff {
idx += 1;
}
let Some(marker) = bytes.get(idx).copied() else {
break;
};
idx += 1;
if marker == 0xda || marker == 0xd9 {
break;
}
if !marker_has_length(marker) {
continue;
}
let Some(segment_len) = read_be_u16(bytes, idx).map(usize::from) else {
break;
};
if segment_len < 2 || idx + segment_len > bytes.len() {
break;
}
if is_start_of_frame_marker(marker)
&& segment_len >= 7
&& let (Some(frame_height), Some(frame_width)) =
(read_be_u16(bytes, idx + 3), read_be_u16(bytes, idx + 5))
{
height = Some(frame_height);
width = Some(frame_width);
break;
}
idx += segment_len;
}
let mut inspection = MjpegFrameInspection {
bytes: bytes.len(),
complete,
width,
height,
..MjpegFrameInspection::default()
};
let Some(range) = jpeg_entropy_range(bytes) else {
return inspection;
};
let payload = &bytes[range];
if payload.is_empty() {
return inspection;
}
let mut counts = [0u32; 256];
let mut max_run = 1usize;
let mut current_run = 0usize;
let mut previous = None;
for byte in payload {
counts[*byte as usize] += 1;
if previous == Some(*byte) {
current_run += 1;
} else {
current_run = 1;
previous = Some(*byte);
}
max_run = max_run.max(current_run);
}
let distinct = counts.iter().filter(|count| **count > 0).count() as u16;
let dominant = counts.iter().copied().max().unwrap_or(0) as u64;
let total = payload.len() as u64;
inspection.entropy_bytes = payload.len();
inspection.entropy_distinct_bytes = distinct;
inspection.entropy_dominant_pct =
((dominant.saturating_mul(100) + total.saturating_sub(1)) / total) as u8;
inspection.entropy_max_run = max_run;
inspection
}
/// Return whether one complete JPEG has an implausibly flat payload.
///
/// Inputs: decoded MJPEG bytes. Output: true for dominant-byte or very low
@ -154,29 +349,13 @@ fn suspiciously_flat_payload(bytes: &[u8]) -> bool {
if bytes.len() < min_reference_bytes() as usize / 4 {
return false;
}
let start = bytes
.windows(2)
.position(|pair| pair == [0xff, 0xda])
.map(|idx| (idx + 2).min(bytes.len()))
.unwrap_or_else(|| bytes.len().min(256));
let end = bytes.len().saturating_sub(2);
if end <= start || end - start < 512 {
let inspection = inspect_mjpeg_frame(bytes);
if inspection.entropy_bytes < 512 {
return false;
}
let payload = &bytes[start..end];
let mut counts = [0u32; 256];
for byte in payload {
counts[*byte as usize] += 1;
}
let distinct = counts.iter().filter(|count| **count > 0).count() as u32;
let dominant = counts.iter().copied().max().unwrap_or(0) as u64;
let total = payload.len() as u64;
distinct < min_payload_distinct_bytes()
|| dominant.saturating_mul(100) >= total.saturating_mul(u64::from(dominant_byte_pct()))
u32::from(inspection.entropy_distinct_bytes) < min_payload_distinct_bytes()
|| u32::from(inspection.entropy_dominant_pct) >= dominant_byte_pct()
}
/// Decide whether a decoded HEVC-to-MJPEG frame should be frozen out.
@ -218,19 +397,59 @@ pub(super) fn should_freeze_decoded_mjpeg_frame(previous_bytes: u64, decoded_mjp
/// implausibly flat, or a dramatic size collapse. Why: direct MJPEG should be
/// less aggressive than decoded HEVC filtering, but complete black/collapsed
/// frames are still worse than a short last-good-frame freeze.
#[allow(dead_code)]
pub(super) fn should_reject_direct_mjpeg_frame(previous_bytes: u64, mjpeg: &[u8]) -> bool {
direct_mjpeg_reject_reason(previous_bytes, None, None, mjpeg).is_some()
}
/// Return the concrete direct-MJPEG freeze reason, if any.
///
/// Inputs: last accepted frame size, optional UVC byte budget/profile, and the
/// next MJPEG payload. Output: a rejection reason or `None`. Why: the UVC path
/// needs operator-visible evidence for freezes; this avoids another round of
/// opaque threshold guessing when the RCT preview jumps or tears.
pub(super) fn direct_mjpeg_reject_reason(
previous_bytes: u64,
max_bytes: Option<usize>,
expected_profile: Option<(u16, u16)>,
mjpeg: &[u8],
) -> Option<DirectMjpegRejectReason> {
let inspection = inspect_mjpeg_frame(mjpeg);
if !looks_like_complete_jpeg(mjpeg) {
return true;
return Some(DirectMjpegRejectReason::Incomplete);
}
if let Some(max_bytes) = max_bytes
&& mjpeg.len() > max_bytes
{
return Some(DirectMjpegRejectReason::Oversized { max_bytes });
}
if direct_mjpeg_reject_profile_mismatch_enabled()
&& let (Some((expected_width, expected_height)), Some(actual_width), Some(actual_height)) =
(expected_profile, inspection.width, inspection.height)
&& (actual_width, actual_height) != (expected_width, expected_height)
{
return Some(DirectMjpegRejectReason::ProfileMismatch {
expected_width,
expected_height,
actual_width,
actual_height,
});
}
if !direct_mjpeg_visual_guard_enabled()
|| previous_bytes < u64::from(direct_mjpeg_min_reference_bytes())
{
return false;
return None;
}
let threshold_bytes = previous_bytes.saturating_mul(u64::from(direct_mjpeg_size_drop_pct()))
/ 100;
suspiciously_flat_payload(mjpeg) || (mjpeg.len() as u64) < threshold_bytes
if suspiciously_flat_payload(mjpeg) {
return Some(DirectMjpegRejectReason::FlatPayload);
}
if (mjpeg.len() as u64) < threshold_bytes {
return Some(DirectMjpegRejectReason::SizeCollapse { threshold_bytes });
}
None
}
#[cfg(test)]
@ -367,6 +586,60 @@ mod tests {
);
}
#[test]
fn direct_mjpeg_guard_reports_oversize_and_profile_mismatch_when_configured() {
fn jpeg_with_sof(width: u16, height: u16, payload: &[u8]) -> Vec<u8> {
let mut bytes = vec![
0xff, 0xd8, // SOI
0xff, 0xc0, 0x00, 0x11, 0x08, // SOF0 len + precision
(height >> 8) as u8,
height as u8,
(width >> 8) as u8,
width as u8,
0x03, 0x01, 0x11, 0x00, 0x02, 0x11, 0x00, 0x03, 0x11, 0x00,
0xff, 0xda, 0x00, 0x08, 0x01, 0x01, 0x00, 0x00, 0x3f, 0x00,
];
bytes.extend_from_slice(payload);
bytes.extend_from_slice(&[0xff, 0xd9]);
bytes
}
let healthy_payload: Vec<u8> = (0..8_000).map(|idx| (idx % 251) as u8).collect();
let frame = jpeg_with_sof(1920, 1080, &healthy_payload);
let inspection = super::inspect_mjpeg_frame(&frame);
assert_eq!(inspection.width, Some(1920));
assert_eq!(inspection.height, Some(1080));
assert!(inspection.entropy_distinct_bytes > 64);
temp_env::with_var(
"LESAVKA_UVC_DIRECT_MJPEG_REJECT_PROFILE_MISMATCH",
Some("1"),
|| {
assert_eq!(
super::direct_mjpeg_reject_reason(
0,
Some(frame.len() + 1),
Some((1280, 720)),
&frame,
),
Some(super::DirectMjpegRejectReason::ProfileMismatch {
expected_width: 1280,
expected_height: 720,
actual_width: 1920,
actual_height: 1080,
})
);
},
);
assert_eq!(
super::direct_mjpeg_reject_reason(0, Some(frame.len() - 1), Some((1920, 1080)), &frame),
Some(super::DirectMjpegRejectReason::Oversized {
max_bytes: frame.len() - 1,
})
);
}
#[test]
fn direct_mjpeg_visual_guard_can_be_disabled_without_allowing_truncation() {
let mut complete = vec![0xff, 0xd8, 0xff, 0xda];

110
server/src/video_sinks/mjpeg_spool.rs
View File

@ -9,6 +9,8 @@ use gstreamer_app as gst_app;
static SPOOL_SEQUENCE: AtomicU64 = AtomicU64::new(1);
static SPOOL_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
const MAX_MJPEG_FRAME_BYTES: usize = 8 * 1024 * 1024;
const DEFAULT_UVC_MJPEG_BUDGET_BYTES_PER_SEC: u32 = 10_000_000;
#[derive(Clone, Copy)]
pub(super) struct MjpegSpoolTiming {
@ -75,6 +77,61 @@ pub(super) fn mjpeg_spool_path() -> PathBuf {
.unwrap_or_else(|_| PathBuf::from("/run/lesavka-uvc-frame.mjpg"))
}
fn env_u32_opt(name: &str) -> Option<u32> {
std::env::var(name)
.ok()
.and_then(|value| value.trim().parse::<u32>().ok())
}
fn env_flag_enabled(name: &str, default: bool) -> bool {
std::env::var(name)
.ok()
.map(|value| {
let trimmed = value.trim();
if trimmed.eq_ignore_ascii_case("0")
|| trimmed.eq_ignore_ascii_case("false")
|| trimmed.eq_ignore_ascii_case("no")
|| trimmed.eq_ignore_ascii_case("off")
{
false
} else if trimmed.eq_ignore_ascii_case("1")
|| trimmed.eq_ignore_ascii_case("true")
|| trimmed.eq_ignore_ascii_case("yes")
|| trimmed.eq_ignore_ascii_case("on")
{
true
} else {
default
}
})
.unwrap_or(default)
}
/// Resolve the MJPEG byte budget used before publishing to the helper.
///
/// Inputs: active FPS plus `LESAVKA_UVC_FRAME_MAX_BYTES`,
/// `LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC`, and
/// `LESAVKA_UVC_FRAME_SIZE_GUARD`. Output: maximum accepted frame bytes.
/// Why: oversized MJPEG frames are a common source of host-visible UVC tearing;
/// a short freeze is better than letting the USB gadget emit partial pictures.
pub(super) fn mjpeg_spool_frame_max_bytes(fps: u32) -> usize {
if !env_flag_enabled("LESAVKA_UVC_FRAME_SIZE_GUARD", true) {
return MAX_MJPEG_FRAME_BYTES;
}
if let Some(limit) = env_u32_opt("LESAVKA_UVC_FRAME_MAX_BYTES")
&& limit > 0
{
return (limit as usize).min(MAX_MJPEG_FRAME_BYTES);
}
let fps = fps.max(1);
let budget_per_sec = env_u32_opt("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC")
.unwrap_or(DEFAULT_UVC_MJPEG_BUDGET_BYTES_PER_SEC)
.max(1);
let per_frame = (budget_per_sec / fps).max(64 * 1024);
per_frame.min(MAX_MJPEG_FRAME_BYTES as u32) as usize
}
/// Decide whether frame spool metadata should be published.
///
/// Inputs: `LESAVKA_UVC_FRAME_META`. Output: false unless explicitly enabled.
@ -288,6 +345,59 @@ mod tests {
});
}
#[test]
fn mjpeg_spool_frame_budget_uses_live_budget_when_zero_or_unset() {
temp_env::with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", None::<&str>),
("LESAVKA_UVC_FRAME_MAX_BYTES", None::<&str>),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", None::<&str>),
],
|| {
assert_eq!(super::mjpeg_spool_frame_max_bytes(30), 333_333);
},
);
temp_env::with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", Some("1")),
("LESAVKA_UVC_FRAME_MAX_BYTES", Some("0")),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", Some("9")),
],
|| {
assert_eq!(super::mjpeg_spool_frame_max_bytes(30), 65_536);
},
);
}
#[test]
fn mjpeg_spool_frame_budget_allows_explicit_limit_or_diagnostic_disable() {
temp_env::with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", Some("1")),
("LESAVKA_UVC_FRAME_MAX_BYTES", Some("123456")),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", Some("1")),
],
|| {
assert_eq!(super::mjpeg_spool_frame_max_bytes(30), 123_456);
},
);
temp_env::with_vars(
[
("LESAVKA_UVC_FRAME_SIZE_GUARD", Some("0")),
("LESAVKA_UVC_FRAME_MAX_BYTES", Some("123456")),
("LESAVKA_UVC_MJPEG_BUDGET_BYTES_PER_SEC", Some("1")),
],
|| {
assert_eq!(
super::mjpeg_spool_frame_max_bytes(30),
super::MAX_MJPEG_FRAME_BYTES
);
},
);
}
/// Verifies spool metadata remains opt-in and path-configurable.
///
/// Input: default and explicit metadata env vars. Output: disabled by

48
server/src/video_sinks/webcam_sink.rs
View File

@ -23,7 +23,10 @@ mod mjpeg_spool;
#[cfg(not(coverage))]
use gst::MessageView::{Error, StateChanged, Warning};
#[cfg(not(coverage))]
use mjpeg_spool::{freshest_mjpeg_sample, spool_mjpeg_frame_with_timing, MjpegSpoolTiming};
use mjpeg_spool::{
MjpegSpoolTiming, freshest_mjpeg_sample, mjpeg_spool_frame_max_bytes,
spool_mjpeg_frame_with_timing,
};
use mjpeg_spool::{mjpeg_spool_enabled, mjpeg_spool_path};
/// Push H.264 or MJPEG frames into the USB UVC gadget.
@ -42,6 +45,10 @@ pub struct WebcamSink {
hevc_mjpeg_appsrc: Option<gst_app::AppSrc>,
decoded_mjpeg_sink: Option<gst_app::AppSink>,
last_mjpeg_passthrough_bytes: AtomicU64,
direct_mjpeg_max_bytes: usize,
uvc_width: u16,
uvc_height: u16,
direct_mjpeg_profile_mismatch_seen: AtomicBool,
last_decoded_mjpeg_bytes: AtomicU64,
decoded_mjpeg_miss_count: AtomicU64,
decode_recovery_needs_irap: AtomicBool,
@ -377,6 +384,10 @@ impl WebcamSink {
hevc_mjpeg_appsrc: None,
decoded_mjpeg_sink: None,
last_mjpeg_passthrough_bytes: AtomicU64::new(0),
direct_mjpeg_max_bytes: mjpeg_spool::mjpeg_spool_frame_max_bytes(cfg.fps),
uvc_width: cfg.width.min(u32::from(u16::MAX)) as u16,
uvc_height: cfg.height.min(u32::from(u16::MAX)) as u16,
direct_mjpeg_profile_mismatch_seen: AtomicBool::new(false),
last_decoded_mjpeg_bytes: AtomicU64::new(0),
decoded_mjpeg_miss_count: AtomicU64::new(0),
decode_recovery_needs_irap: AtomicBool::new(false),
@ -650,6 +661,10 @@ impl WebcamSink {
hevc_mjpeg_appsrc,
decoded_mjpeg_sink,
last_mjpeg_passthrough_bytes: AtomicU64::new(0),
direct_mjpeg_max_bytes: mjpeg_spool_frame_max_bytes(cfg.fps),
uvc_width: cfg.width.min(u32::from(u16::MAX)) as u16,
uvc_height: cfg.height.min(u32::from(u16::MAX)) as u16,
direct_mjpeg_profile_mismatch_seen: AtomicBool::new(false),
last_decoded_mjpeg_bytes: AtomicU64::new(0),
decoded_mjpeg_miss_count: AtomicU64::new(0),
decode_recovery_needs_irap: AtomicBool::new(false),
@ -787,11 +802,40 @@ impl WebcamSink {
let previous_bytes = self
.last_mjpeg_passthrough_bytes
.load(std::sync::atomic::Ordering::Relaxed);
if hevc_mjpeg_guard::should_reject_direct_mjpeg_frame(previous_bytes, &pkt.data) {
let inspection = hevc_mjpeg_guard::inspect_mjpeg_frame(&pkt.data);
if let (Some(width), Some(height)) = (inspection.width, inspection.height)
&& (width, height) != (self.uvc_width, self.uvc_height)
&& !self
.direct_mjpeg_profile_mismatch_seen
.swap(true, std::sync::atomic::Ordering::Relaxed)
{
warn!(
target:"lesavka_server::video",
frame_width = width,
frame_height = height,
uvc_width = self.uvc_width,
uvc_height = self.uvc_height,
"📸⚠️ direct MJPEG frame dimensions differ from the live UVC profile; this can make browser output unstable"
);
}
if let Some(reason) = hevc_mjpeg_guard::direct_mjpeg_reject_reason(
previous_bytes,
Some(self.direct_mjpeg_max_bytes),
Some((self.uvc_width, self.uvc_height)),
&pkt.data,
) {
warn!(
target:"lesavka_server::video",
?reason,
previous_bytes,
next_bytes = pkt.data.len(),
max_bytes = self.direct_mjpeg_max_bytes,
frame_width = ?inspection.width,
frame_height = ?inspection.height,
entropy_bytes = inspection.entropy_bytes,
entropy_distinct_bytes = inspection.entropy_distinct_bytes,
entropy_dominant_pct = inspection.entropy_dominant_pct,
entropy_max_run = inspection.entropy_max_run,
"📸⚠️ freezing suspicious direct MJPEG frame before UVC spool"
);
return;

15
tests/chaos/server/video_sinks/hevc_mjpeg_guard_chaos_contract.rs
View File

@ -33,6 +33,14 @@ mod guard {
pub fn should_reject_direct_frame(previous_bytes: u64, mjpeg: &[u8]) -> bool {
should_reject_direct_mjpeg_frame(previous_bytes, mjpeg)
}
pub fn should_reject_direct_frame_with_budget(
previous_bytes: u64,
max_bytes: usize,
mjpeg: &[u8],
) -> bool {
direct_mjpeg_reject_reason(previous_bytes, Some(max_bytes), None, mjpeg).is_some()
}
}
const WEBCAM_SINK: &str = include_str!(concat!(
@ -115,7 +123,7 @@ fn server_hevc_recovery_and_freshest_spool_paths_remain_wired() {
"last_decoded_mjpeg_bytes",
"last_mjpeg_passthrough_bytes",
"should_freeze_decoded_mjpeg_frame(previous_bytes, map.as_slice())",
"should_reject_direct_mjpeg_frame(previous_bytes, &pkt.data)",
"direct_mjpeg_reject_reason(",
"spool_direct_mjpeg_frame",
"freezing suspicious decoded HEVC->MJPEG frame",
"freezing suspicious direct MJPEG frame before UVC spool",
@ -159,6 +167,11 @@ fn direct_mjpeg_guard_is_conservative_but_filters_obvious_black_or_truncated_fra
|| {
assert!(!guard::should_reject_direct_frame(0, &flat));
assert!(!guard::should_reject_direct_frame(220_000, &healthy));
assert!(guard::should_reject_direct_frame_with_budget(
220_000,
healthy.len() - 1,
&healthy
));
assert!(guard::should_reject_direct_frame(220_000, &flat));
assert!(guard::should_reject_direct_frame(220_000, &tiny));
assert!(guard::should_reject_direct_frame(220_000, &truncated));

2
tests/installer/scripts/install/server_install_script_contract.rs
View File

@ -186,6 +186,8 @@ fn server_install_pins_hdmi_camera_and_display_defaults() {
assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_WIDTH 1280"));
assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_HEIGHT 720"));
assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_CONTROL_READ_ONLY 0"));
assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_FRAME_SIZE_GUARD 1"));
assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_FRAME_MAX_BYTES 0"));
assert!(
!SERVER_INSTALL.contains("LESAVKA_UVC_CODEC=${LESAVKA_UVC_CODEC:-mjpeg}"),
"install script should not let ambient LESAVKA_UVC_CODEC leak into persisted defaults"