Fix: two CI flakes (macOS capture start, observation log cap)

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 {

crates/share-worker/src/mac/sck.rs

@@ -168,6 +168,13 @@ pub fn find_window(target: &WindowTarget) -> Result<Retained<SCWindow>, 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<SckSource, String> {
+        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<SckSource, String> {
         // SAFETY: plain construction and configuration of ScreenCaptureKit objects, and calls
         // that retain what they are given.
         unsafe {

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,