diff --git a/docs/LIVE_TAP.md b/docs/LIVE_TAP.md index 8033a34..98b4759 100644 --- a/docs/LIVE_TAP.md +++ b/docs/LIVE_TAP.md @@ -38,13 +38,18 @@ or `qorcap record ... --live-tap udp://127.0.0.1:1937 --practice-lane [--live-ta Refused (logged, tap stays off) when the lane is not practice, when the target is not `udp://` or not a loopback address, or when it has no port. `/status` gains a `live_tap` object (frames, cues, datagrams, -send errors) only while the tap is on. +send errors, queue drops) only while the tap is on. ## What it does not change - The recording, `manifest.jsonl`, the checkpoint chain, `receipt.draft.json`, `receipt.v0` and `contracts/` are untouched. `verify-bundle` passes with the tap on (box end-to-end test). - Sending is non-blocking UDP on loopback. If nothing listens, datagrams are dropped and counted. +- The capture thread never does tap work itself (since the v0.3.43 stall fix). It copies the recorded + slice to a small queue (8 slices) and a tap worker sends the UDP and writes `live_weave.jsonl`. If the + worker falls behind, tap slices are dropped and counted in `queue_drops`; the recording is not affected + and later cues keep the recording's frame index. The cue's controller/seal fields come from a snapshot + the input log keeps, so the capture thread never waits on the input-log mutex. - Nothing is sent to the controller. Nothing leaves the machine. - REAL MATCH builds are not changed by this file; the tap is not in v0.3.38. diff --git a/src/engine.rs b/src/engine.rs index d3a2b4a..3c5744e 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -87,6 +87,9 @@ pub struct Engine { pub frame_clock: Arc, /// Opt-in practice live tap (src/livetap.rs). Off: every call is a no-op. pub live_tap: Arc, + /// WeaveCtx snapshot for the live tap, kept current by the input log only while the tap is on. + /// The capture thread reads this short mutex, never `input`. + weave_snap: Arc>, /// The supervisor loop (`run`) is active; the API keeps serving until it has finished the bundle. pub running: AtomicBool, /// Set when `run` has returned (or /shutdown arrives while not running): the API thread exits. @@ -398,6 +401,7 @@ impl Engine { .or_else(|| args.out.as_ref().map(|o| PathBuf::from(format!("{o}.inputs.jsonl")))); let mut log = crate::inputs::InputLog::new(input_path, !args.no_controller, wall_ms()); if args.game.as_deref() == Some("cod_mw4") { log.enable_seal(); } + let weave_snap = Arc::new(Mutex::new(log.weave_ctx())); let input = Arc::new(Mutex::new(log)); // After the input logger's wall anchor, before `started`. Opening the file does not read the clock. let frame_on = crate::frameclock::flag_enabled(args.frame_clock, std::env::var(crate::frameclock::ENV).ok().as_deref()); @@ -410,6 +414,8 @@ impl Engine { let tap_every = args.live_tap_every.filter(|n| (1..=600).contains(n)) .unwrap_or_else(|| crate::livetap::every_from_env(std::env::var(crate::livetap::ENV_EVERY).ok().as_deref())); let live_tap = Arc::new(crate::livetap::LiveTap::open(tap_dest, tap_every, &input.lock().unwrap_or_else(|e| e.into_inner()).weave_ctx().wall_anchor_ms.to_string(), args.fps, args.bundle.as_deref().map(Path::new))); + // Only a tap that is on gets the snapshot; with the tap off the input log does no extra work. + if live_tap.enabled() { input.lock().unwrap_or_else(|e| e.into_inner()).set_weave_snap(weave_snap.clone()); } let st = Status { state: "starting".into(), device: args.device.clone(), res: format!("{}x{}", args.res.0, args.res.1), fps_target: args.fps, replay_window: args.replay.unwrap_or(0), role: args.role.clone(), side_source: args.side_source.clone(), ..Default::default() }; let started = Instant::now(); // One origin names the engine clock on both the snapshot and the input header, so the @@ -422,7 +428,7 @@ impl Engine { Arc::new(Self { ffmpeg, ring: Mutex::new(ReplayRing::new(window, cap)), rec: Mutex::new(None), status: Mutex::new(st), shutdown: AtomicBool::new(false), ev: Mutex::new(MatchState { started_wall_ms: wall_ms(), score: Value::Null, ..Default::default() }), subs: Mutex::new(vec![]), stamp_file: dir.join("stamp.txt"), - snapshot: dir.join("snapshot.jpg"), snap_pub: Mutex::new(snap), tap: crate::glimpse::FrameTap::with_ring(args.glimpse_frame || crate::glimpse::env_enabled(), crate::glimpse::ring_enabled()), exit_when_recording_ends: AtomicBool::new(false), started, last_data_ms: AtomicU64::new(0), input, frame_clock, live_tap, + snapshot: dir.join("snapshot.jpg"), snap_pub: Mutex::new(snap), tap: crate::glimpse::FrameTap::with_ring(args.glimpse_frame || crate::glimpse::env_enabled(), crate::glimpse::ring_enabled()), exit_when_recording_ends: AtomicBool::new(false), started, last_data_ms: AtomicU64::new(0), input, frame_clock, live_tap, weave_snap, running: AtomicBool::new(false), api_closed: AtomicBool::new(false), sealed: AtomicBool::new(false), countersigned: std::sync::atomic::AtomicI64::new(-1), stop_reason: Mutex::new(None), bundle_write: Mutex::new(()), bundle_finalized: AtomicBool::new(false), @@ -614,9 +620,11 @@ impl Engine { // Practice live tap: the same bytes that reached the recording, never anything else. if written && self.live_tap.enabled() { let file = Path::new(&r.path).file_name().map(|s| s.to_string_lossy().into_owned()).unwrap_or_default(); - let input = Arc::clone(&self.input); - self.live_tap.observe(slice, t_us, &file, pat.as_deref(), pmt.as_deref(), - &move || input.lock().unwrap_or_else(|e| e.into_inner()).weave_ctx()); + // Read the short snapshot only. Never take the input-log mutex on this thread: the HID + // logger and segment sealing hold it, and waiting on it stalled the capture pipe. + // `submit` hands the slice to the tap worker and never waits on UDP or disk. + let ctx = self.weave_snap.lock().unwrap_or_else(|e| e.into_inner()).clone(); + self.live_tap.submit(slice, t_us, &file, pat.as_deref(), pmt.as_deref(), ctx); } } if r.deadline.map(|d| Instant::now() >= d).unwrap_or(false) { @@ -1572,6 +1580,47 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// v0.3.43 stall: with the practice tap on, `ingest` (capture stdout thread) must not wait on the + /// input-log mutex, which the HID logger and segment sealing hold. + #[cfg(unix)] + #[test] + fn live_tap_ingest_does_not_wait_on_the_input_log() { + use clap::Parser; + let rx = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let url = format!("udp://{}", rx.local_addr().unwrap()); + let dir = std::env::temp_dir().join(format!("qorcap-livetap-eng-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let d = dir.to_string_lossy().into_owned(); + let crate::cli::Cmd::Record(r) = crate::cli::Cli::try_parse_from(["q", "record", "--device", "testpattern", "--bundle", &d, + "--live-tap", &url, "--practice-lane", "--live-tap-every", "1"]).unwrap().cmd else { panic!() }; + let e = Engine::new(r, "ffmpeg-x".into()); + assert!(e.live_tap.enabled()); + // A stand-in recorder that just swallows stdin, so `ingest` reaches the tap. + let mut child = std::process::Command::new("cat").stdin(Stdio::piped()).stdout(Stdio::null()).spawn().unwrap(); + let stdin = child.stdin.take(); + *e.rec.lock().unwrap() = Some(Recorder { reader: None, child, stdin, path: dir.join("video.mp4").to_string_lossy().into_owned(), + started: Instant::now(), deadline: None, wait_key: false, base: d.clone(), part: 1 }); + e.input.lock().unwrap().on_status(1, "connected", "DualSense"); + let held = e.input.lock().unwrap(); + let (tx, done) = std::sync::mpsc::channel(); + let e2 = e.clone(); + std::thread::spawn(move || { + for k in 0..30u64 { e2.ingest(&crate::frameclock::video_ts_packet(0x100, 900 + 1500 * k)); } + let _ = tx.send(()); + }); + done.recv_timeout(Duration::from_secs(3)).expect("ingest waited on the input-log mutex"); + drop(held); + let end = Instant::now() + Duration::from_secs(5); + while e.live_tap.stats().map(|s| s.frames + s.queue_drops).unwrap_or(0) < 30 && Instant::now() < end { std::thread::sleep(Duration::from_millis(10)); } + let s = e.live_tap.stats().unwrap(); + assert_eq!(s.frames + s.queue_drops, 30); + let first: Value = serde_json::from_str(std::fs::read_to_string(dir.join(crate::livetap::FILE_NAME)).unwrap().lines().nth(1).unwrap()).unwrap(); + assert_eq!(first["pad"], "connected", "cue reads the snapshot the input log published"); + if let Some(r) = e.rec.lock().unwrap().take() { finish(r); } + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn frame_clock_off_leaves_the_bundle_unchanged_and_on_records_only_when_asked() { use clap::Parser; @@ -1651,7 +1700,9 @@ impl Engine { crate::inputproof::touch_recording_marker(self.bundle_dir().as_deref()); self.broadcast("segment", &m); // Seal controller inputs received up to now into this segment's chunk (same clock, same checkpoint). - let (ih, ihead, n) = self.input.lock().unwrap_or_else(|e| e.into_inner()).seal(seq, self.started.elapsed().as_micros() as u64); + // The fsync runs outside the input-log mutex (HID logger and /status keep going); this still returns + // only once the chunk is durable, so the checkpoint below is written after it. + let (ih, ihead, n) = crate::inputs::seal_shared(&self.input, seq, || self.started.elapsed().as_micros() as u64); if self.args.no_controller { self.checkpoint("segment", json!({})); } else { self.checkpoint_with("segment", json!({"input_seg": seq, "input_n": n, "input_head": ihead}), Some(ih)); } } diff --git a/src/inputs.rs b/src/inputs.rs index fc59f65..5e613b8 100644 --- a/src/inputs.rs +++ b/src/inputs.rs @@ -220,6 +220,72 @@ pub struct InputLog { prev_counter: Option, prev_hw: Option, seal: crate::seal::SealWatch, + /// Short mutex, written by the HID/logger thread, read by the capture thread. Never the input-log lock. + weave_snap: Option>>, + /// Serializes `seal_shared` from prepare to finish, so chunks reach the file in chain order. + /// Never held together with a blocking wait on the input-log mutex by anyone else. + seal_io: std::sync::Arc>, + /// A chunk was prepared and its bytes are being written outside the input-log lock. + seal_in_flight: bool, +} + +/// One input chunk, fixed under the input-log lock and written to disk outside it. +/// `body` is exactly the bytes `seal` appends; nothing added to the log later can change it. +pub struct SealJob { + sha: String, + head: String, + n: usize, + recs: Vec, + body: String, + haptic: String, + path: Option, + haptic_path: Option, +} + +impl SealJob { + /// The blocking part: open, append, fsync. Takes no lock. Same calls and bytes as before. + pub fn write(&self) -> std::io::Result<()> { + if self.n == 0 { return Ok(()); } + if let Some(p) = &self.path { + if let Some(d) = p.parent() { let _ = std::fs::create_dir_all(d); } + if let Err(e) = append_bytes(p, self.body.as_bytes()) { + warn!("cannot write {}: {e}; input chunk not sealed (chain unchanged)", p.display()); + return Err(e); + } + } + if let Some(hp) = &self.haptic_path { + if !self.haptic.is_empty() { + if let Err(e) = append_bytes(hp, self.haptic.as_bytes()) { + warn!("cannot write {}: {e}", hp.display()); + } + } + } + Ok(()) + } +} + +/// Seal into chunk `seg` without holding the input-log mutex across disk I/O. +/// +/// 1. `seal_io` is taken first (never while holding `log`), so two seals cannot interleave. +/// 2. Under `log`: the pending records are taken and the chunk (bytes, sha256, prev, head) is fixed. +/// 3. `log` is released; the chunk is appended and fsynced (`append_bytes`, unchanged). +/// 4. Under `log` again: the chain moves only if step 3 succeeded; otherwise the records go back in +/// front of anything that arrived meanwhile, exactly where a blocked writer would have put them. +/// +/// Returns only after the fsync, so the caller's checkpoint is still written after the chunk is durable. +/// `now_us` is read under the input-log lock, right before the records are taken (as the old +/// `input.lock().seal(seg, now)` call did), so the chunk `t_us` keeps its meaning. +pub fn seal_shared(log: &std::sync::Mutex, seg: u64, now_us: impl FnOnce() -> u64) -> (String, String, usize) { + seal_shared_with(log, seg, now_us, |job| job.write()) +} + +/// `seal_shared` with the disk step injectable (tests). +pub fn seal_shared_with(log: &std::sync::Mutex, seg: u64, now_us: impl FnOnce() -> u64, write: impl FnOnce(&SealJob) -> std::io::Result<()>) -> (String, String, usize) { + let io = log.lock().unwrap_or_else(|e| e.into_inner()).seal_io.clone(); + let _serial = io.lock().unwrap_or_else(|e| e.into_inner()); + let job = { let mut g = log.lock().unwrap_or_else(|e| e.into_inner()); let t_us = now_us(); g.seal_prepare(seg, t_us) }; + let res = write(&job); + log.lock().unwrap_or_else(|e| e.into_inner()).seal_finish(job, res) } impl InputLog { @@ -232,7 +298,8 @@ impl InputLog { dev_counter: None, faults_since_hb: 0, max_rx_gap_us: 0, stale_over_50ms: 0, short_reads: 0, parse_faults: 0, first_rx_t_us: None, span: RxSpan::default(), prev_counter: None, prev_hw: None, - seal: crate::seal::SealWatch::new(false) }; + seal: crate::seal::SealWatch::new(false), weave_snap: None, + seal_io: std::sync::Arc::new(std::sync::Mutex::new(())), seal_in_flight: false }; if enabled { l.pending.push(json!({"type": "header", "schema": SCHEMA, "t_us": 0, "clock": "qorcap-mono-us (same Instant as manifest/checkpoint mono_ms)", "wall_anchor_ms": wall_anchor_ms, @@ -272,6 +339,7 @@ impl InputLog { self.seal.set_session(&self.wall_anchor_ms.to_string()); self.live.seal_line = self.seal.line.clone(); info!("{}", self.live.seal_line); + self.publish_weave(); } pub fn close_seal(&mut self) { @@ -282,6 +350,7 @@ impl InputLog { info!("{}", self.live.seal_line); self.persist_seal(); } + self.publish_weave(); } /// Read-only view for the practice live tap (src/livetap.rs). Nothing here touches the controller. @@ -295,6 +364,18 @@ impl InputLog { } } + /// Live tap only. The capture thread reads this snapshot, never `InputLog`'s own mutex. + pub fn set_weave_snap(&mut self, snap: std::sync::Arc>) { + self.weave_snap = Some(snap); + self.publish_weave(); + } + + fn publish_weave(&self) { + if let Some(s) = &self.weave_snap { + *s.lock().unwrap_or_else(|e| e.into_inner()) = self.weave_ctx(); + } + } + /// A finalized seal revision. Does not mark it bound. pub fn seal_binding(&self) -> Option { self.seal.binding() } @@ -314,6 +395,7 @@ impl InputLog { self.live.seal_line = self.seal.line.clone(); info!("{}", self.live.seal_line); self.persist_seal(); + self.publish_weave(); } } @@ -322,6 +404,7 @@ impl InputLog { self.live.seal_line = self.seal.line.clone(); info!("{}", self.live.seal_line); self.persist_seal(); + self.publish_weave(); } } @@ -409,6 +492,7 @@ impl InputLog { self.live.logged += 1; } self.poll_heartbeat(t_us); + self.publish_weave(); log } @@ -566,6 +650,7 @@ impl InputLog { } self.pending.push(rec); if state != "connected" { self.last_logged = None; } + self.publish_weave(); } /// Human-readable connect line. The structured `pad_id` is what the proof compares. @@ -592,7 +677,23 @@ impl InputLog { pub fn path(&self) -> Option<&Path> { self.path.as_deref() } /// Seal everything pending into chunk `seg`. Returns (chunk_sha256, chain_head, n_records). + /// Single-threaded form (tests, red team, offline tools). The engine uses `seal_shared`, which writes + /// the same bytes without holding the input-log mutex across the fsync. pub fn seal(&mut self, seg: u64, t_us: u64) -> (String, String, usize) { + let job = self.seal_prepare(seg, t_us); + let res = job.write(); + self.seal_finish(job, res) + } + + /// Fix the next chunk: take every pending record and compute its bytes and chain link. Moves nothing + /// on the chain; `seal_finish` does that once the bytes are durable. + pub fn seal_prepare(&mut self, seg: u64, t_us: u64) -> SealJob { + let empty = |head: String| SealJob { sha: empty_hash(), head, n: 0, recs: vec![], body: String::new(), haptic: String::new(), path: None, haptic_path: None }; + if self.seal_in_flight { + // Only reachable if a caller skips `seal_io`. Leave everything pending; the next seal takes it. + warn!("input seal for segment {seg} skipped: another chunk is still being written (chain unchanged)"); + return empty(self.head.clone()); + } if self.live.reports > 0 { let now = (self.detector.live.imu_present, self.detector.live.imu_rate_hz); let changed = match self.imu_committed { @@ -611,32 +712,34 @@ impl InputLog { } let sha = ev::jcs_sha(&Value::Array(recs.clone())); let n = recs.len(); - if n == 0 { return (sha, self.head.clone(), 0); } + if n == 0 { return SealJob { sha, ..empty(self.head.clone()) }; } let prev = self.head.clone(); let head = ev::link(&prev, self.chunks, &sha); let chunk = json!({"type": "chunk", "seq": self.chunks, "seg": seg, "n": n, "sha256": sha, "prev": prev, "head": head, "t_us": t_us}); let mut body = String::new(); for r in recs.iter().chain(std::iter::once(&chunk)) { body.push_str(&r.to_string()); body.push('\n'); } - if let Some(p) = &self.path { - if let Some(d) = p.parent() { let _ = std::fs::create_dir_all(d); } - if let Err(e) = append_bytes(p, body.as_bytes()) { - // The chunk is not on disk, so the chain must not move. The records stay pending and - // the next segment seals them. A checkpoint for this segment then commits the empty hash. - warn!("cannot write {}: {e}; input chunk not sealed (chain unchanged)", p.display()); - self.pending = recs; - return (empty_hash(), self.head.clone(), 0); - } - } - if let Some(hp) = &self.haptic_path { - let hs: String = recs.iter().filter(|r| r["type"] == "haptic").map(|r| format!("{r}\n")).collect(); - if !hs.is_empty() { - if let Err(e) = append_bytes(hp, hs.as_bytes()) { - warn!("cannot write {}: {e}", hp.display()); - } - } + let haptic: String = if self.haptic_path.is_some() { + recs.iter().filter(|r| r["type"] == "haptic").map(|r| format!("{r}\n")).collect() + } else { String::new() }; + self.seal_in_flight = true; + SealJob { sha, head, n, recs, body, haptic, path: self.path.clone(), haptic_path: self.haptic_path.clone() } + } + + /// Commit a prepared chunk after `SealJob::write`. On a failed write the chain does not move and the + /// records go back in front of any record that arrived during the write (same order as before). + pub fn seal_finish(&mut self, job: SealJob, written: std::io::Result<()>) -> (String, String, usize) { + if job.n == 0 { return (job.sha, job.head, 0); } + self.seal_in_flight = false; + if written.is_err() { + // The chunk is not on disk, so the chain must not move. The records stay pending and + // the next segment seals them. A checkpoint for this segment then commits the empty hash. + let mut recs = job.recs; + recs.append(&mut self.pending); + self.pending = recs; + return (empty_hash(), self.head.clone(), 0); } - self.head = head.clone(); self.chunks += 1; self.total_records += n as u64; - (sha, head, n) + self.head = job.head.clone(); self.chunks += 1; self.total_records += job.n as u64; + (job.sha, job.head, job.n) } } @@ -763,6 +866,129 @@ pub fn verify_chunks(recs: &[Value], chunks: &[Value]) -> (std::collections::Has mod tests { use super::*; fn pad(b: u32, l2: u8) -> Pad { Pad { buttons: b, l2, ..Pad::neutral() } } + /// Fixed input stream for the seal byte-equality checks. Only uses API that exists at v0.3.43. + fn seal_scenario_feed(l: &mut InputLog, from_us: u64, to_us: u64) { + let mut t = from_us; + let mut k = 0u32; + while t < to_us { + let b = if (k / 40) & 1 == 0 { 0 } else { hid::CROSS }; + let mut p = pad(b, ((k * 7) % 200) as u8); + p.counter = (k & 0xff) as u8; + p.sensor_ts = k.wrapping_mul(12_000); + l.on_report(t, &p); + t += 4_000; + k += 1; + } + } + + /// v0.3.43 `seal()` on `seal_scenario_feed`: inputs.jsonl sha256 / length and the last chunk's + /// (sha, head, n). Captured from the 36debdd tree before the seal was split. + const GOLDEN_FILE_SHA: &str = "e577d5af1762d2a21b4fb9102a64b9ce270f76dfa2ee346dbaa9a4859a5d7fc2"; + const GOLDEN_FILE_LEN: usize = 234_476; + const GOLDEN_LAST: (&str, &str, usize) = ("a28c108d1bab80f66a8acccbab3a0bb1ac13f577456e1118b50d1be4b56351f9", + "c75e33207c2f632b7f48570258c4ac92c93063fd9f20d63def2f2bcf86392518", 502); + + fn golden_run(tag: &str, shared: bool) -> (Vec, Vec<(String, String, usize)>) { + let dir = std::env::temp_dir().join(format!("qorcap-inputs-golden-{tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let p = dir.join("inputs.jsonl"); + let mut l = InputLog::new(Some(p.clone()), true, 1_700_000_000_000); + l.set_mono_origin_ns(42); + l.on_handle_open(10, "DualSense (USB) serial=T1", "serial:T1"); + let log = Mutex::new(l); + let mut outs = vec![]; + for seg in 0..4u64 { + seal_scenario_feed(&mut log.lock().unwrap(), seg * 2_000_000 + 20, (seg + 1) * 2_000_000); + let t = (seg + 1) * 2_000_000 + 5; + outs.push(if shared { seal_shared(&log, seg + 1, || t) } else { log.lock().unwrap().seal(seg + 1, t) }); + } + let bytes = std::fs::read(&p).unwrap(); + let _ = std::fs::remove_dir_all(&dir); + (bytes, outs) + } + + #[test] + fn seal_and_seal_shared_write_the_v0343_bytes() { + for shared in [false, true] { + let (bytes, outs) = golden_run(if shared { "shared" } else { "direct" }, shared); + assert_eq!((ev::sha256_hex(&bytes).as_str(), bytes.len()), (GOLDEN_FILE_SHA, GOLDEN_FILE_LEN), "shared={shared}"); + let last = outs.last().unwrap(); + assert_eq!((last.0.as_str(), last.1.as_str(), last.2), GOLDEN_LAST, "shared={shared}"); + } + } + + #[test] + fn seal_shared_fsyncs_without_the_input_lock_and_later_records_wait_for_the_next_chunk() { + let dir = std::env::temp_dir().join(format!("qorcap-inputs-shared-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let p = dir.join("inputs.jsonl"); + let log = Mutex::new(InputLog::new(Some(p.clone()), true, 0)); + log.lock().unwrap().on_status(1, "connected", "DualSense"); + log.lock().unwrap().on_report(1_000, &pad(0, 0)); + let mut on_disk_during = 0usize; + let (sha, head, n) = seal_shared_with(&log, 1, || 2_000, |job| { + // The HID logger can take the lock while the chunk is being written and fsynced... + let mut g = log.try_lock().expect("input-log mutex must be free during the disk write"); + assert_eq!(g.chunks, 0, "chain does not move before the chunk is durable"); + g.on_report(1_500, &pad(hid::CROSS, 0)); + drop(g); + let r = job.write(); + on_disk_during = read_file(&p).map(|(recs, _)| recs.len()).unwrap_or(0); + r + }); + // ...and what it adds is not in this chunk: header, status, 1 report, imu. + assert_eq!((n, on_disk_during), (4, 4)); + let l = log.lock().unwrap(); + assert_eq!((l.chunks, l.head.as_str()), (1, head.as_str())); + assert_eq!(l.pending().len(), 1); + assert_eq!(l.pending()[0]["t_us"], 1_500); + drop(l); + let (recs, chunks) = read_file(&p).unwrap(); + let (m, h2, errs) = verify_chunks(&recs, &chunks); + assert!(errs.is_empty(), "{errs:?}"); + assert_eq!((m[&1].as_str(), h2.as_str()), (sha.as_str(), head.as_str())); + let (_, _, n2) = seal_shared(&log, 2, || 3_000); + assert_eq!(n2, 1, "the record added during the write is sealed next"); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn seal_shared_failed_write_keeps_the_chain_and_the_order() { + let log = Mutex::new(InputLog::new(Some(std::env::temp_dir().join("unused-qorcap-inputs.jsonl")), true, 0)); + log.lock().unwrap().on_status(1, "connected", "DualSense"); + let before: Vec = log.lock().unwrap().pending().to_vec(); + let (sha, head, n) = seal_shared_with(&log, 1, || 10, |_| { + log.lock().unwrap().on_status(20, "unplugged", "gone"); + Err(std::io::Error::other("disk full")) + }); + assert_eq!((n, sha.as_str(), head.as_str()), (0, empty_hash().as_str(), GENESIS)); + let l = log.lock().unwrap(); + assert_eq!(l.chunks, 0); + let types: Vec<(String, u64)> = l.pending().iter().map(|r| (r["type"].as_str().unwrap().to_string(), r["t_us"].as_u64().unwrap())).collect(); + let want: Vec<(String, u64)> = before.iter().map(|r| (r["type"].as_str().unwrap().to_string(), r["t_us"].as_u64().unwrap())) + .chain(std::iter::once(("status".to_string(), 20))).collect(); + assert_eq!(types, want, "failed chunk first, then what arrived during the write"); + assert!(!l.seal_in_flight); + } + + #[test] + fn weave_snapshot_follows_pad_state_and_presses_only_when_attached() { + let snap = std::sync::Arc::new(Mutex::new(crate::livetap::WeaveCtx::default())); + let mut off = InputLog::new(None, true, 5); + off.on_status(1, "connected", "DualSense"); + assert_eq!(*snap.lock().unwrap(), crate::livetap::WeaveCtx::default(), "no snapshot attached: nothing published"); + let mut l = InputLog::new(None, true, 5); + l.set_weave_snap(snap.clone()); + assert_eq!(snap.lock().unwrap().pad, "searching"); + l.on_status(1, "connected", "DualSense"); + assert_eq!(snap.lock().unwrap().pad, "connected"); + l.on_report(1_000, &pad(0, 0)); + l.on_report(9_000, &pad(hid::CROSS, 0)); + assert_eq!(snap.lock().unwrap().last_press.as_ref().map(|p| p.2.clone()), Some("CROSS".into())); + l.on_status(20_000, "unplugged", "gone"); + assert_eq!(*snap.lock().unwrap(), l.weave_ctx()); + assert_eq!(snap.lock().unwrap().pad, "unplugged"); + } #[test] fn recent_presses_keep_short_taps_for_the_console() { let mut l = InputLog::new(None, true, 1_700_000_000_000); diff --git a/src/livetap.rs b/src/livetap.rs index 5559328..1797a0c 100644 --- a/src/livetap.rs +++ b/src/livetap.rs @@ -21,7 +21,9 @@ use std::fs::OpenOptions; use std::io::Write; use std::net::{IpAddr, Ipv4Addr, SocketAddr, ToSocketAddrs, UdpSocket}; use std::path::Path; -use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread; pub const SCHEMA: &str = "qorweave/v0"; pub const ENV_URL: &str = "QORCAP_LIVE_TAP"; @@ -84,6 +86,8 @@ pub struct TapStats { pub cues: u64, pub datagrams: u64, pub send_errors: u64, + /// TS slices dropped because the tap worker was behind. Not dshow drops; the recording is unchanged. + pub queue_drops: u64, pub last_hash: Option, } @@ -105,10 +109,38 @@ struct State { stats: TapStats, } -pub struct LiveTap { every: u32, state: Option> } +pub struct LiveTap { + every: u32, + state: Option>>, + tx: Option>, + /// Frame numbering kept on the submitting (capture) thread. Only that thread locks it, so it never + /// waits on the worker. A dropped slice still advances it, so later cue `f` values keep the + /// frame_clock.jsonl rule instead of shifting by the frames that were dropped. + count: Mutex, + /// TS slices dropped because the worker was behind. Atomic: the capture thread never takes `state`. + queue_drops: AtomicU64, +} + +#[derive(Default)] +struct FrameCount { file: Option, frame: i64, last_pts: Option } + +/// Work item for the tap worker. The capture thread only try_sends this; it never waits on disk or UDP. +struct TapWork { + data: Vec, + t_mono_us: u64, + file: String, + pat: Option>, + pmt: Option>, + ctx: WeaveCtx, + /// Frame index and last video PTS at the start of this slice, from `FrameCount`. + start: (i64, Option), +} + +/// A few queued slices is enough; if the worker falls behind we drop tap data, never the recording. +const TAP_QUEUE: usize = 8; impl LiveTap { - pub fn off() -> Self { Self { every: DEFAULT_EVERY, state: None } } + pub fn off() -> Self { Self { every: DEFAULT_EVERY, state: None, tx: None, count: Mutex::new(FrameCount::default()), queue_drops: AtomicU64::new(0) } } pub fn open(dest: Option, every: u32, session: &str, fps: u32, bundle: Option<&Path>) -> Self { let Some(dest) = dest else { return Self::off() }; @@ -126,58 +158,124 @@ impl LiveTap { "note": "Practice-lane live tap. Each cue is also sent in-band as timed ID3 on the live stream. Not in the hash chain or the receipt."}); let _ = writeln!(f, "{h}"); } + let state = Arc::new(Mutex::new(st)); + let (tx, rx) = mpsc::sync_channel::(TAP_QUEUE); + let state_w = state.clone(); + let spawned = thread::Builder::new().name("qorcap-livetap".into()).spawn(move || { + while let Ok(w) = rx.recv() { + let mut g = state_w.lock().unwrap_or_else(|e| e.into_inner()); + let ctx = w.ctx; + ingest(&mut g, every, &w.data, w.t_mono_us, &w.file, w.pat.as_deref(), w.pmt.as_deref(), &|| ctx.clone(), Some(w.start)); + } + }); + if let Err(e) = spawned { + log::error!("live tap: cannot start the tap worker: {e}; tap stays off (recording unchanged)"); + return Self::off(); + } log::info!("live tap ON -> udp://{dest} (practice lane, weave cue every {every} frames, recording unchanged)"); - Self { every, state: Some(Mutex::new(st)) } + Self { every, state: Some(state), tx: Some(tx), count: Mutex::new(FrameCount::default()), queue_drops: AtomicU64::new(0) } } pub fn enabled(&self) -> bool { self.state.is_some() } pub fn stats(&self) -> Option { - self.state.as_ref().map(|m| m.lock().unwrap_or_else(|e| e.into_inner()).stats.clone()) + self.state.as_ref().map(|m| { + let mut s = m.lock().unwrap_or_else(|e| e.into_inner()).stats.clone(); + s.queue_drops = self.queue_drops.load(Ordering::Relaxed); + s + }) } /// `data` is exactly the TS bytes that just reached the recording. `pat`/`pmt` are the cached tables /// (sent first whenever a new part file starts). `ctx` is called only when a cue is due. + /// Synchronous; used by unit tests. The capture loop calls `submit` so this work cannot stall ffmpeg. pub fn observe(&self, data: &[u8], t_mono_us: u64, file: &str, pat: Option<&[u8]>, pmt: Option<&[u8]>, ctx: &dyn Fn() -> WeaveCtx) { let Some(m) = &self.state else { return }; let mut g = m.lock().unwrap_or_else(|e| e.into_inner()); - let st = &mut *g; - if st.file.as_deref() != Some(file) { - st.file = Some(file.to_string()); - st.frame = 0; - st.last_pts = None; - if let Some(p) = pat { if p.len() == TS { st.pmt_pid = ring::pat_pmt_pid(p); push(st, p); } } - if let Some(p) = pmt { if p.len() == TS { let q = rewrite_or_pass(st, p); push(st, &q); } } + ingest(&mut g, self.every, data, t_mono_us, file, pat, pmt, ctx, None); + } + + /// Non-blocking enqueue for the capture thread. `ctx` must already be a snapshot: the caller must not + /// take the input-log mutex. Never locks `state` and never waits on the worker, UDP or disk. If the + /// worker is behind, the slice is dropped and counted in `queue_drops`; the recording is unchanged. + pub fn submit(&self, data: &[u8], t_mono_us: u64, file: &str, pat: Option<&[u8]>, pmt: Option<&[u8]>, ctx: WeaveCtx) { + let Some(tx) = &self.tx else { return }; + let start = self.count.lock().unwrap_or_else(|e| e.into_inner()).advance(data, file); + let work = TapWork { + data: data.to_vec(), + t_mono_us, + file: file.to_string(), + pat: pat.map(|p| p.to_vec()), + pmt: pmt.map(|p| p.to_vec()), + ctx, + start, + }; + if tx.try_send(work).is_err() { + self.queue_drops.fetch_add(1, Ordering::Relaxed); } - let mut cached_ctx: Option = None; + } +} + +impl FrameCount { + /// Same frame rule as `ingest` (new part file restarts at 0; one frame per new video PES PTS; + /// our own ID3 PID is skipped). Returns the position at the start of `data`. + fn advance(&mut self, data: &[u8], file: &str) -> (i64, Option) { + if self.file.as_deref() != Some(file) { + self.file = Some(file.to_string()); + self.frame = 0; + self.last_pts = None; + } + let start = (self.frame, self.last_pts); for pkt in data.chunks_exact(TS) { - let pid = ring::pid(pkt); - if pid == 0 && pkt[1] & 0x40 != 0 { if let Some(pp) = ring::pat_pmt_pid(pkt) { st.pmt_pid = Some(pp); } } - if Some(pid) == st.pmt_pid && pkt[1] & 0x40 != 0 { - let q = rewrite_or_pass(st, pkt); - push(st, &q); - continue; - } - if pid == ID3_PID { continue; } // never forward a foreign packet on our PID - push(st, pkt); + if ring::pid(pkt) == ID3_PID { continue; } let Some(pts) = crate::frameclock::video_pes_pts(pkt) else { continue }; - if st.last_pts == Some(pts) { continue; } - st.last_pts = Some(pts); - let frame = st.frame; - st.frame += 1; - st.stats.frames += 1; - if !st.weave_ok || frame % self.every as i64 != 0 { continue; } - let c = cached_ctx.get_or_insert_with(ctx).clone(); - let (line, h) = cue_json(st, &c, file, frame, pts, t_mono_us); - for p in id3_pes_packets(&mut st.cc, &id3_txxx(CUE_KEY, &line), pts) { push(st, &p); } - st.prev = h.clone(); - st.n += 1; - st.stats.cues += 1; - st.stats.last_hash = Some(h); - if let Some(f) = st.log.as_mut() { let _ = writeln!(f, "{line}"); } + if self.last_pts == Some(pts) { continue; } + self.last_pts = Some(pts); + self.frame += 1; + } + start + } +} + +#[allow(clippy::too_many_arguments)] +fn ingest(st: &mut State, every: u32, data: &[u8], t_mono_us: u64, file: &str, pat: Option<&[u8]>, pmt: Option<&[u8]>, ctx: &dyn Fn() -> WeaveCtx, start: Option<(i64, Option)>) { + if st.file.as_deref() != Some(file) { + st.file = Some(file.to_string()); + st.frame = 0; + st.last_pts = None; + if let Some(p) = pat { if p.len() == TS { st.pmt_pid = ring::pat_pmt_pid(p); push(st, p); } } + if let Some(p) = pmt { if p.len() == TS { let q = rewrite_or_pass(st, p); push(st, &q); } } + } + // Worker path: the submitting thread's count, so slices dropped before this one do not shift `f`. + if let Some((frame, last_pts)) = start { st.frame = frame; st.last_pts = last_pts; } + let mut cached_ctx: Option = None; + for pkt in data.chunks_exact(TS) { + let pid = ring::pid(pkt); + if pid == 0 && pkt[1] & 0x40 != 0 { if let Some(pp) = ring::pat_pmt_pid(pkt) { st.pmt_pid = Some(pp); } } + if Some(pid) == st.pmt_pid && pkt[1] & 0x40 != 0 { + let q = rewrite_or_pass(st, pkt); + push(st, &q); + continue; } - flush(st); + if pid == ID3_PID { continue; } // never forward a foreign packet on our PID + push(st, pkt); + let Some(pts) = crate::frameclock::video_pes_pts(pkt) else { continue }; + if st.last_pts == Some(pts) { continue; } + st.last_pts = Some(pts); + let frame = st.frame; + st.frame += 1; + st.stats.frames += 1; + if !st.weave_ok || frame % every as i64 != 0 { continue; } + let c = cached_ctx.get_or_insert_with(ctx).clone(); + let (line, h) = cue_json(st, &c, file, frame, pts, t_mono_us); + for p in id3_pes_packets(&mut st.cc, &id3_txxx(CUE_KEY, &line), pts) { push(st, &p); } + st.prev = h.clone(); + st.n += 1; + st.stats.cues += 1; + st.stats.last_hash = Some(h); + if let Some(f) = st.log.as_mut() { let _ = writeln!(f, "{line}"); } } + flush(st); } fn rewrite_or_pass(st: &mut State, pkt: &[u8]) -> Vec { @@ -440,4 +538,83 @@ mod tests { assert_eq!(cues[1]["h"].as_str().unwrap(), chain_hash(&pre)); let _ = std::fs::remove_dir_all(&dir); } + + fn frame_pkt(k: u64) -> [u8; TS] { crate::frameclock::video_ts_packet(0x100, 900 + 1500 * k) } + + fn wait_for(mut done: impl FnMut() -> bool) { + let end = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !done() { + assert!(std::time::Instant::now() < end, "tap worker did not catch up"); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + } + + #[test] + fn submit_never_waits_on_a_busy_worker_and_counts_queue_drops() { + let rx = UdpSocket::bind("127.0.0.1:0").unwrap(); + let dir = std::env::temp_dir().join(format!("qorcap-livetap-q-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let tap = std::sync::Arc::new(LiveTap::open(Some(rx.local_addr().unwrap()), 1, "1700", 60, Some(&dir))); + const N: u64 = 40; + // Stand in for a worker stuck on UDP / disk: hold its state lock. + let busy = tap.state.as_ref().unwrap().lock().unwrap(); + let (done_tx, done_rx) = mpsc::channel(); + let t2 = tap.clone(); + std::thread::spawn(move || { + for k in 0..N { t2.submit(&frame_pkt(k), 1_000 + k, "video.mp4", None, None, WeaveCtx::default()); } + let _ = done_tx.send(()); + }); + done_rx.recv_timeout(std::time::Duration::from_secs(2)).expect("submit blocked on the tap worker"); + let drops = tap.queue_drops.load(Ordering::Relaxed); + assert!(drops >= N - TAP_QUEUE as u64 - 1, "queue holds at most {TAP_QUEUE} (+1 in the worker); dropped {drops}"); + drop(busy); + wait_for(|| { let s = tap.stats().unwrap(); s.frames + s.queue_drops == N }); + // Slices after the gap: their cues must still carry the recording's frame index. + for k in N..N + 5 { + tap.submit(&frame_pkt(k), 1_000 + k, "video.mp4", None, None, WeaveCtx::default()); + wait_for(|| { let s = tap.stats().unwrap(); s.frames + s.queue_drops == k + 1 }); + } + let s = tap.stats().unwrap(); + assert_eq!(s.queue_drops, drops); + assert!(drops > 0); + // Every cue that was cut still names its true frame: dropped slices did not shift `f`. + let log = std::fs::read_to_string(dir.join(FILE_NAME)).unwrap(); + let cues: Vec = log.lines().skip(1).map(|l| serde_json::from_str(l).unwrap()).collect(); + assert_eq!(cues.len() as u64, N + 5 - drops); + assert_eq!(cues.last().unwrap()["f"], N + 4); + for c in &cues { + let k = (c["pts"].as_u64().unwrap() - 900) / 1500; + assert_eq!(c["f"].as_u64().unwrap(), k, "{c}"); + assert_eq!(c["t"].as_u64().unwrap(), 1_000 + k); + } + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn submit_matches_observe_when_nothing_is_dropped() { + let rx = UdpSocket::bind("127.0.0.1:0").unwrap(); + let base = std::env::temp_dir().join(format!("qorcap-livetap-eq-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&base); + let ctx = WeaveCtx { wall_anchor_ms: 1_700_000_000_000, pad: "connected".into(), seal: "off".into(), seal_hash: None, last_press: Some((1, 1_700_000_000_100, "CROSS".into())) }; + let mut logs = vec![]; + for (name, async_) in [("sync", false), ("async", true)] { + let dir = base.join(name); + let tap = LiveTap::open(Some(rx.local_addr().unwrap()), 3, "1700", 60, Some(&dir)); + for k in 0..20u64 { + let file = if k < 12 { "video.mp4" } else { "video_part2.mp4" }; + let data: Vec = [frame_pkt(k), frame_pkt(k)].concat(); // repeated PTS counts once + if async_ { + tap.submit(&data, 10 * k, file, None, Some(&pmt_packet()), ctx.clone()); + wait_for(|| tap.stats().unwrap().frames == k + 1); + } else { + tap.observe(&data, 10 * k, file, None, Some(&pmt_packet()), &|| ctx.clone()); + } + } + assert_eq!(tap.stats().unwrap().queue_drops, 0); + logs.push(std::fs::read_to_string(dir.join(FILE_NAME)).unwrap()); + } + assert_eq!(logs[0], logs[1]); + assert!(logs[0].lines().count() > 5); + let _ = std::fs::remove_dir_all(&base); + } }