diff --git a/core/docs/architecture/76-native-mount-session.md b/core/docs/architecture/76-native-mount-session.md index 4932c3641..5f9d7ffc2 100644 --- a/core/docs/architecture/76-native-mount-session.md +++ b/core/docs/architecture/76-native-mount-session.md @@ -211,6 +211,44 @@ terminal service fence before the second drain. A failed or short abort reports `Abort` with its original errno or byte count; later drain, Revoke and Close failures retain the recorded abort effects in `TeardownCustody.forced`. +### Observing the abort path + +A traced Unmount that takes that path reports it beside the ordinary reply, in +the daemon's `unmount` telemetry section of the existing `Observed` envelope +(issue #2). The reply itself is unchanged: success still answers `Unmounted`, +and a stop keeps its `TeardownCustody`. The section's stable keys are: + +| Field | Meaning | +| --- | --- | +| `stage` | The boundary the unmount was at: `Join` when the stalled drain was ended by the abort, `Abort` when the one write was short or failed, and `Revoke` or `Close` when a later step stopped with the abort recorded | +| `abort` | The connection's single abort write: `Written`, `Short` or `Failed` | +| `bytes` | The count of a `Short` write only | +| `errno` | The original `Failed` write errno only | +| `detached` | Kernel detach was already known when the abort was made | +| `loops_exited`, `loops_joined` | Receive loops exited and joined, as observed at that stop or completion | +| `active`, `retained` | Request owner counts at the same observation | +| `waited_ms` | Milliseconds from the start of the connection drain to the end of the abort path | + +An unmount that never stalled carries no such section, and no normal or forced +teardown carries one otherwise. The same fields are written once to stderr as +a `LAYERFS_DAEMON_UNMOUNT_ABORT =…` line when the abort path +ends, traced or not; it is one diagnostic line per abort, never per-request +logging. That line's `stage` is therefore the abort path's own (`Join`, or +`Abort` for a short or failed write), while a later `Revoke` or `Close` stop +appears in the retained entry's section for the reply that reports it. +Aggregating these facts in a Host view belongs to the +[dsh-layerfs](https://github.com/Ephemeral-AI-Lab/dsh-layerfs) follow-up, outside +this repository. + +A retained Revoke or Close stop keeps that observation in its entry, so the +traced reply that later reports the stop's custody carries the same section at +the later boundary. `layerfs-api/tests/unmount_observability.rs` covers both +forms: the completed abort path through the actual daemon executable, and the +retained stop through the in-process service, where the entry's own owner can +be poisoned or closed. That target mounts a Workspace, so `tools/ci_tests.py` +lists it with the SDK mount targets: the fast lane leaves it out and the +nightly mount lane runs it in its own container with `/dev/fuse`. + ## Forced unmount [`ForceUnmount`](80-forced-teardown.md) takes the session out of the entry the diff --git a/core/docs/architecture/80-forced-teardown.md b/core/docs/architecture/80-forced-teardown.md index 6ede45548..4626574b8 100644 --- a/core/docs/architecture/80-forced-teardown.md +++ b/core/docs/architecture/80-forced-teardown.md @@ -140,6 +140,27 @@ A later normal or forced unmount returns the retained stage, detail and abort facts unchanged. An unavailable control cannot be forced, and a consumed control is never written again. Successful completion still replies `Unmounted`. +A traced normal unmount reports the path it took as the `unmount` telemetry +section of its reply, whether it completed or stopped: `stage` (`Join`, +`Abort`, `Revoke` or `Close`), `abort` (`Written`, `Short` or `Failed`), the +short write's `bytes` or the failed write's `errno`, `detached`, +`loops_exited`, `loops_joined`, `active`, `retained`, and `waited_ms` from the +stalled drain to the end of the abort path. The same fields form one +`LAYERFS_DAEMON_UNMOUNT_ABORT` stderr line, written when that path ends (traced +or not); a `Revoke` or `Close` stop that follows it is reported by the retained +entry's section, not by a second line. Both are observations of a normal +unmount only: a forced teardown already reports its `ForcedFacts` in the reply, +and neither envelope changes. The keys are described in +[the normal unmount](76-native-mount-session.md#observing-the-abort-path). +Aggregating them in a Host view is a +[dsh-layerfs](https://github.com/Ephemeral-AI-Lab/dsh-layerfs) follow-up, +outside this repository. + +A retained `Revoke` or `Close` stop reports the same section through the +traced reply that reports its custody. Both forms are staged in-process +through the entry's own owner and are covered beside the completed path in +`layerfs-api/tests/unmount_observability.rs`. + `ForceUnmount` remains for a stalled custody that is still forceable (one whose abort was not attempted) and for a live connection. It takes the session back and makes the same sequence with two differences. Step 1 is `NativeSession::abort_detached`, diff --git a/core/docs/architecture/README.md b/core/docs/architecture/README.md index 6f3fd5f1a..60f7cd7f8 100644 --- a/core/docs/architecture/README.md +++ b/core/docs/architecture/README.md @@ -434,8 +434,8 @@ process-specific cancellation, and it supplies no filesystem drain evidence. - [Native read observation and lookup custody](73-native-read-custody.md): atomic Workspace answer/count/read ownership, exact source association and bounded retirement; native Fuse integration remains R2. - [Native directory sources and indexed cookies](74-native-directory-custody.md): exact directory owners, accepted-entry cookie publication, current parent relationships and bounded live cleanup; native kernel integration remains R2. - [Native request dispatcher and direct ports](75-native-request-service.md): shared fixed workers, event-driven custody, actual Owner/Store adapters and Linux read callbacks with complete opcode accounting. -- [Native mount session, Ready and reversible unmount](76-native-mount-session.md): direct mount and session ownership, Attach/Locate/Ready records, reversible EBUSY probe, connection drain, drain-qualified revocation and Sandbox FUSE access; mutation is described in record 77, Commit in record 79 and forced unmount in record 80. -- [Forced teardown](80-forced-teardown.md): `ForceUnmount` admission order and typed refusals, one abort write, terminal fence, local drain and one plain detach, Revoke and Close, the forced facts kept with retained custody, and the limits (abort control availability, no exit from `Retained`, held Lifecycle slots). +- [Native mount session, Ready and reversible unmount](76-native-mount-session.md): direct mount and session ownership, Attach/Locate/Ready records, reversible EBUSY probe, connection drain, drain-qualified revocation and Sandbox FUSE access, and the traced abort-path observation of a normal unmount; mutation is described in record 77, Commit in record 79 and forced unmount in record 80. +- [Forced teardown](80-forced-teardown.md): `ForceUnmount` admission order and typed refusals, one abort write, terminal fence, local drain and one plain detach, Revoke and Close, the forced facts and traced observation kept with retained custody, and the limits (abort control availability, no exit from `Retained`, held Lifecycle slots). - [Compare and merge](81-compare-and-merge.md): page-skipping root, binding and line differences, and the three-way merge of roots with conflict markers and `.lww` rules, built on theirs through `update_filesystem`. - [One Commit per file write](85-commit-per-write-storage.md): what one Commit writes, why file content costs about its change (piece compare, content addressing, deltas, CDC), where the fixed per-Commit cost is and what F-027 removes. - [Global SQLite Store](84-global-sqlite-store.md): the one global database: lifecycle, the fixed Disposable/WAL/OFF profile, schema, objects and packs, connections, transactions, reclamation and the rules that do not change. diff --git a/engine/crates/layerfs-api/tests/unmount_observability.rs b/engine/crates/layerfs-api/tests/unmount_observability.rs new file mode 100644 index 000000000..15bf4b5c3 --- /dev/null +++ b/engine/crates/layerfs-api/tests/unmount_observability.rs @@ -0,0 +1,545 @@ +//! One traced normal unmount whose connection outlived its mount (issue #2). +//! +//! The actual daemon executable mounts a Workspace, a second mount namespace +//! holds a copy of it, so the daemon's plain detach is followed by a stalled +//! drain, and the unmount ends its own connection with one abort write +//! (#426, #460). A traced reply must then carry the `unmount` telemetry +//! section, and the daemon must write exactly one structured stderr line for +//! it. A second Workspace, unmounted without a copy, reports neither. +//! +//! The second test takes the same observation through the retained custody a +//! later Revoke or Close stop keeps for the reply that reports it. A retained +//! stop can only be staged in-process (the entry's owner is not a control +//! request), so it drives the production service over the same traced +//! exchange; the reply envelope is the test's, the outcome and section are the +//! daemon's. This target mounts FUSE, so `tools/ci_tests.py` routes it to the +//! nightly mount lane and the fast lane never runs it. +#![cfg(target_os = "linux")] +#[path = "support/daemon_binary.rs"] +mod daemon_binary; +#[allow(dead_code)] +#[path = "../../layerfs-daemon/tests/support/fusectl.rs"] +mod fusectl; +#[allow(dead_code)] +#[path = "../../layerfs-daemon/tests/support/installed_store.rs"] +mod installed; +#[allow(dead_code)] +#[path = "../../layerfs-daemon/tests/support/mounted.rs"] +mod mounted; +#[allow(dead_code)] +#[path = "../../layerfs-daemon/tests/support/host_install.rs"] +mod support; +use layerfs_api::internal::{take_telemetry, trace, Control}; +use layerfs_bridge::{ + control::{ + Answer, Call, Reply, Request, SdkValue, TeardownStage, UnmountControl, WorkspaceToken, + }, + daemon_setup::{DaemonLimits, DaemonSetup}, + duplex, +}; +use layerfs_daemon::{ + bootstrap::open_store, + control::{Failure, Service}, + telemetry, ServiceClass, +}; +use layerfs_history::WorkspaceId; +use layerfs_storage::ReservationBlocks; +use std::{ + fs::{self, File}, + io::{BufRead, BufReader, Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + os::unix::fs::PermissionsExt, + path::{Path, PathBuf}, + process::{Child, ChildStdin, Command, Stdio}, + sync::{mpsc, Arc}, + thread, + time::Duration, +}; + +struct Process(Child); +impl Drop for Process { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Removes a fixture's directory on every exit, including a panic, so an +/// aborted run leaves no kernel tree behind. +struct Scratch(PathBuf); +impl Drop for Scratch { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +/// The daemon's own startup accounts for every missing LISTEN line: a runner +/// without `/dev/fuse` or mount privileges exits before printing it. +fn startup(error: &str, log: &Path) -> ! { + let stderr = fs::read_to_string(log).unwrap_or_else(|read| format!("")); + panic!("{error}; daemon stderr:\n{stderr}") +} + +/// Starts the actual executable, installs the fixture's Store through it, and +/// returns the control owner that receives traced replies. +fn start(f: &support::Fixture, log: &Path) -> (Process, Control) { + fs::set_permissions(&f.directory, fs::Permissions::from_mode(0o700)).unwrap(); + let setup = DaemonSetup { + listen: "127.0.0.1:0".into(), + private_key: support::SERVER_PRIVATE, + control_peer: duplex::public_key(&support::CLIENT_PRIVATE).unwrap(), + store: f.project.manifest.locator.clone(), + overlay: f.directory.join("overlay.sqlite").to_str().unwrap().into(), + mounts: f.directory.join("mounts").to_str().unwrap().into(), + command_uid: 65534, + command_gid: 65534, + limits: DaemonLimits { + connections: 4, + handshake_ms: 3000, + cache_bytes: 64 * 1024, + namespaces: 8, + pager_kib: 1024, + serial_low_water: 0, + idle_exit_s: 0, + }, + existing_store: None, + }; + let config = f.directory.join("daemon.config"); + fs::write(&config, setup.encode().unwrap()).unwrap(); + fs::set_permissions(&config, fs::Permissions::from_mode(0o600)).unwrap(); + let mut process = Process( + Command::new(daemon_binary::path()) + .arg("--config") + .arg(&config) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::from(File::create(log).unwrap())) + .spawn() + .unwrap(), + ); + let stdout = process.0.stdout.take().unwrap(); + let (send, receive) = mpsc::sync_channel(1); + let reader = thread::spawn(move || { + let mut line = String::new(); + let result = BufReader::new(stdout.take(256)) + .read_line(&mut line) + .map(|_| line); + let _ = send.send(result); + }); + let line = match receive.recv_timeout(Duration::from_secs(5)) { + Ok(Ok(line)) => line, + Ok(Err(error)) => startup(&format!("no daemon stdout: {error}"), log), + // The reader thread still waits for the child, which Process kills. + Err(_) => startup("no LAYERFS_DAEMON_LISTEN within 5 s", log), + }; + reader.join().unwrap(); + let Some(address) = line + .strip_prefix("LAYERFS_DAEMON_LISTEN ") + .and_then(|rest| rest.trim().parse::().ok()) + else { + startup("daemon printed no LAYERFS_DAEMON_LISTEN", log) + }; + // The stalled drain waits out `drain_wait` (5 s here) before the abort, so + // this control owner waits longer than the fixture's handshake timeout. + let stream = + support::socket(TcpStream::connect_timeout(&address, Duration::from_secs(2)).unwrap()); + stream + .set_read_timeout(Some(Duration::from_secs(60))) + .unwrap(); + let mut connection = duplex::initiate( + stream, + &support::CLIENT_PRIVATE, + duplex::public_key(&support::SERVER_PRIVATE).unwrap(), + ) + .unwrap(); + layerfs_api::internal::install(&f.project, &mut connection).unwrap(); + (process, Control::new(connection)) +} + +/// Runs in a private copy of the mount namespace, so it holds a copy of the +/// Workspace mount without touching it. It answers one line for each word sent +/// and ends on any other word. +const COPY: &str = r#"echo held +while read word; do + [ "$word" = probe ] || exit 0 + if out=$(stat "$1/stall-probe" 2>&1); then echo ok; else echo "$out"; fi +done"#; + +/// The process holding the copied mount. Dropped, it is killed and reaped. +struct Copy { + child: Child, + stdin: ChildStdin, + lines: mpsc::Receiver, +} + +impl Copy { + fn spawn(mount: &Path) -> Self { + let mut child = Command::new("unshare") + .args(["-m", "--propagation", "private", "sh", "-c", COPY, "copy"]) + .arg(mount) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + let stdout = child.stdout.take().unwrap(); + let stdin = child.stdin.take().unwrap(); + let (send, lines) = mpsc::channel(); + // Ends when the process's output closes, whatever the test does. + thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + if send.send(line.unwrap_or_default()).is_err() { + break; + } + } + }); + let mut copy = Self { + child, + stdin, + lines, + }; + assert_eq!(copy.line(), "held"); + copy + } + fn line(&mut self) -> String { + self.lines + .recv_timeout(Duration::from_secs(10)) + .expect("the copy answered within its wait") + } + fn ask(&mut self, word: &str) -> String { + writeln!(self.stdin, "{word}").unwrap(); + self.line() + } + /// Asks it to leave and waits a bounded time for its own exit. + fn leave(&mut self) -> bool { + let _ = writeln!(self.stdin, "exit"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if self.child.try_wait().unwrap().is_some() { + return true; + } + thread::sleep(Duration::from_millis(10)); + } + false + } +} +impl Drop for Copy { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn field_names(section: &SdkValue) -> Vec { + match section { + SdkValue::Map(fields) => fields.iter().map(|(name, _)| name.clone()).collect(), + other => panic!("not a section: {other:?}"), + } +} + +/// The telemetry of the one traced reply received since the last take. +fn one(control: &mut Control) -> SdkValue { + let mut telemetry = layerfs_api::internal::take_telemetry(control); + assert_eq!(telemetry.len(), 1, "{telemetry:?}"); + telemetry.pop().unwrap() +} + +#[test] +fn a_traced_stalled_unmount_reports_its_abort_path_and_a_plain_one_stays_silent() { + support::shared_control(); + let f = support::Fixture::new("unmount-observability", None); + let _scratch = Scratch(f.directory.clone()); + let log = f.directory.join("stderr.log"); + let (process, mut control) = start(&f, &log); + let branch = f.project.branch.branch.id; + + // The first Workspace is mounted, and a copy of its mount outlives the + // daemon's plain detach. + let bound = layerfs_api::internal::workspace(&mut control) + .bind(WorkspaceId::from_authority([150; 32]).unwrap(), branch) + .unwrap(); + let ready = layerfs_api::internal::workspace(&mut control) + .attach(bound.token) + .unwrap(); + let mut copy = Copy::spawn(Path::new(&ready.directory)); + + layerfs_api::internal::trace(&mut control, Some([7; 32])); + layerfs_api::internal::workspace(&mut control) + .unmount(ready.token, false) + .unwrap(); + let after = one(&mut control); + let section = after + .get("unmount") + .expect("the abort path's telemetry section") + .clone(); + assert_eq!( + field_names(§ion), + [ + "stage", + "abort", + "detached", + "loops_exited", + "loops_joined", + "active", + "retained", + "waited_ms" + ] + ); + assert_eq!(section.get_text("stage"), Some("Join")); + assert_eq!(section.get_text("abort"), Some("Written")); + assert!(section.get_bool("detached")); + assert_eq!(section.get_int("active"), Some(0)); + assert_eq!(section.get_int("retained"), Some(0)); + let joined = section.get_int("loops_joined").unwrap(); + assert!(joined > 1, "{section:?}"); + assert_eq!(section.get_int("loops_exited"), Some(joined)); + let waited = section.get_int("waited_ms").unwrap(); + assert!((0..60_000).contains(&waited), "{section:?}"); + assert!(section.get("bytes").is_none() && section.get("errno").is_none()); + + // The copy's mount died with the connection; it finds nothing to look up. + let answer = copy.ask("probe"); + assert!( + answer.contains("No such file or directory") + || answer.contains("Transport endpoint is not connected"), + "copy's call answered {answer:?}" + ); + assert!(copy.leave(), "the copy did not exit by itself"); + drop(copy); + + // A second Workspace unmounts without a copy and takes no abort path. + layerfs_api::internal::trace(&mut control, None); + let plain = layerfs_api::internal::workspace(&mut control) + .bind(WorkspaceId::from_authority([151; 32]).unwrap(), branch) + .unwrap(); + let plain_ready = layerfs_api::internal::workspace(&mut control) + .attach(plain.token) + .unwrap(); + assert_ne!(plain_ready.directory, ready.directory); + layerfs_api::internal::trace(&mut control, Some([8; 32])); + layerfs_api::internal::workspace(&mut control) + .unmount(plain.token, false) + .unwrap(); + let after = one(&mut control); + assert!( + after.get("unmount").is_none(), + "an unmount that never stalled reports no abort path: {after:?}" + ); + + // One structured line for the one abort path, none for the plain unmount. + let text = fs::read_to_string(&log).unwrap(); + let lines: Vec<_> = text + .lines() + .filter(|line| line.starts_with("LAYERFS_DAEMON_UNMOUNT_ABORT ")) + .collect(); + assert_eq!(lines.len(), 1, "{text}"); + assert!( + lines[0].contains(" stage=Join abort=Written detached=true "), + "{}", + lines[0] + ); + assert!( + lines[0].contains(&format!(" loops_joined={joined} ")), + "{}", + lines[0] + ); + + layerfs_api::internal::trace(&mut control, None); + assert_eq!( + control.call(Request::EndSession).unwrap(), + Reply::SessionEnded + ); + drop(process); + f.cleanup(); +} + +/// The reply form the daemon application composes for these outcomes +/// (`application/diagnostics.rs`): a retained stop answers `Reply::Retained`, +/// and a traced reply carries the `unmount` section built from the abort +/// observation the entry kept. The application's startup owner is not a +/// library API and a retained Revoke or Close stop can only be staged +/// in-process, so the test runs the real `Service` and wraps its real outcome +/// here; only the envelope around the production observation is the test's. +fn observed(service: &Service, call: Call) -> Answer { + let operation = call + .request + .without_observation() + .expect("the client sent a one-layer observed request"); + let outcome = service.execute_control(operation); + let abort = match &outcome { + Ok(success) => success.abort.clone(), + Err(Failure::Retained(custody)) => service.retained_abort(custody.token), + Err(other) => panic!("unexpected control outcome: {other:?}"), + }; + let reply = match &outcome { + Ok(success) => success.reply.clone(), + Err(Failure::Retained(custody)) => Reply::Retained(custody.clone()), + Err(other) => panic!("unexpected control outcome: {other:?}"), + }; + let abort = abort.expect("a retained stop that aborted its connection keeps the observation"); + Answer { + id: call.id, + reply: Reply::Observed { + reply: Box::new(reply), + telemetry: SdkValue::Map(vec![( + "unmount".to_owned(), + telemetry::unmount_abort(&abort), + )]), + }, + } +} + +/// One traced Unmount through the SDK's control owner, answered by the +/// in-process service above: the original reply and its telemetry. +fn traced_unmount(service: Arc, token: WorkspaceToken) -> (Reply, Vec) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = + TcpStream::connect_timeout(&listener.local_addr().unwrap(), Duration::from_secs(2)) + .unwrap(); + let (server, _) = listener.accept().unwrap(); + for socket in [&client, &server] { + // The stalled drain waits seconds before the reply; the bounded wait + // is far above it and below the test ceiling. + socket + .set_read_timeout(Some(Duration::from_secs(60))) + .unwrap(); + socket + .set_write_timeout(Some(Duration::from_secs(60))) + .unwrap(); + } + let worker = thread::spawn(move || { + let mut channel = duplex::accept( + server, + &support::SERVER_PRIVATE, + duplex::public_key(&support::CLIENT_PRIVATE).unwrap(), + ) + .unwrap(); + let call = Call::decode(channel.receive.receive().unwrap()).unwrap(); + let answer = observed(&service, call); + channel.send.send(&answer.encode().unwrap()).unwrap(); + }); + let connection = duplex::initiate( + client, + &support::CLIENT_PRIVATE, + duplex::public_key(&support::SERVER_PRIVATE).unwrap(), + ) + .unwrap(); + let mut control = Control::new(connection); + trace(&mut control, Some([9; 32])); + let reply = control + .call(Request::Unmount(UnmountControl { + token, + discard: true, + })) + .unwrap(); + let telemetry = take_telemetry(&mut control); + worker.join().unwrap(); + (reply, telemetry) +} + +/// Stages the later boundary a retained stop reports: a failed Revoke job or an +/// already-closed overlay, as the product's own teardown sees them. +fn stage_stop(harness: &mounted::Harness, token: WorkspaceToken, stage: TeardownStage) { + let operation = harness.service.operation(token).unwrap(); + match stage { + TeardownStage::Revoke => { + let owner = operation.overlay().clone(); + let route = operation.workspace().route(); + // A failed owner job poisons only this Workspace's connection. + assert!(thread::spawn(move || { + owner.call::<(), _>(route, ServiceClass::General, None, |_| { + panic!("staged owner failure before Revoke") + }) + }) + .join() + .unwrap() + .is_err()); + } + TeardownStage::Close => { + // Revoke accepts a closed Workspace; a second Close refuses. + operation + .overlay() + .close(operation.workspace().route()) + .unwrap(); + } + _ => unreachable!("{stage:?}"), + } +} + +/// The abort path a retained Revoke or Close stop keeps, reported through a +/// traced reply: the same section a completed unmount carries, at the later +/// boundary the retained custody named. +fn retained_stop_reports_its_abort_path(stage: TeardownStage) { + fusectl::mount(); + let fixture = installed::Fixture::new(2, &format!("unmount-observability-{stage:?}")); + let _scratch = Scratch(fixture.directory.clone()); + let store = open_store( + fixture.config.clone(), + installed::BINDING, + installed::CURSOR, + 4, + 2 * 1024 * 1024, + ReservationBlocks::default(), + ) + .unwrap(); + let harness = mounted::Harness::new(store.clone(), &fixture.directory, fixture.branch); + let ready = harness.mount(1); + let token = ready.token; + let mut copy = Copy::spawn(Path::new(&ready.directory)); + stage_stop(&harness, token, stage); + + let (reply, telemetry) = traced_unmount(harness.service.clone(), token); + let left = copy.leave(); + drop(copy); + harness.stop_retained(); + drop(store); + + assert!(left, "the copy did not exit by itself"); + let Reply::Retained(custody) = &reply else { + panic!("the retained stop answered {reply:?}") + }; + assert_eq!(custody.token, token); + assert_eq!(custody.stage, stage); + assert_eq!(telemetry.len(), 1, "{telemetry:?}"); + let section = telemetry[0] + .get("unmount") + .expect("the retained stop's telemetry section") + .clone(); + assert_eq!( + field_names(§ion), + [ + "stage", + "abort", + "detached", + "loops_exited", + "loops_joined", + "active", + "retained", + "waited_ms" + ] + ); + assert_eq!( + section.get_text("stage"), + Some(format!("{stage:?}").as_str()) + ); + assert_eq!(section.get_text("abort"), Some("Written")); + assert!(section.get_bool("detached")); + assert_eq!(section.get_int("active"), Some(0)); + assert_eq!(section.get_int("retained"), Some(0)); + let joined = section.get_int("loops_joined").unwrap(); + assert!(joined > 1, "{section:?}"); + assert_eq!(section.get_int("loops_exited"), Some(joined)); + let waited = section.get_int("waited_ms").unwrap(); + assert!((0..60_000).contains(&waited), "{section:?}"); + assert!(section.get("bytes").is_none() && section.get("errno").is_none()); +} + +#[test] +fn a_retained_revoke_stop_reports_its_abort_path_in_a_traced_reply() { + retained_stop_reports_its_abort_path(TeardownStage::Revoke); +} + +#[test] +fn a_retained_close_stop_reports_its_abort_path_in_a_traced_reply() { + retained_stop_reports_its_abort_path(TeardownStage::Close); +} diff --git a/engine/crates/layerfs-daemon/src/application/diagnostics.rs b/engine/crates/layerfs-daemon/src/application/diagnostics.rs index 40d039c4e..9909fbfbf 100644 --- a/engine/crates/layerfs-daemon/src/application/diagnostics.rs +++ b/engine/crates/layerfs-daemon/src/application/diagnostics.rs @@ -315,6 +315,17 @@ impl Application { rows.finish() })(); let diagnostic = resources.into_diagnostic_error(result.err()); + // A normal unmount that took its abort path reports it with the reply, + // on success and on a retained stop; a later stop reads the + // observation the entry kept. + let abort = match &outcome { + Ok(success) => success.abort.clone(), + Err(Failure::Retained(custody)) => service + .as_ref() + .and_then(|service| service.retained_abort(custody.token)), + Err(Failure::After { original, .. }) => original.abort.clone(), + Err(_) => None, + }; match reports { // The SDK trace route reports a failed diagnostic write beside // the reply and keeps its connection. @@ -335,6 +346,7 @@ impl Application { &outcome, diagnostic.as_ref(), [sql, delta, fuse], + abort, ); (outcome, None, Some(telemetry)) } diff --git a/engine/crates/layerfs-daemon/src/control/attach.rs b/engine/crates/layerfs-daemon/src/control/attach.rs index 047813663..79fc292de 100644 --- a/engine/crates/layerfs-daemon/src/control/attach.rs +++ b/engine/crates/layerfs-daemon/src/control/attach.rs @@ -187,6 +187,7 @@ impl Service { observer: Some(session.observer()), evidence: Evidence::Other(Box::new(session)), forced: None, + abort: None, }))), }; } @@ -208,6 +209,7 @@ impl Service { observer, evidence: Evidence::Other(Box::new((failure, mount))), forced: None, + abort: None, }))); } let phase = if failure.mounted { @@ -239,6 +241,7 @@ impl Service { observer: None, evidence: Evidence::Other(Box::new((failure, mount, other))), forced: None, + abort: None, })), } } diff --git a/engine/crates/layerfs-daemon/src/control/detach.rs b/engine/crates/layerfs-daemon/src/control/detach.rs index 0bd63f90a..fe4f8ca57 100644 --- a/engine/crates/layerfs-daemon/src/control/detach.rs +++ b/engine/crates/layerfs-daemon/src/control/detach.rs @@ -5,19 +5,23 @@ use super::{ kernel::Kernel, registry::bound_mut, - serving::{stage, Attached, Evidence, Kept, Leaving}, - Failure, Service, Success, + serving::{stage, work, Attached, Evidence, Kept, Leaving, Stop}, + Failure, Service, Success, UnmountAbort, }; use crate::store::BoundWorkspace; use layerfs_bridge::control::{ AbortDisposition, Activity, CommitKnowledge, ControlCode, DetachDisposition, ForcedFacts, - MountPhase, TeardownStage, WorkspaceToken, + MountPhase, MountWork, TeardownStage, WorkspaceToken, }; use layerfs_fuse::{ mount::AbortWrite, - session::{Detach, DrainStage, Forced, SessionObserver}, + session::{Detach, DrainStage, Forced}, +}; +use std::{ + fmt, + sync::Arc, + time::{Duration, Instant}, }; -use std::{fmt, sync::Arc}; /// Abort effects after normal admission, which established no Commit custody /// and completed the one plain detach before attempting the write. @@ -34,6 +38,36 @@ pub(super) fn detached_abort_facts(forced: Option) -> Option, + detached: bool, + work: MountWork, + waited: Duration, +) -> Option> { + let forced = forced?; + Some(Box::new(UnmountAbort { + token, + stage, + abort: forced.abort, + detached, + work, + waited, + })) +} + +/// One structured stderr line for the abort path, on success and on a stop. +/// It is the only line this path writes and nothing is logged per request. +fn warn(abort: &Option>) { + if let Some(abort) = abort { + eprintln!("{}", crate::telemetry::unmount_abort_line(abort)); + } +} + /// Terminal admission over a connection-owning entry. No daemon-side request /// count refuses here: the kernel's own reversible answer is the fence, because /// every blocked caller, descriptor, working directory and mapping holds a @@ -95,26 +129,46 @@ impl Service { Err(errno) => { return Err(self.retain( token, - TeardownStage::Detach, - format!("plain detach: {errno}"), - Some(observer), + Stop { + stage: TeardownStage::Detach, + detail: format!("plain detach: {errno}"), + observer: Some(observer), + forced: None, + abort: None, + }, Box::new(session), - None, )) } } self.leaving(token, MountPhase::Draining); - let drained = match session.drain() { - Ok(drained) => drained, + // The stall this unmount waits out, from the connection drain to the + // end of the abort path; reported as `unmount.waited_ms` when the path + // ran, and as nothing when it did not. + let draining = Instant::now(); + let (drained, mut abort) = match session.drain() { + Ok(drained) => (drained, None), // Detached yet unjoined: the connection outlived its mount because // another mount namespace holds a copy of it (an engine or VM agent // outside the Sandbox). Make the same one abort write a forced // unmount makes, then drain again; a connection whose control is // unusable, or one that still stops, keeps its custody (#426, #460). Err(undrained) => match undrained.end_stalled() { - Ok(drained) => drained, + Ok(drained) => { + let observed = observer.facts(); + let abort = abort_observation( + token, + TeardownStage::Join, + drained.forced, + observed.detached, + work(&observed), + draining.elapsed(), + ); + warn(&abort); + (drained, abort) + } Err(undrained) => { let facts = detached_abort_facts(undrained.forced); + let observed = observer.facts(); let (stopped, detail) = match undrained.forced.map(|forced| forced.abort) { Some(AbortWrite::Failed(errno)) => ( TeardownStage::Abort, @@ -128,16 +182,35 @@ impl Service { stage(undrained.stage), format!( "connection drain stopped at {:?}: abort {abort:?}, abort_bound {}, loops {:?}, work {:?}", - undrained.stage, observer.facts().abort_bound, + undrained.stage, observed.abort_bound, undrained.loops, undrained.work ), ), }; + let abort = abort_observation( + token, + stopped, + undrained.forced, + observed.detached, + work(&observed), + draining.elapsed(), + ); + warn(&abort); let evidence = match undrained.stage { DrainStage::Join => Evidence::Stalled(undrained), _ => Evidence::Other(undrained), }; - return Err(self.keep(token, stopped, detail, Some(observer), evidence, facts)); + return Err(self.keep( + token, + Stop { + stage: stopped, + detail, + observer: Some(observer), + forced: facts, + abort, + }, + evidence, + )); } }, }; @@ -149,17 +222,23 @@ impl Service { ) { Ok(()) => {} Err(failure) => { + if let Some(abort) = abort.as_mut() { + abort.stage = TeardownStage::Revoke; + } return Err(self.retain( token, - TeardownStage::Revoke, - failure.wire().detail, - Some(observer), + Stop { + stage: TeardownStage::Revoke, + detail: failure.wire().detail, + observer: Some(observer), + forced: facts, + abort, + }, Box::new((failure, drained)), - facts, )); } }; - self.close(token, &workspace, Some(drained)) + self.close(token, &workspace, Some(drained), abort) } /// Describes the exact retained kernel owner for an operator. Observation /// only: nothing is transferred, released, retried or settled. @@ -194,33 +273,22 @@ impl Service { pub(super) fn retain( &self, token: WorkspaceToken, - stage: TeardownStage, - detail: String, - observer: Option, + stop: Stop, evidence: Box, - forced: Option, ) -> Failure { - let evidence = Evidence::Other(evidence); - self.keep(token, stage, detail, observer, evidence, forced) + self.keep(token, stop, Evidence::Other(evidence)) } /// The same, with the owner already sorted into its kind. - pub(super) fn keep( - &self, - token: WorkspaceToken, - stage: TeardownStage, - detail: String, - observer: Option, - evidence: Evidence, - forced: Option, - ) -> Failure { + pub(super) fn keep(&self, token: WorkspaceToken, stop: Stop, evidence: Evidence) -> Failure { let mut entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); let mut kept = Kept { ready: None, - stage, - detail, - observer, + stage: stop.stage, + detail: stop.detail, + observer: stop.observer, evidence, - forced, + forced: stop.forced, + abort: stop.abort, }; let mut binding = bound_mut(&mut entries, token).ok(); if let Some(Kernel::Leaving(leaving)) = binding.as_ref().map(|binding| &binding.kernel) { diff --git a/engine/crates/layerfs-daemon/src/control/force.rs b/engine/crates/layerfs-daemon/src/control/force.rs index 6eb780c61..548b0dada 100644 --- a/engine/crates/layerfs-daemon/src/control/force.rs +++ b/engine/crates/layerfs-daemon/src/control/force.rs @@ -3,7 +3,7 @@ //! A normal unmount stalled at Join is forced the same way, its plain detach //! counting as the one detach. #[cfg(target_os = "linux")] -use super::serving::{stage, work, Attached, Evidence, Leaving}; +use super::serving::{stage, work, Attached, Evidence, Leaving, Stop}; use super::{ kernel::Kernel, registry::{bound_mut, Binding}, @@ -269,11 +269,14 @@ impl Service { Err(refusal) if stalled => { return Err(self.retain( token, - TeardownStage::Join, - format!("abort refused before any effect: {refusal:?}"), - Some(observer), + Stop { + stage: TeardownStage::Join, + detail: format!("abort refused before any effect: {refusal:?}"), + observer: Some(observer), + forced: None, + abort: None, + }, Box::new(session), - None, )) } Err(refusal) => { @@ -308,11 +311,14 @@ impl Service { facts.abort = abort; return Err(self.retain( token, - TeardownStage::Abort, - detail, - Some(observer), + Stop { + stage: TeardownStage::Abort, + detail, + observer: Some(observer), + forced: Some(facts), + abort: None, + }, Box::new(session), - Some(facts), )); } let drained = match session.force_drain() { @@ -325,14 +331,17 @@ impl Service { facts.fenced = undrained.forced.map_or(0, |forced| forced.terminal_replies); return Err(self.retain( token, - stage(undrained.stage), - format!( - "forced drain stopped at {:?}: detach {attempt:?}, loops {:?}, work {:?}", - undrained.stage, undrained.loops, undrained.work - ), - Some(observer), + Stop { + stage: stage(undrained.stage), + detail: format!( + "forced drain stopped at {:?}: detach {attempt:?}, loops {:?}, work {:?}", + undrained.stage, undrained.loops, undrained.work + ), + observer: Some(observer), + forced: Some(facts), + abort: None, + }, undrained, - Some(facts), )); } }; @@ -354,15 +363,18 @@ impl Service { Err(failure) => { return Err(self.retain( token, - TeardownStage::Revoke, - failure.wire().detail, - Some(observer), + Stop { + stage: TeardownStage::Revoke, + detail: failure.wire().detail, + observer: Some(observer), + forced: Some(facts), + abort: None, + }, Box::new((failure, drained)), - Some(facts), )); } } - let kernel = self.closed(token, &workspace, Some(drained), Some(facts.clone()))?; + let kernel = self.closed(token, &workspace, Some(drained), Some(facts.clone()), None)?; // One observation of the closed namespace; a failed one claims no // state. Its completion is read and dropped here. let (cleanup, observation_failure) = match self @@ -396,6 +408,7 @@ impl Service { commit: None, observation_failure, kernel, + abort: None, }, ) } diff --git a/engine/crates/layerfs-daemon/src/control/mod.rs b/engine/crates/layerfs-daemon/src/control/mod.rs index 9d42da651..30c627344 100644 --- a/engine/crates/layerfs-daemon/src/control/mod.rs +++ b/engine/crates/layerfs-daemon/src/control/mod.rs @@ -19,9 +19,11 @@ mod serving; mod status; mod types; mod unmount; +mod unmount_abort; pub use kernel::{KernelConfig, KernelServing}; pub(crate) use observation::{reply_token, request_token}; pub use registry::Service; pub(crate) use serve::answer_call; pub use serve::{ServeCause, ServeFailure, Served}; pub use types::{Failure, KernelEvidence, KernelFailure, Success}; +pub use unmount_abort::UnmountAbort; diff --git a/engine/crates/layerfs-daemon/src/control/operations.rs b/engine/crates/layerfs-daemon/src/control/operations.rs index 3b44d8105..3944f6e5c 100644 --- a/engine/crates/layerfs-daemon/src/control/operations.rs +++ b/engine/crates/layerfs-daemon/src/control/operations.rs @@ -189,6 +189,7 @@ impl Service { commit: None, observation_failure: None, kernel: None, + abort: None, }; let installed = self.entries.lock().map(|mut entries| { entries.insert( diff --git a/engine/crates/layerfs-daemon/src/control/registry.rs b/engine/crates/layerfs-daemon/src/control/registry.rs index f21131caa..1852050b0 100644 --- a/engine/crates/layerfs-daemon/src/control/registry.rs +++ b/engine/crates/layerfs-daemon/src/control/registry.rs @@ -1,7 +1,7 @@ //! Short control admission over bounded daemon-owned Workspace routes. use super::{ kernel::{Kernel, KernelServing}, - Failure, + Failure, UnmountAbort, }; use crate::{ store::{BoundWorkspace, Store}, @@ -68,6 +68,17 @@ impl Service { }; workspace.operation().map_err(Failure::Workspace) } + /// The abort path a retained normal unmount recorded, if it took one. + /// Observation only: nothing is transferred, released, retried or settled. + pub fn retained_abort(&self, token: WorkspaceToken) -> Option> { + let entries = self.entries.lock().ok()?; + let binding = bound(&entries, token).ok()?; + match &binding.kernel { + #[cfg(target_os = "linux")] + Kernel::Retained(kept) => kept.abort.clone(), + _ => None, + } + } /// Workspaces bound or binding; None when the registry is poisoned. pub fn workspaces(&self) -> Option { self.entries.lock().ok().map(|entries| entries.len()) diff --git a/engine/crates/layerfs-daemon/src/control/serving.rs b/engine/crates/layerfs-daemon/src/control/serving.rs index 8a936eba6..89bbaed9c 100644 --- a/engine/crates/layerfs-daemon/src/control/serving.rs +++ b/engine/crates/layerfs-daemon/src/control/serving.rs @@ -1,5 +1,8 @@ //! One shared dispatcher for every mount, and the exact owners an entry keeps. -use super::kernel::{bounded, KernelConfig}; +use super::{ + kernel::{bounded, KernelConfig}, + UnmountAbort, +}; use layerfs_bridge::control::{ ForcedFacts, MountPhase, MountReceipt, MountWork, ReadyMount, TeardownCustody, TeardownStage, WorkspaceToken, @@ -49,6 +52,16 @@ pub(super) struct Leaving { pub observer: SessionObserver, pub phase: MountPhase, } +/// One terminal stop's inputs: the boundary, its bounded detail, the owner +/// observation and the forced and abort facts kept beside it. `keep` fills in +/// the mount receipt from the entry. +pub(super) struct Stop { + pub stage: TeardownStage, + pub detail: String, + pub observer: Option, + pub forced: Option, + pub abort: Option>, +} /// Custody that a terminal or attach operation could not finish disposing. /// Nothing here is dropped, retried or settled by a later observation. pub(super) struct Kept { @@ -61,6 +74,9 @@ pub(super) struct Kept { /// Abort effects as recorded where teardown stopped, including a normal /// unmount's stalled connection. Absent when no abort was attempted. pub forced: Option, + /// The normal unmount's abort path when it ran, exactly as it stood there, + /// for the retained entry's observation. Absent when no abort was made. + pub abort: Option>, } impl fmt::Debug for Kept { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -71,6 +87,7 @@ impl fmt::Debug for Kept { .field("observer", &self.observer) .field("evidence", &self.evidence) .field("forced", &self.forced) + .field("abort", &self.abort) .finish() } } diff --git a/engine/crates/layerfs-daemon/src/control/status.rs b/engine/crates/layerfs-daemon/src/control/status.rs index 9b86d5249..2c31eae45 100644 --- a/engine/crates/layerfs-daemon/src/control/status.rs +++ b/engine/crates/layerfs-daemon/src/control/status.rs @@ -84,6 +84,7 @@ impl Service { commit: None, observation_failure, kernel: None, + abort: None, }) } } diff --git a/engine/crates/layerfs-daemon/src/control/types.rs b/engine/crates/layerfs-daemon/src/control/types.rs index 320b574fd..bf44476d9 100644 --- a/engine/crates/layerfs-daemon/src/control/types.rs +++ b/engine/crates/layerfs-daemon/src/control/types.rs @@ -22,6 +22,10 @@ pub struct Success { pub observation_failure: Option>, /// Original kernel connection receipt of a terminal unmount. pub kernel: Option>, + /// The normal unmount's abort path, when its stalled drain took it. A + /// caller-visible observation only: it is encoded in no reply field and + /// authorizes no retry or second abort. + pub abort: Option>, } /// A kernel attach or unmount refusal, or a Mount refused while maintenance /// is stopped, with its original evidence. @@ -61,6 +65,7 @@ impl Success { commit: None, observation_failure: None, kernel: None, + abort: None, } } } diff --git a/engine/crates/layerfs-daemon/src/control/unmount.rs b/engine/crates/layerfs-daemon/src/control/unmount.rs index 062b26a19..c37e63aa2 100644 --- a/engine/crates/layerfs-daemon/src/control/unmount.rs +++ b/engine/crates/layerfs-daemon/src/control/unmount.rs @@ -2,7 +2,8 @@ use super::{ kernel::Kernel, registry::{bound_mut, idle}, - Failure, KernelEvidence, Service, Success, + serving::Stop, + Failure, KernelEvidence, Service, Success, UnmountAbort, }; use crate::store::BoundWorkspace; use layerfs_bridge::control::{Activity, ForcedFacts, Reply, UnmountControl, WorkspaceToken}; @@ -35,7 +36,7 @@ impl Service { return self.unmount_kernel(token, workspace, *attached) } }; - self.close(token, &workspace, None) + self.close(token, &workspace, None, None) } /// One short registry step: no effect unless the entry is idle and its /// kernel half is either absent or a serving connection with no request. @@ -77,13 +78,14 @@ impl Service { token: WorkspaceToken, workspace: &BoundWorkspace, kernel: Receipt, + abort: Option>, ) -> Result { #[cfg(target_os = "linux")] let forced = super::detach::detached_abort_facts(kernel.as_ref().and_then(|done| done.forced)); #[cfg(not(target_os = "linux"))] let forced = None; - let kernel = self.closed(token, workspace, kernel, forced)?; + let kernel = self.closed(token, workspace, kernel, forced, abort.clone())?; self.depart( token, Success { @@ -91,6 +93,7 @@ impl Service { commit: None, observation_failure: None, kernel, + abort, }, ) } @@ -102,6 +105,7 @@ impl Service { workspace: &BoundWorkspace, kernel: Receipt, forced: Option, + abort: Option>, ) -> Result { match workspace.close() { Ok(()) => Ok(kernel), @@ -112,7 +116,7 @@ impl Service { layerfs_workspace::WorkspaceError::Service(Box::new(cause)), ), }; - Err(self.unclosed(token, failure, kernel, forced)) + Err(self.unclosed(token, failure, kernel, forced, abort)) } } } @@ -149,20 +153,29 @@ impl Service { failure: Failure, kernel: Receipt, forced: Option, + mut abort: Option>, ) -> Failure { #[cfg(target_os = "linux")] if kernel.is_some() { + // Close stopped with the abort path already recorded: the retained + // observation names this boundary, not the drain that preceded it. + if let Some(abort) = abort.as_mut() { + abort.stage = layerfs_bridge::control::TeardownStage::Close; + } return self.retain( token, - layerfs_bridge::control::TeardownStage::Close, - failure.wire().detail, - None, + Stop { + stage: layerfs_bridge::control::TeardownStage::Close, + detail: failure.wire().detail, + observer: None, + forced, + abort, + }, Box::new((failure, kernel)), - forced, ); } #[cfg(not(target_os = "linux"))] - let _ = (kernel, forced); + let _ = (kernel, forced, abort); if let Ok(mut entries) = self.entries.lock() { if let Ok(binding) = bound_mut(&mut entries, token) { binding.activity = if failure.uncertain() { diff --git a/engine/crates/layerfs-daemon/src/control/unmount_abort.rs b/engine/crates/layerfs-daemon/src/control/unmount_abort.rs new file mode 100644 index 000000000..1ae1fd311 --- /dev/null +++ b/engine/crates/layerfs-daemon/src/control/unmount_abort.rs @@ -0,0 +1,39 @@ +//! What one normal unmount observed when its drain stalled at `Join` and it +//! took the abort path (issues #426, #460; observation added for #2). +//! Observation only: it is built where that path ends, is encoded in no reply +//! field, and carries no authority to retry, abort again or release an owner. + +#[cfg(target_os = "linux")] +use layerfs_bridge::control::{MountWork, TeardownStage, WorkspaceToken}; +#[cfg(target_os = "linux")] +use layerfs_fuse::mount::AbortWrite; +#[cfg(target_os = "linux")] +use std::time::Duration; + +/// The abort path of one normal unmount, as it stood when the unmount was +/// completed or stopped. The same facts reach a traced caller as the +/// `unmount` telemetry section and the daemon's one stderr line. +#[cfg(target_os = "linux")] +#[derive(Clone, Debug)] +pub struct UnmountAbort { + /// The exact incarnation whose connection was aborted. + pub token: WorkspaceToken, + /// The boundary the unmount was at: `Join` when the one abort write ended + /// a stalled drain, `Abort` when that write was short or failed, and + /// `Revoke` or `Close` when a later step stopped with it recorded. + pub stage: TeardownStage, + /// The connection's single abort write and its original result. + pub abort: AbortWrite, + /// Kernel detach was already known when the abort was made. + pub detached: bool, + /// Loop, lane and request counters observed at the stop or completion. + pub work: MountWork, + /// Time from the stalled drain to the end of the abort path. + pub waited: Duration, +} + +/// Kernel sessions exist only on Linux, so no other platform reaches the +/// abort path and none carries an observation. +#[cfg(not(target_os = "linux"))] +#[derive(Clone, Debug)] +pub enum UnmountAbort {} diff --git a/engine/crates/layerfs-daemon/src/telemetry.rs b/engine/crates/layerfs-daemon/src/telemetry.rs index 31d876ac8..09e82840c 100644 --- a/engine/crates/layerfs-daemon/src/telemetry.rs +++ b/engine/crates/layerfs-daemon/src/telemetry.rs @@ -1,7 +1,7 @@ //! The daemon's half of a traced reply: Commit storage and construction //! receipts and kernel opcode counts. Phase timing and SQL work are the Store //! service's ([`layerfs_store_service::telemetry`]). -use crate::control::{Failure, Success}; +use crate::control::{Failure, Success, UnmountAbort}; use layerfs_bridge::control::SdkValue; pub(crate) use layerfs_store_service::telemetry::{delta, sql, sql_values}; use layerfs_store_service::telemetry::{named, timing}; @@ -9,19 +9,26 @@ pub use layerfs_store_service::telemetry::{phase, recording, timed, Recording}; use layerfs_telemetry::timer::TimingReport; /// The telemetry of one traced request: its timing, its Store SQL work, the -/// Workspace's kernel opcode counts when given, the Commit's storage -/// diagnostics and construction receipt when it committed, and the failed -/// diagnostic write, if any. +/// Workspace's kernel opcode counts when given, the normal unmount's abort +/// path when it took one, the Commit's storage diagnostics and construction +/// receipt when it committed, and the failed diagnostic write, if any. pub(crate) fn value( reports: Vec, outcome: &Result, diagnostic: Option<&std::io::Error>, [sql, delta, fuse]: [Option; 3], + abort: Option>, ) -> SdkValue { let mut fields = vec![("timing".to_owned(), timing(&reports))]; for (name, value) in [("sql", sql), ("delta", delta), ("fuse", fuse)] { fields.extend(value.map(|value| (name.to_owned(), value))); } + #[cfg(target_os = "linux")] + if let Some(abort) = abort { + fields.push(("unmount".to_owned(), unmount_abort(&abort))); + } + #[cfg(not(target_os = "linux"))] + let _ = abort; let (namespace, storage) = crate::application::commit_parts(outcome); if let Some(storage) = storage { fields.push(( @@ -47,6 +54,103 @@ pub(crate) fn value( SdkValue::Map(fields) } +/// The `unmount` telemetry section of a normal unmount that took the abort +/// path (issue #2), on success and on a stop. Its stable keys, in order, are +/// `stage`, `abort`, `bytes` when the write was short or `errno` when it +/// failed, `detached`, `loops_exited`, `loops_joined`, `active`, `retained` +/// and `waited_ms`. +#[cfg(target_os = "linux")] +pub fn unmount_abort(abort: &UnmountAbort) -> SdkValue { + SdkValue::Map(unmount_abort_fields(abort)) +} + +/// The same facts as one `LAYERFS_DAEMON_UNMOUNT_ABORT` stderr line. The +/// daemon writes it once per abort path, on success and on a stop; it is not +/// per-request logging. +#[cfg(target_os = "linux")] +pub fn unmount_abort_line(abort: &UnmountAbort) -> String { + let mut line = format!( + "LAYERFS_DAEMON_UNMOUNT_ABORT workspace={} namespace={}", + hex(abort.token.workspace.as_slice()), + abort.token.namespace + ); + for (name, value) in unmount_abort_fields(abort) { + line.push(' '); + line.push_str(&name); + line.push('='); + match value { + SdkValue::Bool(flag) => line.push_str(if flag { "true" } else { "false" }), + SdkValue::Int(number) => line.push_str(&number.to_string()), + SdkValue::Text(text) => line.push_str(&text), + _ => {} + } + } + line +} + +/// The observation's fields, in one order for both forms. +#[cfg(target_os = "linux")] +fn unmount_abort_fields(abort: &UnmountAbort) -> Vec<(String, SdkValue)> { + use layerfs_fuse::mount::AbortWrite; + let named = |name: &str, value: SdkValue| (name.to_owned(), value); + let mut fields = vec![ + named("stage", SdkValue::text(format!("{:?}", abort.stage))), + named("abort", SdkValue::text(abort_disposition(abort.abort))), + ]; + match abort.abort { + AbortWrite::Short(bytes) => fields.push(named( + "bytes", + SdkValue::Int(i64::try_from(bytes).unwrap_or(i64::MAX)), + )), + AbortWrite::Failed(errno) => { + fields.push(named("errno", SdkValue::text(format!("{errno:?}")))) + } + AbortWrite::Written => {} + } + fields.extend([ + named("detached", SdkValue::Bool(abort.detached)), + named( + "loops_exited", + SdkValue::Int(i64::from(abort.work.loops_exited)), + ), + named( + "loops_joined", + SdkValue::Int(i64::from(abort.work.loops_joined)), + ), + named("active", SdkValue::Int(i64::from(abort.work.active))), + named("retained", SdkValue::Int(i64::from(abort.work.retained))), + named("waited_ms", SdkValue::Int(waited_ms(abort.waited))), + ]); + fields +} + +#[cfg(target_os = "linux")] +fn abort_disposition(write: layerfs_fuse::mount::AbortWrite) -> &'static str { + use layerfs_fuse::mount::AbortWrite; + match write { + AbortWrite::Written => "Written", + AbortWrite::Short(_) => "Short", + AbortWrite::Failed(_) => "Failed", + } +} + +/// Milliseconds, saturated; the observation is diagnostic, never a budget. +#[cfg(target_os = "linux")] +fn waited_ms(waited: std::time::Duration) -> i64 { + i64::try_from(waited.as_millis()).unwrap_or(i64::MAX) +} + +#[cfg(target_os = "linux")] +fn hex(bytes: &[u8]) -> String { + use std::fmt::Write as _; + bytes + .iter() + .fold(String::with_capacity(bytes.len() * 2), |mut text, byte| { + let _ = write!(text, "{byte:02x}"); + text + }) +} + /// The Workspace's kernel opcode counts, frame dispositions, largest request /// sizes and per-opcode callback nanoseconds (`ns`) since attach. #[cfg(target_os = "linux")] diff --git a/engine/crates/layerfs-daemon/tests/stalled_unmount.rs b/engine/crates/layerfs-daemon/tests/stalled_unmount.rs index 6a74af6f4..3ee00cf0a 100644 --- a/engine/crates/layerfs-daemon/tests/stalled_unmount.rs +++ b/engine/crates/layerfs-daemon/tests/stalled_unmount.rs @@ -152,6 +152,20 @@ fn a_normal_unmount_stalled_at_join_ends_itself_with_one_abort_write() { forced.detach, layerfs_fuse::session::DetachAttempt::Detached ); + // The same abort path is attached to the successful reply (#2). + let abort = done.abort.as_deref().expect("abort-path observation"); + assert_eq!(abort.stage, TeardownStage::Join); + assert_eq!(abort.abort, layerfs_fuse::mount::AbortWrite::Written); + assert!(abort.detached); + assert!(abort.work.loops_configured > 1, "{:?}", abort.work); + assert_eq!(abort.work.loops_exited, abort.work.loops_joined); + assert_eq!(abort.work.loops_joined, abort.work.loops_configured); + assert_eq!((abort.work.active, abort.work.retained), (0, 0)); + assert!( + abort.waited <= Duration::from_secs(60), + "{:?}", + abort.waited + ); assert!( mount_entry(&ready.directory).is_none(), "the daemon's mount row remains" @@ -212,6 +226,12 @@ fn rejected_abort(errno: Option, disposition: AbortDisposition, detail: &st let again = rig.harness.try_unmount(token); let forced = rig.harness.try_force(token, false); let evidence = rig.harness.service.retained_kernel(token).unwrap().unwrap(); + // The retained entry keeps the observation for a later reply (#2). + let observed = rig + .harness + .service + .retained_abort(token) + .expect("retained abort observation"); assert!(copy.leave()); drop(copy); until("retained connection loops joined", || { @@ -235,6 +255,17 @@ fn rejected_abort(errno: Option, disposition: AbortDisposition, detail: &st assert_eq!(facts.fenced, 0); assert!(custody.detail.contains(detail), "{custody:?}"); assert!(evidence.contains("forced: Some(Forced"), "{evidence}"); + assert_eq!(observed.stage, stage); + assert!(observed.detached); + assert_eq!( + observed.abort, + match errno { + Some(0) => layerfs_fuse::mount::AbortWrite::Short(0), + Some(errno) => + layerfs_fuse::mount::AbortWrite::Failed(nix::errno::Errno::from_raw(errno as i32)), + None => layerfs_fuse::mount::AbortWrite::Written, + } + ); for result in [again, forced] { let Err(Failure::Retained(later)) = result else { panic!("retained custody changed: {result:?}"); @@ -289,6 +320,11 @@ fn later_failure(stage: TeardownStage) { } drop(operation); let result = rig.harness.try_unmount(token); + let observed = rig + .harness + .service + .retained_abort(token) + .expect("retained abort observation"); let again = rig.harness.try_force(token, false); assert!(copy.leave()); drop(copy); @@ -301,6 +337,10 @@ fn later_failure(stage: TeardownStage) { assert_eq!(custody.stage, stage, "{custody:?}"); let facts = custody.forced.as_ref().expect("successful abort facts"); assert_eq!(facts.abort, AbortDisposition::Written); + // A stop after the abort keeps the observation and its later boundary. + assert_eq!(observed.stage, stage); + assert!(observed.detached); + assert_eq!(observed.abort, layerfs_fuse::mount::AbortWrite::Written); assert_eq!(facts.detach, DetachDisposition::Detached); assert_eq!(facts.commit, CommitKnowledge::Absent); let Err(Failure::Retained(later)) = again else { diff --git a/engine/crates/layerfs-daemon/tests/unmount_observability.rs b/engine/crates/layerfs-daemon/tests/unmount_observability.rs new file mode 100644 index 000000000..efa0eba42 --- /dev/null +++ b/engine/crates/layerfs-daemon/tests/unmount_observability.rs @@ -0,0 +1,134 @@ +//! The normal unmount's abort path as an observation (issue #2): the one +//! telemetry section a traced reply carries, the one stderr line, and the +//! silence of an unmount that never took that path. +//! +//! The abort path itself is staged by `stalled_unmount`; these tests pin the +//! stable key set both forms render and the absence of any observation when +//! the drain never stalled. +#![cfg(target_os = "linux")] +#[allow(dead_code)] +#[path = "support/installed_store.rs"] +mod installed; +#[allow(dead_code)] +#[path = "support/namespace_model.rs"] +mod model; +#[allow(dead_code)] +#[path = "support/mounted.rs"] +mod mounted; +#[allow(dead_code)] +#[path = "support/mounted_commit.rs"] +mod rig; +use layerfs_bridge::control::{MountWork, SdkValue, TeardownStage, WorkspaceToken}; +use layerfs_daemon::{control::UnmountAbort, telemetry}; +use layerfs_fuse::mount::AbortWrite; +use layerfs_history::WorkspaceId; +use nix::errno::Errno; +use rig::Rig; +use std::time::Duration; + +/// One observation of a connection that stalled at `Join` and was aborted. +fn observation(abort: AbortWrite) -> UnmountAbort { + UnmountAbort { + token: WorkspaceToken { + workspace: WorkspaceId::from_authority([9; 32]).unwrap(), + namespace: 3, + }, + stage: TeardownStage::Join, + abort, + detached: true, + work: MountWork { + loops_configured: 4, + loops_entered: 4, + loops_exited: 4, + loops_joined: 4, + active: 0, + retained: 0, + completed: 3, + ..MountWork::default() + }, + waited: Duration::from_millis(7), + } +} + +/// The section's field names, in the order the daemon writes them. +fn names(section: &SdkValue) -> Vec { + match section { + SdkValue::Map(fields) => fields.iter().map(|(name, _)| name.clone()).collect(), + other => panic!("not a section: {other:?}"), + } +} + +#[test] +fn the_abort_section_keeps_one_key_set_for_every_abort_result() { + let written = telemetry::unmount_abort(&observation(AbortWrite::Written)); + assert_eq!( + names(&written), + [ + "stage", + "abort", + "detached", + "loops_exited", + "loops_joined", + "active", + "retained", + "waited_ms" + ] + ); + assert_eq!(written.get_text("stage"), Some("Join")); + assert_eq!(written.get_text("abort"), Some("Written")); + assert!(written.get_bool("detached")); + assert_eq!(written.get_int("loops_exited"), Some(4)); + assert_eq!(written.get_int("loops_joined"), Some(4)); + assert_eq!(written.get_int("active"), Some(0)); + assert_eq!(written.get_int("retained"), Some(0)); + assert_eq!(written.get_int("waited_ms"), Some(7)); + assert!(written.get("bytes").is_none() && written.get("errno").is_none()); + + // A short write names its count, a failed one its errno, and only then. + let short = telemetry::unmount_abort(&observation(AbortWrite::Short(0))); + assert_eq!(short.get_text("abort"), Some("Short")); + assert_eq!(short.get_int("bytes"), Some(0)); + assert!(short.get("errno").is_none()); + let failed = telemetry::unmount_abort(&observation(AbortWrite::Failed(Errno::EIO))); + assert_eq!(failed.get_text("abort"), Some("Failed")); + assert_eq!(failed.get_text("errno"), Some("EIO")); + assert!(failed.get("bytes").is_none()); +} + +#[test] +fn the_stderr_line_carries_the_same_fields_as_the_section() { + let line = telemetry::unmount_abort_line(&observation(AbortWrite::Written)); + assert!( + line.starts_with("LAYERFS_DAEMON_UNMOUNT_ABORT workspace=09090909"), + "{line}" + ); + assert!( + line.ends_with( + "namespace=3 stage=Join abort=Written detached=true loops_exited=4 loops_joined=4 active=0 retained=0 waited_ms=7" + ), + "{line}" + ); + let failed = telemetry::unmount_abort_line(&observation(AbortWrite::Failed(Errno::EIO))); + assert!( + failed.contains(" abort=Failed errno=EIO detached=true "), + "{failed}" + ); + let short = telemetry::unmount_abort_line(&observation(AbortWrite::Short(0))); + assert!( + short.contains(" abort=Short bytes=0 detached=true "), + "{short}" + ); +} + +#[test] +fn an_unmount_that_never_stalled_reports_no_abort_path() { + let rig = Rig::new("unmount-observability"); + let ready = rig.mount(1); + let done = rig.harness.unmount(&ready); + assert!(done.abort.is_none(), "{:?}", done.abort); + assert!( + rig.harness.service.retained_abort(ready.token).is_none(), + "a completed unmount retains no abort observation" + ); + rig.finish(); +} diff --git a/tools/ci_tests.py b/tools/ci_tests.py index ac284bf54..a92a686a2 100644 --- a/tools/ci_tests.py +++ b/tools/ci_tests.py @@ -36,7 +36,16 @@ MANIFEST = ROOT / "Cargo.toml" LINUX_ONLY = '#![cfg(target_os = "linux")]' MOUNT_PACKAGE = "layerfs-daemon" -API_MOUNT_TARGETS = {"kernel_mount_routes", "kernel_application", "observed_application", "r5_1_actual_daemon_commit"} +# SDK mount targets mount a real kernel Workspace: the fast lane leaves them out +# like the daemon ones, and the mount lane runs each in its own container. +API_MOUNT_TARGETS = { + "kernel_mount_routes", + "kernel_application", + "observed_application", + "r5_1_actual_daemon_commit", + # Binds and unmounts a Workspace through the actual daemon (#2). + "unmount_observability", +} # (package, test target): run seconds on the macOS host, 2026-10-23. SLOW = { diff --git a/tools/test_ci_tests.py b/tools/test_ci_tests.py index 21c4c7a8e..9714c3048 100644 --- a/tools/test_ci_tests.py +++ b/tools/test_ci_tests.py @@ -35,6 +35,7 @@ def test_mount_group_includes_moved_sdk_targets(self): self.assertNotIn("lane_merge", names) self.assertIn(("layerfs-api", "kernel_application"), mount) self.assertIn(("layerfs-api", "observed_application"), mount) + self.assertIn(("layerfs-api", "unmount_observability"), mount) self.assertIn(("layerfs-daemon", "forced_unavailable"), mount) def test_fast_filter_excludes_the_nightly_groups_and_nightly_selects_them(self):