From 3b769982991914b2bf142a9b8ec556e31a7245f4 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:06:04 -0700 Subject: [PATCH 1/3] fix(supervisor): release registry ownership the moment its owner drops Registry ownership is an flock on supervisor.lock, and flock belongs to the open file description. A child process spawned by any other thread holds a duplicate of every descriptor between fork and exec (close-on-exec only closes it at exec), so closing the owner's descriptor left the root owned until that child exec'd. A successor opening the registry in that window was refused as Busy. Seen as intermittent Busy failures in tests that drop a supervisor and cold-open its registry while neighbouring tests spawn processes (provider_accepted_completion_retries_after_registry_success_on_fresh_healthy_startup, retained_output_owned_preparing_replacement_keeps_committed_prior, recovery_five_natural_successes_exhaust_durable_budget_until_manual_retry). OwnerLock now unlocks explicitly on drop, which releases the lock for every duplicate at once, and uses std's File::try_lock/unlock instead of raw libc flock calls. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/src/registry.rs | 92 +++++++++++++++------- 1 file changed, 62 insertions(+), 30 deletions(-) diff --git a/crates/freshell-supervisor/src/registry.rs b/crates/freshell-supervisor/src/registry.rs index e5ff52e43..aae172c86 100644 --- a/crates/freshell-supervisor/src/registry.rs +++ b/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 { + 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(); From 2ea595224c78ab638ddbcaa1595717ee1a7b414e Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:12:28 -0700 Subject: [PATCH 2/3] test(supervisor): expire mountpoint budgets at the stall each mode exercises provider_mountpoint_candidate_target_joins_owned_preparation_and_cleanup ran on the wall clock with 250 ms preparation/cleanup budgets and a 50 ms caller cancellation. Under machine load those bounds expired at arbitrary points: mid-request (the fixture's request read failed with "early eof"), mid private exchange (UnexpectedEof in the fixture's helper), or before success paths finished ("image lookup timed out" in the prepared and empty modes). The test now runs on a paused tokio clock held still by a blocking task (tokio's documented way to stop auto-advance during real I/O), so no budget can expire on its own. The fixture reports where it parks the operation, and the test advances the clock past the first bound only at that stall: the withheld private exchange after the helper's exit was answered (timeout, cancel-private-exchange) or the withheld image lookup (slow-image). If the code under test wedges, the hold ends after 60 real seconds and the test's own bound fails it instead of letting it hang (checked by removing the operation and image-lookup deadlines). Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/Cargo.toml | 3 + .../src/backend/provider_mountpoints.rs | 74 +++++++++++++++++-- 2 files changed, 70 insertions(+), 7 deletions(-) diff --git a/crates/freshell-supervisor/Cargo.toml b/crates/freshell-supervisor/Cargo.toml index 5590283ae..e0f42248d 100644 --- a/crates/freshell-supervisor/Cargo.toml +++ b/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"] } diff --git a/crates/freshell-supervisor/src/backend/provider_mountpoints.rs b/crates/freshell-supervisor/src/backend/provider_mountpoints.rs index 7e24dfe8c..fe9c86241 100644 --- a/crates/freshell-supervisor/src/backend/provider_mountpoints.rs +++ b/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::)); let requests = Arc::new(Mutex::new(Vec::::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::().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=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 From 92ee90428fb636fecdac91b32482eb7f013a5616 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 23:18:52 -0700 Subject: [PATCH 3/3] fix(supervisor): a control connection lets go of the supervisor before it replies serve_control answered each connection in a detached task that kept its Supervisor clone, and so the registry's Arc and owner lock, until after it had written the reply. A client that read its answer and then handed the registry to a successor could be refused as Busy whenever that task's worker was descheduled between the write and the end of the task. provider_accepted_completion_retries_after_registry_success_on_fresh_healthy_startup does exactly that: it reads a Recover reply, drops its handles and cold-opens the registry. A 10-20 ms stall injected after the reply write made it fail with Busy at the reopen nearly every time. The per-connection work is now serve_control_connection, which drops its supervisor handle before it writes the reply, so the reply happens after the release. The regression test writes the reply through a stream that cold-opens the registry at the moment the reply goes out; it was refused as Busy before this change and succeeds after it, for both a valid and a malformed request. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/src/service.rs | 215 +++++++++++++++++++--- 1 file changed, 186 insertions(+), 29 deletions(-) diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs index 03a17f45e..ac85cad96 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -3144,38 +3144,48 @@ pub async fn serve_control(supervisor: Supervisor, socket_path: &Path) -> Result serde_json::json!({"installationId":supervisor.registry.installation_id(),"controlEpoch":supervisor.registry.control_epoch(),"socket":socket_path}), ); loop { - let (mut stream, _) = listener.accept().await.map_err(|e| e.to_string())?; - let supervisor = supervisor.clone(); - tokio::spawn(async move { - let envelope: Result, _> = read_frame(&mut stream).await; - let reply = match envelope { - Ok(envelope) => supervisor.dispatch(envelope).await, - Err(error) => AdminReply { - request_id: RequestId::new(), - result: Err(RuntimeError::new( - if matches!(error, freshell_runtime_protocol::FrameError::TooLarge) { - RuntimeErrorCode::FrameTooLarge - } else { - RuntimeErrorCode::InvalidRequest - }, - error.to_string(), - )), - }, - }; - let limit = if matches!( - reply.result, - Ok(AdminResult::FreshAgentHistory(_) | AdminResult::FreshAgentSnapshot(_)) - ) { - freshell_runtime_protocol::MAX_NATIVE_HISTORY_FRAME_BYTES - } else { - freshell_runtime_protocol::MAX_CONTROL_FRAME_BYTES - }; - let _ = - freshell_runtime_protocol::write_frame_with_limit(&mut stream, &reply, limit).await; - }); + let (stream, _) = listener.accept().await.map_err(|e| e.to_string())?; + tokio::spawn(serve_control_connection(supervisor.clone(), stream)); } } +/// Answers the one request a control connection carries. +/// +/// The connection lets go of its supervisor handle before it writes the +/// reply. A client holding its answer may then hand the registry to a +/// successor at once: this connection no longer keeps the registry owned, +/// however long its worker is descheduled before the task finishes. +async fn serve_control_connection(supervisor: Supervisor, mut stream: S) +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + let envelope: Result, _> = read_frame(&mut stream).await; + let reply = match envelope { + Ok(envelope) => supervisor.dispatch(envelope).await, + Err(error) => AdminReply { + request_id: RequestId::new(), + result: Err(RuntimeError::new( + if matches!(error, freshell_runtime_protocol::FrameError::TooLarge) { + RuntimeErrorCode::FrameTooLarge + } else { + RuntimeErrorCode::InvalidRequest + }, + error.to_string(), + )), + }, + }; + drop(supervisor); + let limit = if matches!( + reply.result, + Ok(AdminResult::FreshAgentHistory(_) | AdminResult::FreshAgentSnapshot(_)) + ) { + freshell_runtime_protocol::MAX_NATIVE_HISTORY_FRAME_BYTES + } else { + freshell_runtime_protocol::MAX_CONTROL_FRAME_BYTES + }; + let _ = freshell_runtime_protocol::write_frame_with_limit(&mut stream, &reply, limit).await; +} + pub fn default_backend(socket: impl Into) -> Arc { Arc::new(DockerEngineBackend::new(socket)) } @@ -4434,3 +4444,150 @@ pub(crate) fn crash_if(point: &'static str) { } #[cfg(not(feature = "runtime-test-faults"))] pub(crate) fn crash_if(_point: &'static str) {} + +#[cfg(test)] +mod control_connection_tests { + use super::{default_backend, serve_control_connection, Supervisor, SupervisorConfig}; + use crate::registry::Registry; + use freshell_runtime_protocol::{ + read_frame, AdminCommand, AdminReply, AdminResult, ControlRole, Envelope, RequestId, + RuntimeErrorCode, + }; + use std::path::{Path, PathBuf}; + use std::pin::Pin; + use std::sync::{Arc, Mutex as StdMutex}; + use std::task::{Context, Poll}; + use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream, ReadBuf}; + + /// What a cold open of the registry returned when the reply was written. + type ReopenAtReply = Arc>>>; + + /// The supervisor end of a control connection. When the reply starts to + /// go out, it cold-opens the registry, as a successor would once the + /// client holds its answer, and records whether that open was refused. + struct ReopenOnReply { + stream: DuplexStream, + registry_root: PathBuf, + reopened: ReopenAtReply, + } + + impl AsyncRead for ReopenOnReply { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_read(cx, buf) + } + } + + impl AsyncWrite for ReopenOnReply { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let mut reopened = self.reopened.lock().unwrap(); + if reopened.is_none() { + *reopened = Some( + Registry::open(&self.registry_root, None) + .map(drop) + .map_err(|error| format!("{error:?}")), + ); + } + drop(reopened); + Pin::new(&mut self.stream).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stream).poll_flush(cx) + } + + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_shutdown(cx) + } + } + + fn supervisor(root: &Path, registry_root: &Path) -> Supervisor { + let host_binary_path = root.join("synthetic-host"); + std::fs::write(&host_binary_path, "synthetic-host-artifact").unwrap(); + Supervisor::new( + Registry::open(registry_root, None).unwrap(), + default_backend(root.join("absent-docker.sock")), + SupervisorConfig { + runtime_root: root.join("runtime"), + control_socket_path: root.join("control.sock"), + host_binary_path, + image_ref: format!("sha256:{}", "a".repeat(64)), + test_run_id: "control-connection".into(), + control_secret: "synthetic-control-secret".into(), + lifecycle_log: root.join("lifecycle.jsonl"), + admission: crate::admission::AdmissionPolicy::default(), + }, + ) + .unwrap() + } + + /// A control client that has read its reply may hand the registry to a + /// successor at once: the connection that answered it must already have + /// let go of the supervisor, and with it the registry ownership, before + /// the reply went out. Otherwise the successor's open is refused as Busy + /// whenever the connection's worker is descheduled between writing the + /// reply and finishing. + #[tokio::test] + async fn control_connection_releases_supervisor_before_it_replies() { + let valid_request = || { + let mut envelope = + Envelope::new(RequestId::new(), ControlRole::Web, AdminCommand::Health); + envelope.auth = Some("synthetic-control-secret".into()); + serde_json::to_vec(&envelope).unwrap() + }; + for (request, answered) in [ + (valid_request(), true), + (b"not a control envelope".to_vec(), false), + ] { + let temp = tempfile::tempdir().unwrap(); + let registry_root = temp.path().join("registry"); + // The connection holds the only handle to this supervisor. + let supervisor = supervisor(temp.path(), ®istry_root); + let (mut client, server) = tokio::io::duplex(64 * 1024); + let reopened = ReopenAtReply::default(); + let connection = tokio::spawn(serve_control_connection( + supervisor, + ReopenOnReply { + stream: server, + registry_root: registry_root.clone(), + reopened: reopened.clone(), + }, + )); + + client + .write_all(&(request.len() as u32).to_be_bytes()) + .await + .unwrap(); + client.write_all(&request).await.unwrap(); + let reply: AdminReply = read_frame(&mut client).await.unwrap(); + if answered { + assert!( + matches!(reply.result, Ok(AdminResult::Health { .. })), + "the supervisor answers a valid request" + ); + } else { + assert_eq!( + reply.result.unwrap_err().code, + RuntimeErrorCode::InvalidRequest + ); + } + connection.await.unwrap(); + + assert_eq!( + reopened.lock().unwrap().take(), + Some(Ok(())), + "the registry must be free for a successor by the time the reply is written" + ); + } + } +}