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 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(); 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" + ); + } + } +}