diff --git a/crates/net/src/observe.rs b/crates/net/src/observe.rs index 84482fa..7d86b4d 100644 --- a/crates/net/src/observe.rs +++ b/crates/net/src/observe.rs @@ -57,7 +57,11 @@ impl Observer { if size.saturating_add(text.len() as u64) > max_file { continue; } - let _ = file.write_all(text.as_bytes()).await; + // tokio hands the write to a thread; without the flush the next line may measure + // the file before this one landed and write past the cap. + if file.write_all(text.as_bytes()).await.is_ok() { + let _ = file.flush().await; + } } }); Self { diff --git a/crates/share-worker/src/mac/sck.rs b/crates/share-worker/src/mac/sck.rs index b7e30d9..9ecc838 100644 --- a/crates/share-worker/src/mac/sck.rs +++ b/crates/share-worker/src/mac/sck.rs @@ -168,6 +168,13 @@ pub fn find_window(target: &WindowTarget) -> Result, String> } } +/// The reason when the system never answered the start of a stream. +const NOT_IN_TIME: &str = "the system did not start the capture in time"; + +/// How often a start the system never answered is tried again. On a loaded machine (a CI +/// runner right after another capture ended) the first start can be lost; a later one works. +const START_TRIES: usize = 3; + /// Starts a stream and waits until the system says it runs. fn start(stream: &SCStream) -> Result<(), String> { connect_to_window_server(); @@ -187,7 +194,7 @@ fn start(stream: &SCStream) -> Result<(), String> { match rx.recv_timeout(ANSWER) { Ok(None) => Ok(()), Ok(Some(why)) => Err(why), - Err(_) => Err("the system did not start the capture in time".into()), + Err(_) => Err(NOT_IN_TIME.into()), } } @@ -427,6 +434,16 @@ impl SckSource { /// Starts capturing `window`. Fails with a reason that [`is_permission_error`] recognises /// when Screen Recording is not allowed. pub fn open(window: &SCWindow) -> Result { + let mut tries = 1; + loop { + match Self::open_once(window) { + Err(why) if why == NOT_IN_TIME && tries < START_TRIES => tries += 1, + other => return other, + } + } + } + + fn open_once(window: &SCWindow) -> Result { // SAFETY: plain construction and configuration of ScreenCaptureKit objects, and calls // that retain what they are given. unsafe { diff --git a/crates/share-worker/tests/access.rs b/crates/share-worker/tests/access.rs index 72a8d1b..4c98f05 100644 --- a/crates/share-worker/tests/access.rs +++ b/crates/share-worker/tests/access.rs @@ -311,27 +311,53 @@ mod real_window { }; let folder = std::env::temp_dir().join(format!("anyplay-mac-real-{}", std::process::id())); let path = folder.join("real.mp4"); - let mut child = Command::new(WORKER) - .args(["--clips", "2", "--pid", &calculator.pid.to_string()]) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn() - .expect("the worker runs"); - let mut input = child.stdin.take().unwrap(); - let output = child.stdout.take().unwrap(); - let (tx, rx) = std::sync::mpsc::channel(); - std::thread::spawn(move || { - use std::io::BufRead; - for line in std::io::BufReader::new(output) - .lines() - .map_while(Result::ok) - { - let _ = tx.send(line); + // A CI runner now and then lets a freshly started worker's capture fail at its start + // ("the system did not start the capture in time", or a declined TCC although the test + // process itself was allowed a moment earlier). That is the runner, not the worker: + // start a new worker, up to three times. Any other end is a failure at once. + let mut attempt = 0; + let (mut child, mut input, rx) = loop { + attempt += 1; + let mut child = Command::new(WORKER) + .args(["--clips", "2", "--pid", &calculator.pid.to_string()]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .expect("the worker runs"); + let input = child.stdin.take().unwrap(); + let output = child.stdout.take().unwrap(); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + use std::io::BufRead; + for line in std::io::BufReader::new(output) + .lines() + .map_while(Result::ok) + { + let _ = tx.send(line); + } + }); + let deadline = Instant::now() + Duration::from_secs(60); + let first = loop { + let left = deadline.saturating_duration_since(Instant::now()); + let line = rx.recv_timeout(left).expect("an event in time"); + let value: serde_json::Value = serde_json::from_str(&line).unwrap(); + if value["event"] == "ready" || value["event"] == "ended" { + break value; + } + }; + if first["event"] == "ready" { + break (child, input, rx); } - }); + let reason = first["reason"].as_str().unwrap_or_default(); + assert!( + attempt < 3 && (reason == "no_permission" || reason == "no_window"), + "{first}" + ); + let _ = child.wait(); + }; let wait = |event: &str| { - let deadline = Instant::now() + Duration::from_secs(40); + let deadline = Instant::now() + Duration::from_secs(60); loop { let left = deadline.saturating_duration_since(Instant::now()); let line = rx.recv_timeout(left).expect("an event in time"); @@ -342,7 +368,6 @@ mod real_window { assert_ne!(value["event"], "ended", "{value}"); } }; - wait("ready"); std::thread::sleep(Duration::from_millis(4500)); writeln!( input,