From 41db94c801451852e940aab95923b02751461ac5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 09:29:08 +0000 Subject: [PATCH] runner: tests check the Runner's order and the ends it counts (P1R-C04) The macOS stage of main CI after #105 failed once in the Runner's DevGuard tests, and its re-run passed. Looping those tests here found one that fails about 1 run in 25, and 43 runs in 120 with 8 at a time: an_independent_runner_answers_registration_before_the_next_claim compared the moments two test tasks read the registration's answer and the next Claim's answer, while the Runner only orders its writes. It now takes the Claim's answer first and checks that the registration's answer is already waiting on the first connection. Five tests that #105 added or changed took a process's exit for its end. The Runner counts a process as ended once its output ended too: that starts its retention, makes an acknowledgement count and numbers its exit event. The output can end later, when a child that another test spawns holds the pipe until its exec, which is slow on macOS for a new executable. The retention tests now wait for the end of the output, and the two Claim tests wait until the Runner numbered the exit. Only tests change. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/runner/src/process.rs | 15 ++++++- crates/runner/src/wire.rs | 83 +++++++++++++++++++++++++----------- 2 files changed, 71 insertions(+), 27 deletions(-) diff --git a/crates/runner/src/process.rs b/crates/runner/src/process.rs index 5d05ba8..4c0e4ac 100644 --- a/crates/runner/src/process.rs +++ b/crates/runner/src/process.rs @@ -3269,7 +3269,10 @@ mod tests { runner.kill_host_workspace("demo").unwrap(); } - /// A quick process `process_id` that ended, run by `runner`. + /// A quick process `process_id` that ended, run by `runner`: its root exited and its output + /// ended, which is when the runner counts it as ended. The output can end well after the + /// exit: a child that another test spawns while the pipe or terminal is open holds a copy + /// until its exec, and on macOS the first exec of a new executable is slow. async fn ended_process( runner: &InProcessRunner, ws: &Workspace, @@ -3283,7 +3286,15 @@ mod tests { ); req.tty = tty; runner.spawn_host(ws, req).await.unwrap(); - wait_exited(runner, process_id).await; + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let status = runner.process_status(process_id).await.unwrap(); + if status.state == ProcessState::Exited && status.eof { + return; + } + assert!(Instant::now() < deadline, "{status:?}"); + tokio::time::sleep(Duration::from_millis(10)).await; + } } fn released(runner: &InProcessRunner, process_id: &ProcessId) -> bool { diff --git a/crates/runner/src/wire.rs b/crates/runner/src/wire.rs index 087b2e2..96c41b9 100644 --- a/crates/runner/src/wire.rs +++ b/crates/runner/src/wire.rs @@ -2413,7 +2413,6 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn an_independent_runner_answers_registration_before_the_next_claim() { use super::independent::{Host, STORE}; - use std::time::Instant; let dir = private_directory(); let credential = dir.path().join("consumer.secret"); std::fs::OpenOptions::new() @@ -2467,36 +2466,34 @@ mod tests { assert_eq!(first.claim(STORE).await.unwrap().epoch, 1); let registering = first.send(RunnerOp::Registration, Some(1)).await; let detaching = first.send(RunnerOp::Detach, None).await; - // Both are answered before the connection ends; the next connection claims the - // Runner as soon as the detach is answered. + // The detach is answered at once, the registration once its session ends. let mut answered = Vec::new(); - let mut claiming = None; - while let Some(reply) = first.next_response().await { - let id = reply.request_id.unwrap(); - if id == detaching { - let mut next = host.connect_raw().await; - claiming = Some(tokio::spawn(async move { - let hello = next.call(RunnerOp::Hello, None).await; - let claim = next.claim(STORE).await; - (hello, claim, Instant::now()) - })); - } - answered.push((id, Instant::now())); + while !answered.contains(&detaching) { + let reply = first.next_response().await.expect("the detach is answered"); + answered.push(reply.request_id.unwrap()); } - let registered_at = answered - .iter() - .find(|(id, _)| *id == registering) - .expect("the registration is answered") - .1; - let (hello, claim, claimed_at) = - claiming.expect("the detach is answered").await.unwrap(); + // The next connection claims the Runner as soon as the detach is answered, and is + // granted the Claim only once the registration was answered: by then that answer + // waits on the first connection. This is the order of the Runner's answers, not the + // order in which two tasks here would read them. + let mut next = host.connect_raw().await; assert_eq!( - stated(hello), + stated(next.call(RunnerOp::Hello, None).await), (RunnerLifecycle::Independent, true, true, false) ); - let claim = claim.unwrap(); + let claim = next.claim(STORE).await.unwrap(); assert_eq!((claim.epoch, claim.previous_controller), (2, false)); - assert!(registered_at <= claimed_at); + if !answered.contains(®istering) { + assert!( + first.unread() > 0, + "the Claim was granted before the registration was answered" + ); + } + // Both are answered before the connection ends. + while let Some(reply) = first.next_response().await { + answered.push(reply.request_id.unwrap()); + } + assert!(answered.contains(®istering), "{answered:?}"); } } @@ -2724,6 +2721,8 @@ mod tests { pub(in crate::wire::tests) stop: StopHandle, pub(in crate::wire::tests) runner: InProcessRunner, claims: Arc>>, + /// The number of the last exit the Runner numbered. + numbered: Arc>, } impl Host { @@ -2738,6 +2737,7 @@ mod tests { let socket = dir.path().join("runner.sock"); let listener = Arc::new(tokio::net::UnixListener::bind(&socket).unwrap()); let (runner, events) = host_worker(); + let numbered = events.numbered.clone(); let claims = Arc::new(std::sync::Mutex::new(Vec::new())); let noted = claims.clone(); let independent = configure( @@ -2765,6 +2765,7 @@ mod tests { stop, runner, claims, + numbered, } } @@ -2801,6 +2802,22 @@ mod tests { fn claims(&self) -> Vec { self.claims.lock().unwrap().clone() } + + /// Wait until the Runner numbered `count` exits. It numbers a process's exit once + /// the process ended, its output included, which can be well after its root + /// exited: a child that another test spawns while the pipe is open holds a copy + /// until its exec, and on macOS the first exec of a new executable is slow. + async fn exits_numbered(&self, count: u64) { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + let numbered = *self.numbered.lock().unwrap(); + if numbered >= count { + return; + } + assert!(Instant::now() < deadline, "{numbered} exits numbered"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } } impl Drop for Host { @@ -2843,6 +2860,20 @@ mod tests { id } + /// How many bytes the Runner wrote to this connection that are not read yet. + #[cfg(feature = "devguard")] + pub(in crate::wire::tests) fn unread(&self) -> usize { + use std::os::fd::AsRawFd; + let mut count: libc::c_int = 0; + // SAFETY: FIONREAD stores the number of bytes ready to be read from the socket + // in `count`, an `int` this function owns. + let result = unsafe { + libc::ioctl(self.read.as_ref().as_raw_fd(), libc::FIONREAD, &mut count) + }; + assert_eq!(result, 0, "{}", std::io::Error::last_os_error()); + usize::try_from(count).unwrap() + } + /// The next response, past any event, or none once the connection ended. pub(in crate::wire::tests) async fn next_response(&mut self) -> Option { loop { @@ -4049,6 +4080,7 @@ mod tests { // A process that ends while the client reconnects. std::fs::write(root.path().join("go"), "").unwrap(); exited(&host.runner, &gated).await; + host.exits_numbered(1).await; let mut claims = gateway.claims(); relay.hold(false); match next_link_event(&mut events).await { @@ -4379,6 +4411,7 @@ mod tests { exited(&host.runner, &quick).await.termination, Some(ProcessTermination::Exited) ); + host.exits_numbered(1).await; let (on_exit, mut exited_ids) = exits(); let second = host.connect_reporting_exits(on_exit).await; let claim = second.claimed().unwrap();