fix(supervisor): make four flaky supervisor tests reliable

crates/freshell-supervisor/Cargo.toml

@@ -30,4 +30,7 @@ libc = "0.2"
 
 [dev-dependencies]
 tempfile = "3"
+# `test-util` unlocks `#[tokio::test(start_paused = true)]`, so tests expire
+# budgets by advancing a paused clock instead of racing the wall clock.
+tokio = { version = "1", features = ["test-util"] }
 tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "json"] }

crates/freshell-supervisor/src/backend/provider_mountpoints.rs

@@ -414,8 +414,32 @@ mod tests {
         ProviderConfigBind, ProviderResourceKind, ProviderResourceMount,
     };
 
-    #[tokio::test]
+    /// Holds the paused test clock still while real socket I/O runs: tokio
+    /// never auto-advances a paused clock while a blocking task is running.
+    /// A budget then expires only when the test advances the clock at the
+    /// exact stall it exercises, however loaded the machine is. The real-time
+    /// limit matters only if the code under test wedges: the hold ends, the
+    /// clock auto-advances, and the test's own bounds fail it instead of
+    /// letting it hang.
+    struct FrozenClock {
+        _release_on_drop: std::sync::mpsc::Sender<()>,
+    }
+
+    impl FrozenClock {
+        fn hold() -> Self {
+            let (release, held) = std::sync::mpsc::channel::<()>();
+            drop(tokio::task::spawn_blocking(move || {
+                let _ = held.recv_timeout(std::time::Duration::from_secs(60));
+            }));
+            Self {
+                _release_on_drop: release,
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
     async fn provider_mountpoint_candidate_target_joins_owned_preparation_and_cleanup() {
+        use std::collections::BTreeSet;
         use std::sync::{Arc, Mutex};
         use tokio::task::JoinSet;
         for mode in [
@@ -429,6 +453,7 @@ mod tests {
             "cancel-private-exchange",
             "slow-image",
         ] {
+            let _clock = FrozenClock::hold();
             let owned = tempfile::tempdir().unwrap();
             let home = owned.path().join("scratch-home");
             std::fs::create_dir(&home).unwrap();
@@ -453,6 +478,9 @@ mod tests {
             let expected = Arc::new(Mutex::new(None::<Value>));
             let requests = Arc::new(Mutex::new(Vec::<String>::new()));
             let removed = Arc::new(std::sync::atomic::AtomicBool::new(false));
+            // The fixture reports where it parks the operation, so the test
+            // expires a budget exactly at the stall the mode exercises.
+            let (stalls, mut stalled) = tokio::sync::mpsc::unbounded_channel::<&'static str>();
             let (body, seen, deleted, fixture_home, fixture_daemon) = (
                 expected.clone(),
                 requests.clone(),
@@ -469,7 +497,7 @@ mod tests {
                         _=&mut stopped=>break,
                         accepted=listener.accept()=>{
                             let (mut stream,_)=accepted.unwrap();
-                            let (body,seen,deleted,home,daemon,private_jobs)=(body.clone(),seen.clone(),deleted.clone(),fixture_home.clone(),fixture_daemon.clone(),private_jobs.clone());
+                            let (body,seen,deleted,home,daemon,private_jobs,stalls)=(body.clone(),seen.clone(),deleted.clone(),fixture_home.clone(),fixture_daemon.clone(),private_jobs.clone(),stalls.clone());
                             jobs.spawn(async move {
                                 let mut header=Vec::new(); let mut byte=[0];
                                 while !header.ends_with(b"\r\n\r\n") {stream.read_exact(&mut byte).await.unwrap();header.push(byte[0]);assert!(header.len()<=32*1024);}
@@ -478,7 +506,7 @@ mod tests {
                                 let length=header.lines().find_map(|line|line.strip_prefix("Content-Length: ")).unwrap().trim().parse::<usize>().unwrap();
                                 assert!(length<=1024*1024);let mut raw=vec![0;length];stream.read_exact(&mut raw).await.unwrap();
                                 seen.lock().unwrap().push(line.clone());
-                                let (status,value)=if line.contains("/images/") {if mode=="slow-image" {tokio::time::sleep(Duration::from_secs(10)).await;} (200,json!({"Config":{"Env":["PATH=/usr/bin","ANTHROPIC_API_KEY=PRIVATE_UNUSED_IMAGE_CREDENTIAL"]}}))}
+                                let (status,value)=if line.contains("/images/") {if mode=="slow-image" {stalls.send("image-withheld").unwrap();std::future::pending::<()>().await;} (200,json!({"Config":{"Env":["PATH=/usr/bin","ANTHROPIC_API_KEY=PRIVATE_UNUSED_IMAGE_CREDENTIAL"]}}))}
                                 else if line.contains("/info ") {(200,json!({"ID":daemon,"SecurityOptions":["name=rootless"]}))}
                                 else if line.contains("/containers/create?") { *body.lock().unwrap()=Some(serde_json::from_slice(&raw).unwrap());(if mode=="lost-create-ack" {503} else {201},json!({"Id":"c".repeat(64)}))}
                                 else if line.starts_with("GET") && line.contains("/containers/") {
@@ -492,9 +520,10 @@ mod tests {
                                 else if line.contains("/start ") {
                                     let expected=body.lock().unwrap().as_ref().unwrap().clone();
                                     let exchange=expected["HostConfig"]["Mounts"].as_array().unwrap().iter().find(|mount|mount["Target"]=="/run/freshell-input-exchange").unwrap()["Source"].as_str().unwrap().to_owned();
+                                    let stalls=stalls.clone();
                                     private_jobs.lock().unwrap().push(tokio::spawn(async move {
                                         let mut channel=UnixStream::connect(Path::new(&exchange).join("reply.sock")).await.unwrap();
-                                        if matches!(mode,"timeout"|"cancel-private-exchange") {tokio::time::sleep(Duration::from_secs(10)).await;return;}
+                                        if matches!(mode,"timeout"|"cancel-private-exchange") {stalls.send("exchange-withheld").unwrap();std::future::pending::<()>().await;}
                                         let length=channel.read_u32().await.unwrap() as usize; assert!(length<=1024*1024);
                                         let mut raw=vec![0;length];channel.read_exact(&mut raw).await.unwrap();
                                         let selected:Vec<Mountpoint>=serde_json::from_slice(&raw).unwrap();
@@ -512,6 +541,7 @@ mod tests {
                                 else {(500,json!({}))};
                                 let raw=value.to_string();let response=format!("HTTP/1.1 {status} fixture\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{raw}",raw.len());
                                 let _=stream.write_all(response.as_bytes()).await;
+                                if line.contains("/wait?") {stalls.send("wait-answered").unwrap();}
                             });
                         },
                         result=jobs.join_next(),if !jobs.is_empty()=>{result.unwrap().unwrap();}
@@ -528,9 +558,39 @@ mod tests {
             });
             let backend = DockerEngineBackend::new(owned.path().join("docker.sock"));
             let mut helper = None;
-            let outcome=tokio::time::timeout(if mode=="cancel-private-exchange" {Duration::from_millis(50)} else {Duration::from_secs(3)},backend.run_provider_mountpoint_helper(
-                MountpointHelperTarget {image:"synthetic-image",volume:"owned-candidate-volume",binary:&binary,daemon:&daemon,labels:json!({"com.freshell.soul-id":"synthetic-owner","com.freshell.source-request-id":"synthetic-request"})},selected,(Duration::from_millis(250),Duration::from_millis(250)),&mut helper
-            )).await;
+            let budget = Duration::from_millis(250);
+            let caller_bound = if mode == "cancel-private-exchange" {
+                Duration::from_millis(50)
+            } else {
+                Duration::from_secs(3)
+            };
+            // Only the stalled modes move the clock, and only once every
+            // request before the stall has been answered: the private
+            // exchange (with the helper's exit already reported) or the image
+            // lookup is the one thing still pending when its bound expires.
+            let mut stall: BTreeSet<&str> = match mode {
+                "timeout" | "cancel-private-exchange" => {
+                    BTreeSet::from(["exchange-withheld", "wait-answered"])
+                }
+                "slow-image" => BTreeSet::from(["image-withheld"]),
+                _ => BTreeSet::new(),
+            };
+            let expire_at_stall = async {
+                if stall.is_empty() {
+                    return std::future::pending().await;
+                }
+                while !stall.is_empty() {
+                    stall.remove(stalled.recv().await.expect("fixture reports its stall"));
+                }
+                tokio::time::advance(caller_bound.min(budget)).await;
+                std::future::pending().await
+            };
+            let outcome = tokio::select! {
+                outcome = tokio::time::timeout(caller_bound,backend.run_provider_mountpoint_helper(
+                    MountpointHelperTarget {image:"synthetic-image",volume:"owned-candidate-volume",binary:&binary,daemon:&daemon,labels:json!({"com.freshell.soul-id":"synthetic-owner","com.freshell.source-request-id":"synthetic-request"})},selected,(budget,budget),&mut helper
+                )) => outcome,
+                () = expire_at_stall => unreachable!("the stall clock never completes"),
+            };
             if mode == "cancel-private-exchange" {
                 assert!(outcome.is_err());
                 let (name, body) = helper

crates/freshell-supervisor/src/registry.rs

@@ -90,8 +90,47 @@ struct RegistryInner {
     db_path: PathBuf,
     installation_id: InstallationId,
     control_epoch: u64,
-    #[allow(dead_code)]
-    lock_file: std::fs::File,
+    /// Held for its `Drop`: the registry root stays owned until the last
+    /// handle to this registry is gone.
+    _owner_lock: OwnerLock,
+}
+
+/// Exclusive supervisor ownership of one registry root (`supervisor.lock`).
+struct OwnerLock(std::fs::File);
+
+impl OwnerLock {
+    fn acquire(root: &Path) -> Result<Self, RegistryError> {
+        let file = OpenOptions::new()
+            .create(true)
+            .read(true)
+            .write(true)
+            .truncate(false)
+            .open(root.join("supervisor.lock"))?;
+        match file.try_lock() {
+            Ok(()) => Ok(Self(file)),
+            Err(std::fs::TryLockError::WouldBlock) => Err(RegistryError::Busy),
+            Err(std::fs::TryLockError::Error(error)) => Err(RegistryError::Io(error)),
+        }
+    }
+}
+
+impl Drop for OwnerLock {
+    fn drop(&mut self) {
+        // The lock belongs to the open file description, which every
+        // duplicate descriptor shares, including the copy that a child
+        // process spawned by any other thread holds between fork and exec
+        // (close-on-exec only closes it at exec). Closing this descriptor
+        // alone would keep the root owned until that child execs, so a
+        // successor opening the registry right after this owner is gone
+        // would be refused as Busy. Unlocking releases ownership now.
+        if let Err(error) = self.0.unlock() {
+            tracing::warn!(
+                event = "runtime.registry.owner_lock_release_failed",
+                error = %error,
+                "Registry owner lock release failed; it is released when every duplicate descriptor closes"
+            );
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -345,14 +384,7 @@ impl Registry {
         // The registry contains protected queued input and durable provider
         // identity. Keep the entire SQLite/WAL directory supervisor-private.
         std::fs::set_permissions(root, std::fs::Permissions::from_mode(0o700))?;
-        let lock_path = root.join("supervisor.lock");
-        let lock_file = OpenOptions::new()
-            .create(true)
-            .read(true)
-            .write(true)
-            .truncate(false)
-            .open(lock_path)?;
-        acquire_exclusive_lock(&lock_file)?;
+        let owner_lock = OwnerLock::acquire(root)?;
 
         let db_path = root.join("runtime.sqlite3");
         let mut conn = open_connection(&db_path)?;
@@ -400,7 +432,7 @@ impl Registry {
                 db_path,
                 installation_id,
                 control_epoch: next_epoch,
-                lock_file,
+                _owner_lock: owner_lock,
             }),
         })
     }
@@ -3109,25 +3141,6 @@ fn durable_sync_parent(parent: Option<&Path>) -> Result<(), RegistryError> {
     Ok(())
 }
 
-#[cfg(unix)]
-fn acquire_exclusive_lock(file: &std::fs::File) -> Result<(), RegistryError> {
-    use std::os::fd::AsRawFd;
-    // SAFETY: the descriptor is valid for this call and owned by `file`.
-    let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
-    if rc == 0 {
-        Ok(())
-    } else if std::io::Error::last_os_error().kind() == std::io::ErrorKind::WouldBlock {
-        Err(RegistryError::Busy)
-    } else {
-        Err(RegistryError::Io(std::io::Error::last_os_error()))
-    }
-}
-
-#[cfg(not(unix))]
-fn acquire_exclusive_lock(_file: &std::fs::File) -> Result<(), RegistryError> {
-    Ok(())
-}
-
 #[cfg(feature = "runtime-test-faults")]
 pub(crate) fn failpoint(name: &'static str) -> Result<(), RegistryError> {
     if std::env::var("FRESHELL_RUNTIME_DB_FAILPOINT")
@@ -4107,6 +4120,25 @@ mod tests {
         drop(first);
     }
 
+    #[test]
+    fn dropped_registry_releases_ownership_while_a_spawning_child_shares_its_lock() {
+        let dir = tempfile::tempdir().unwrap();
+        let first = Registry::open(dir.path(), None).unwrap();
+        // A child process spawned by any other thread holds a duplicate of
+        // every descriptor between fork and exec, and the ownership lock
+        // belongs to the open file description that duplicate shares.
+        let spawning_child_copy = first.inner._owner_lock.0.try_clone().unwrap();
+        drop(first);
+        let reopened = Registry::open(dir.path(), None)
+            .expect("the dropped owner's lock must not outlive it in a duplicate descriptor");
+        assert!(
+            matches!(Registry::open(dir.path(), None), Err(RegistryError::Busy)),
+            "the reopened registry owns the root exclusively"
+        );
+        drop(reopened);
+        drop(spawning_child_copy);
+    }
+
     #[tokio::test]
     async fn handle_does_not_exist_until_container_identity_is_committed() {
         let dir = tempfile::tempdir().unwrap();