From 2a5f059a0b13a125265673e8d0ffa743402dd487 Mon Sep 17 00:00:00 2001 From: Brad Stein Date: Thu, 13 Aug 2026 02:57:02 -0300 Subject: [PATCH] fix(uvc): preserve control event fairness --- Cargo.lock | 6 ++-- client/Cargo.toml | 2 +- common/Cargo.toml | 2 +- docs/architecture-and-media-readiness.md | 24 ++++++++++++--- scripts/install/server.sh | 1 + scripts/manual/run_synthetic_rct_uvc_probe.py | 28 +++++++++++++---- server/Cargo.toml | 2 +- server/src/bin/lesavka-uvc.real.inc | 22 +++++++++++--- .../install/server_install_script_contract.rs | 1 + .../server/uvc/server_uvc_binary_contract.rs | 20 +++++++++++++ ...synthetic_rct_uvc_probe_manual_contract.rs | 30 +++++++++++++++++++ 11 files changed, 119 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0a38005..665e919 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1658,7 +1658,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "lesavka_client" -version = "0.27.9" +version = "0.27.10" dependencies = [ "anyhow", "async-stream", @@ -1692,7 +1692,7 @@ dependencies = [ [[package]] name = "lesavka_common" -version = "0.27.9" +version = "0.27.10" dependencies = [ "anyhow", "base64", @@ -1704,7 +1704,7 @@ dependencies = [ [[package]] name = "lesavka_server" -version = "0.27.9" +version = "0.27.10" dependencies = [ "anyhow", "base64", diff --git a/client/Cargo.toml b/client/Cargo.toml index b0b6f17..c06a01e 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -4,7 +4,7 @@ path = "src/main.rs" [package] name = "lesavka_client" -version = "0.27.9" +version = "0.27.10" edition = "2024" [dependencies] diff --git a/common/Cargo.toml b/common/Cargo.toml index 88bfbf9..1f55ed7 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lesavka_common" -version = "0.27.9" +version = "0.27.10" edition = "2024" build = "build.rs" diff --git a/docs/architecture-and-media-readiness.md b/docs/architecture-and-media-readiness.md index 6ce1a9f..ed76268 100644 --- a/docs/architecture-and-media-readiness.md +++ b/docs/architecture-and-media-readiness.md @@ -314,7 +314,19 @@ into the Lesavka tree, installs it with the server, requires `libgpiod`, and runs it with `/usr/bin/python` before capture discovery. This removes the unversioned virtualenv from the physical eye-feed power path. -## 7. The Current Downstream Video Failure +### The `0.27.10` UVC Event-Fairness Fix + +The first post-repair synthetic upstream probe reached Tethys but produced no +decodable MJPEG timeline. Tethys recorded repeated `EOI before SOF` packets and +two five-second UVC probe-control timeouts. The helper drained completed V4L2 +output buffers without a per-turn bound before servicing control events, so a +continuously consuming host could starve probe/commit traffic. Release 0.27.10 +bounds dequeue work, reports fairness yields, and makes the native receiver +probe terminate with evidence even when malformed packets produce no decoded +timestamps. This is a hardware-testable correction; marked Tethys capture is +still required before declaring the tear/smear class fixed. + +## 7. The Resolved Downstream Video Failure The current blank downstream feeds fail before transport or decoding. @@ -369,6 +381,10 @@ successfully imported the installed `libgpiod` binding. This proves the common downstream failure was capture-card power automation, not two simultaneous card failures, H.264 transport, or client decoding. Release 0.27.9 makes the working system binding and a versioned helper the installed service contract. +After installing 0.27.9, GPIO27 stayed asserted, both GC311 devices enumerated, +and both downstream RPCs decoded 1280x720 H.264. The right eye carried changing +desktop video; the left eye carried the capture card's explicit `No Signal` +frame, isolating that remaining input condition to its HDMI source/cable. ## 8. Recovery And Deployment Sequence @@ -386,7 +402,7 @@ The safe completion sequence for this incident is: 6. Run `scripts/install/server.sh` as the trusted deployment path. Preserve the already-attached USB gadget unless a controlled rebuild is explicitly required. -7. Confirm Theia reports server version `0.27.9`, the pushed release revision, +7. Confirm Theia reports server version `0.27.10`, the pushed release revision, direct MJPEG normalizer timeout `0`, and a coherent UVC contract. 8. Open both downstream RPCs and prove that each emits changing, decodable H.264 frames. @@ -404,7 +420,7 @@ hardware contract is repeatable. The remaining work falls into five groups. ### A. Install And Version Parity -- Push and deploy `0.27.9` through the client/server install scripts. +- Push and deploy `0.27.10` through the client/server install scripts. - Confirm client/server version and revision in every hardware probe artifact. - Eliminate the current state where a fixed client talks to an unfixed server. @@ -470,6 +486,6 @@ host repair: 9. disconnect/reconnect and device changes recover without stale media; and 10. diagnostics identify the failed physical stage when any item breaks. -Until that sequence passes on the installed `0.27.9` client/server pair, the +Until that sequence passes on the installed `0.27.10` client/server pair, the current release should be described as a validated code correction awaiting hardware deployment and end-to-end acceptance, not as a completed product fix. diff --git a/scripts/install/server.sh b/scripts/install/server.sh index 7791ed4..b9c4dac 100755 --- a/scripts/install/server.sh +++ b/scripts/install/server.sh @@ -418,6 +418,7 @@ LESAVKA_UVC_WIDTH=$(uvc_env_value LESAVKA_UVC_WIDTH 1280) LESAVKA_UVC_HEIGHT=$(uvc_env_value LESAVKA_UVC_HEIGHT 720) LESAVKA_UVC_CODEC=${INSTALL_UVC_CODEC} LESAVKA_UVC_BUFFER_COUNT=$(uvc_env_value LESAVKA_UVC_BUFFER_COUNT 4) +LESAVKA_UVC_PUMP_DEQUEUE_BUDGET=$(uvc_env_value LESAVKA_UVC_PUMP_DEQUEUE_BUDGET 4) 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_QUEUE_PACING=$(uvc_env_value LESAVKA_UVC_QUEUE_PACING 0) diff --git a/scripts/manual/run_synthetic_rct_uvc_probe.py b/scripts/manual/run_synthetic_rct_uvc_probe.py index 6e942a0..7221f12 100755 --- a/scripts/manual/run_synthetic_rct_uvc_probe.py +++ b/scripts/manual/run_synthetic_rct_uvc_probe.py @@ -1672,8 +1672,8 @@ def run_capture(args: argparse.Namespace) -> int: command, capture_width, capture_height, device = ffmpeg_cmd(args, width, height) artifact_dir = pathlib.Path(args.artifact_dir) if args.artifact_dir else pathlib.Path("/tmp") / f"lesavka-synthetic-rct-capture-{timestamp()}" artifact_dir.mkdir(parents=True, exist_ok=True) - if args.deep_capture and args.source != "uvc": - raise SystemExit("--deep-capture requires --source uvc") + if args.deep_capture and args.source != "device": + raise SystemExit("--deep-capture requires --source device") if args.deep_capture: write_deep_capture_snapshot(artifact_dir, "before", device) dynamic_debug_enabled = ( @@ -1688,6 +1688,7 @@ def run_capture(args: argparse.Namespace) -> int: analysis_elapsed = 0.0 raw_capture_bytes = 0 ffmpeg_rc: int | None = None + capture_timed_out = False frame_index = 0 suspicious_count = 0 visual_suspicious_count = 0 @@ -1807,11 +1808,26 @@ def run_capture(args: argparse.Namespace) -> int: capture_command[-1:-1] = ["-t", str(args.duration)] capture_command[-1] = str(raw_path) (artifact_dir / "command.txt").write_text(" ".join(shlex.quote(part) for part in capture_command) + "\n") - print(f"capturing raw RCT frames before analysis: {raw_path}", file=sys.stderr) + capture_target = native_mjpeg if native_mjpeg is not None else raw_path + print(f"capturing RCT frames before analysis: {capture_target}", file=sys.stderr) capture_started = time.monotonic() - proc = subprocess.run(capture_command, stdout=subprocess.DEVNULL, stderr=err, check=False) + try: + proc = subprocess.run( + capture_command, + stdout=subprocess.DEVNULL, + stderr=err, + check=False, + timeout=max(5.0, args.duration + 10.0), + ) + ffmpeg_rc = proc.returncode + except subprocess.TimeoutExpired: + capture_timed_out = True + ffmpeg_rc = 124 + err.write( + f"capture exceeded wall timeout after {args.duration + 10.0:.1f}s\n".encode() + ) + err.flush() capture_elapsed = time.monotonic() - capture_started - ffmpeg_rc = proc.returncode if args.deep_capture and native_mjpeg is not None and native_mjpeg.exists(): decode_command = [ "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning", @@ -1819,6 +1835,7 @@ def run_capture(args: argparse.Namespace) -> int: ] (artifact_dir / "decode-command.txt").write_text(" ".join(shlex.quote(part) for part in decode_command) + "\n") subprocess.run(decode_command, stdout=subprocess.DEVNULL, stderr=err, check=False) + raw_path.touch(exist_ok=True) raw_capture_bytes = raw_path.stat().st_size if raw_path.exists() else 0 print( f"analyzing captured raw RCT frames bytes={raw_capture_bytes} capture_s={capture_elapsed:.3f}", @@ -1855,6 +1872,7 @@ def run_capture(args: argparse.Namespace) -> int: "duration_observed_s": round(elapsed, 3), "analysis_duration_s": round(analysis_elapsed, 3), "ffmpeg_rc": ffmpeg_rc, + "capture_timed_out": capture_timed_out, "raw_capture_bytes": raw_capture_bytes, "frames": frame_index, "fps_observed": round(frame_index / elapsed, 3), diff --git a/server/Cargo.toml b/server/Cargo.toml index 15a0781..bc4edff 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -16,7 +16,7 @@ bench = false [package] name = "lesavka_server" -version = "0.27.9" +version = "0.27.10" edition = "2024" autobins = false diff --git a/server/src/bin/lesavka-uvc.real.inc b/server/src/bin/lesavka-uvc.real.inc index 793a695..01cfc99 100644 --- a/server/src/bin/lesavka-uvc.real.inc +++ b/server/src/bin/lesavka-uvc.real.inc @@ -53,6 +53,7 @@ const IDLE_MJPEG_FRAME: &[u8] = include_bytes!("lesavka_uvc/idle_1280x720_black. // Keep enough output buffers queued that the USB gadget is not starved while a // browser is consuming high-speed isochronous MJPEG. const DEFAULT_UVC_BUFFER_COUNT: u32 = 4; +const DEFAULT_UVC_PUMP_DEQUEUE_BUDGET: usize = 4; const DEFAULT_UVC_IDLE_PUMP_MS: u64 = 2; const DEFAULT_UVC_FRAME_MAX_AGE_MS: u64 = 1_000; const DEFAULT_UVC_QUEUE_PACING: bool = false; @@ -292,6 +293,7 @@ struct UvcVideoStats { last_rejected_oversize_cap: usize, paced_sleeps: u64, paced_sleep_ms: u64, + pump_budget_yields: u64, last_report: Option, } @@ -380,7 +382,7 @@ impl UvcVideoStream { if !self.streaming { return Ok(()); } - loop { + for _ in 0..uvc_pump_dequeue_budget() { let mut buf = empty_v4l2_buffer(); let req = ioctl_readwrite::(b'V', 17); let rc = unsafe { libc::ioctl(self.fd, req, &mut buf) }; @@ -399,6 +401,8 @@ impl UvcVideoStream { } self.queue_buffer(buf.index, buf.flags)?; } + self.stats.pump_budget_yields += 1; + Ok(()) } fn set_format(&self, cfg: UvcConfig) -> Result<()> { @@ -684,7 +688,7 @@ impl UvcVideoStream { } self.stats.last_report = Some(now); eprintln!( - "[lesavka-uvc] video stats queued={} reloaded={} stale_replay={} rejected_oversize={} rejected_invalid={} fallback_idle={} held_last_good={} read_errors={} strict_validation_failures={} dqbuf_ioctl_errors={} dqbuf_flag_errors={} qbuf_ioctl_errors={} latest_bytes={} frame_cap={} last_rejected_oversize_bytes={} last_rejected_oversize_cap={} paced_sleeps={} paced_sleep_ms={}", + "[lesavka-uvc] video stats queued={} reloaded={} stale_replay={} rejected_oversize={} rejected_invalid={} fallback_idle={} held_last_good={} read_errors={} strict_validation_failures={} dqbuf_ioctl_errors={} dqbuf_flag_errors={} qbuf_ioctl_errors={} latest_bytes={} frame_cap={} last_rejected_oversize_bytes={} last_rejected_oversize_cap={} paced_sleeps={} paced_sleep_ms={} pump_budget_yields={}", self.stats.queued, self.stats.reloaded, self.stats.replayed_stale, @@ -702,7 +706,8 @@ impl UvcVideoStream { self.stats.last_rejected_oversize_bytes, self.stats.last_rejected_oversize_cap, self.stats.paced_sleeps, - self.stats.paced_sleep_ms + self.stats.paced_sleep_ms, + self.stats.pump_budget_yields ); if let Some(path) = uvc_stats_path() { let _ = write_atomic_text( @@ -778,6 +783,14 @@ fn uvc_buffer_count() -> u32 { env_u32("LESAVKA_UVC_BUFFER_COUNT", DEFAULT_UVC_BUFFER_COUNT).clamp(1, 8) } +fn uvc_pump_dequeue_budget() -> usize { + env_u64( + "LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", + DEFAULT_UVC_PUMP_DEQUEUE_BUDGET as u64, + ) + .clamp(1, 64) as usize +} + fn uvc_idle_pump_sleep() -> Duration { Duration::from_millis(env_u64( "LESAVKA_UVC_IDLE_PUMP_MS", @@ -903,7 +916,7 @@ fn write_atomic_text(path: &std::path::Path, text: &str) -> Result<()> { fn uvc_stats_snapshot_json(stats: &UvcVideoStats, frame_cap: usize) -> String { let kernel = uvc_kernel_stats_json(); format!( - "{{\"generated_unix_ms\":{},\"queued\":{},\"reloaded\":{},\"stale_replay\":{},\"rejected_oversize\":{},\"rejected_invalid\":{},\"fallback_idle\":{},\"held_last_good\":{},\"read_errors\":{},\"strict_validation_failures\":{},\"dqbuf_ioctl_errors\":{},\"dqbuf_flag_errors\":{},\"qbuf_ioctl_errors\":{},\"latest_bytes\":{},\"frame_cap\":{},\"last_rejected_oversize_bytes\":{},\"last_rejected_oversize_cap\":{},\"paced_sleeps\":{},\"paced_sleep_ms\":{},\"kernel\":{}}}\n", + "{{\"generated_unix_ms\":{},\"queued\":{},\"reloaded\":{},\"stale_replay\":{},\"rejected_oversize\":{},\"rejected_invalid\":{},\"fallback_idle\":{},\"held_last_good\":{},\"read_errors\":{},\"strict_validation_failures\":{},\"dqbuf_ioctl_errors\":{},\"dqbuf_flag_errors\":{},\"qbuf_ioctl_errors\":{},\"latest_bytes\":{},\"frame_cap\":{},\"last_rejected_oversize_bytes\":{},\"last_rejected_oversize_cap\":{},\"paced_sleeps\":{},\"paced_sleep_ms\":{},\"pump_budget_yields\":{},\"kernel\":{}}}\n", unix_now_ns() / 1_000_000, stats.queued, stats.reloaded, @@ -923,6 +936,7 @@ fn uvc_stats_snapshot_json(stats: &UvcVideoStats, frame_cap: usize) -> String { stats.last_rejected_oversize_cap, stats.paced_sleeps, stats.paced_sleep_ms, + stats.pump_budget_yields, kernel ) } diff --git a/tests/contract/scripts/install/server_install_script_contract.rs b/tests/contract/scripts/install/server_install_script_contract.rs index d792acb..01633f5 100644 --- a/tests/contract/scripts/install/server_install_script_contract.rs +++ b/tests/contract/scripts/install/server_install_script_contract.rs @@ -276,6 +276,7 @@ fn server_install_pins_hdmi_camera_and_display_defaults() { 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_BUFFER_COUNT 4")); + assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_PUMP_DEQUEUE_BUDGET 4")); assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_QUEUE_PACING 0")); assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_BULK 1")); assert!(SERVER_INSTALL.contains("uvc_env_value LESAVKA_UVC_FRAME_SIZE_GUARD 1")); diff --git a/tests/contract/server/uvc/server_uvc_binary_contract.rs b/tests/contract/server/uvc/server_uvc_binary_contract.rs index 03e1318..2415466 100644 --- a/tests/contract/server/uvc/server_uvc_binary_contract.rs +++ b/tests/contract/server/uvc/server_uvc_binary_contract.rs @@ -376,6 +376,24 @@ mod uvc_binary { assert!(source.contains("streaming_bulk is not 1; using isochronous payload sizing")); } + #[test] + #[serial] + fn uvc_video_pump_is_bounded_so_control_requests_cannot_starve() { + with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", None::<&str>, || { + assert_eq!(uvc_pump_dequeue_budget(), 4); + }); + with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", Some("1"), || { + assert_eq!(uvc_pump_dequeue_budget(), 1); + }); + with_var("LESAVKA_UVC_PUMP_DEQUEUE_BUDGET", Some("999"), || { + assert_eq!(uvc_pump_dequeue_budget(), 64); + }); + + let source = include_str!("../../../../server/src/bin/lesavka-uvc.real.inc"); + assert!(source.contains("for _ in 0..uvc_pump_dequeue_budget()")); + assert!(source.contains("self.stats.pump_budget_yields += 1")); + } + #[test] #[serial] fn uvc_stats_snapshot_can_be_disabled_or_written() { @@ -397,6 +415,7 @@ mod uvc_binary { last_rejected_oversize_cap: 300_000, paced_sleeps: 5, paced_sleep_ms: 123, + pump_budget_yields: 11, last_report: None, }; let json = uvc_stats_snapshot_json(&stats, 333_333); @@ -407,6 +426,7 @@ mod uvc_binary { assert!(json.contains("\"last_rejected_oversize_cap\":300000")); assert!(json.contains("\"paced_sleeps\":5")); assert!(json.contains("\"paced_sleep_ms\":123")); + assert!(json.contains("\"pump_budget_yields\":11")); let kernel_path = tempfile::NamedTempFile::new().expect("kernel stats"); fs::write( diff --git a/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs b/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs index 9df1bec..671a01d 100644 --- a/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs +++ b/tests/manual/server/rct/synthetic_rct_uvc_probe_manual_contract.rs @@ -152,6 +152,36 @@ fn synthetic_probe_keeps_bundled_network_ingress_and_rct_comparison_markers() { } } +#[test] +fn synthetic_probe_deep_capture_accepts_its_device_source_mode() { + assert!( + PROBE_SRC.contains("if args.deep_capture and args.source != \"device\":") + && PROBE_SRC.contains("--deep-capture requires --source device"), + "deep capture must accept the parser's UVC device source instead of requiring an unavailable source name" + ); + assert!( + !PROBE_SRC.contains("--deep-capture requires --source uvc"), + "the probe must not demand a source name its argument parser rejects" + ); +} + +#[test] +fn synthetic_probe_capture_has_a_wall_clock_timeout() { + for expected in [ + "timeout=max(5.0, args.duration + 10.0)", + "except subprocess.TimeoutExpired:", + "capture_timed_out = True", + "ffmpeg_rc = 124", + "raw_path.touch(exist_ok=True)", + "\"capture_timed_out\": capture_timed_out", + ] { + assert!( + PROBE_SRC.contains(expected), + "receiver capture must terminate and report evidence even when malformed UVC packets produce no decoded timeline: {expected}" + ); + } +} + #[test] fn synthetic_injector_enters_the_public_bundled_media_rpc() { for expected in [