diff --git a/crates/sandlock-core/src/cow/seccomp.rs b/crates/sandlock-core/src/cow/seccomp.rs index f6f71372..b709cf94 100644 --- a/crates/sandlock-core/src/cow/seccomp.rs +++ b/crates/sandlock-core/src/cow/seccomp.rs @@ -103,12 +103,15 @@ impl SeccompCowBranch { let branch_dir = storage_base.join(&branch_id); let upper = branch_dir.join("upper"); - fs::create_dir_all(&upper) - .map_err(|e| BranchError::Operation(format!("create upper: {}", e)))?; - + // Canonicalize the workdir BEFORE creating the branch dir, so a failure + // here (e.g. the workdir was removed between validation and now) can't + // orphan an empty branch/upper dir on disk. let workdir = workdir.canonicalize() .map_err(|e| BranchError::Operation(format!("canonicalize workdir: {}", e)))?; + fs::create_dir_all(&upper) + .map_err(|e| BranchError::Operation(format!("create upper: {}", e)))?; + Ok(Self { workdir_str: workdir.to_string_lossy().into_owned(), workdir, @@ -883,16 +886,64 @@ impl SeccompCowBranch { Ok(()) } + /// Mark the branch as intentionally kept: its upper is left on disk and the + /// `Drop` backstop below will not clean it up. Used for `BranchAction::Keep`, + /// which preserves the changes for later inspection rather than merging or + /// discarding them. + pub(crate) fn keep(&mut self) { + self.finished = true; + } + fn cleanup(&self) { let _ = fs::remove_dir_all(&self.storage_dir); } } +impl Drop for SeccompCowBranch { + /// Backstop against orphaning the upper on disk. A branch dropped without an + /// explicit disposition — an error/timeout/panic in a transactional pipeline, + /// an abandoned sandbox that was never `wait()`ed — removes its own private + /// storage dir. `commit()`, `abort()`, and `keep()` all set `finished`, so + /// this is a no-op on every deliberate path (in particular it does NOT clean + /// a kept branch). `remove_dir_all` is idempotent and scoped to `storage_dir`. + fn drop(&mut self) { + if !self.finished { + self.cleanup(); + } + } +} + #[cfg(test)] mod tests { use super::*; use std::fs; + #[test] + fn drop_cleans_undisposed_branch_but_keep_preserves() { + let workdir = tempfile::tempdir().unwrap(); + + // A branch dropped without commit/abort/keep must remove its private + // storage dir — otherwise a failed/aborted-by-error transaction orphans + // the upper on disk (the leak this Drop backstop closes). + let leaked = { + let branch = SeccompCowBranch::create(workdir.path(), None, 0).unwrap(); + let dir = branch.storage_dir.clone(); + assert!(dir.exists()); + dir + }; + assert!(!leaked.exists(), "an undisposed branch must clean its storage on drop"); + + // keep() marks the branch finished, so Drop preserves the upper. + let kept = { + let mut branch = SeccompCowBranch::create(workdir.path(), None, 0).unwrap(); + let dir = branch.storage_dir.clone(); + branch.keep(); + dir + }; + assert!(kept.exists(), "a kept branch must survive drop"); + let _ = fs::remove_dir_all(&kept); + } + fn setup_workdir() -> (tempfile::TempDir, tempfile::TempDir) { let workdir = tempfile::tempdir().unwrap(); let storage = tempfile::tempdir().unwrap(); diff --git a/crates/sandlock-core/src/lib.rs b/crates/sandlock-core/src/lib.rs index 2cfd68db..b4fe6fc7 100644 --- a/crates/sandlock-core/src/lib.rs +++ b/crates/sandlock-core/src/lib.rs @@ -40,7 +40,7 @@ pub use sandbox::{ BindPorts, Confinement, ConfinementBuilder, Process, Sandbox, SandboxBuilder, StdioMode, }; pub use result::{RunResult, ExitStatus}; -pub use pipeline::{Stage, Pipeline, Gather}; +pub use pipeline::{Stage, Pipeline, Gather, TxnOutcome}; pub use dry_run::{Change, ChangeKind, DryRunResult}; // Sectioned-profile parsing types: ProfileInput is the top-level deserialization // target; ProgramSpec carries [program].exec/args (not a Sandbox field). diff --git a/crates/sandlock-core/src/pipeline.rs b/crates/sandlock-core/src/pipeline.rs index 0f30fc2a..460f5142 100644 --- a/crates/sandlock-core/src/pipeline.rs +++ b/crates/sandlock-core/src/pipeline.rs @@ -108,6 +108,37 @@ impl Pipeline { run_pipeline(self.stages).await } } + + /// Run the pipeline as a filesystem transaction over a shared COW workdir. + /// + /// Unlike [`run`](Self::run) (concurrent stages streaming bytes stdout→stdin), + /// stages run **sequentially** and share **one** COW upper over a common + /// workdir: stage N+1 sees stage N's writes (read-committed), the real + /// workdir is untouched during the run, and at the end either every stage's + /// writes commit together (all stages exited 0) or none do (any stage failed + /// → the shared upper is discarded, the workdir is byte-identical). Data is + /// exchanged through the shared workspace, not inter-stage pipes. + /// + /// Every stage must set the same `workdir`, run with the supervisor + /// (`no_supervisor == false`), leave `on_exit`/`on_error` at their defaults, + /// and set no `chroot` and the same `fs_storage`/`max_disk` — the pipeline + /// owns the single shared upper and its commit/abort, so a per-stage override + /// would conflict and is rejected. + /// + /// `timeout` applies to the whole pipeline; on timeout the transaction aborts. + /// + /// The final commit is **not crash-atomic**: it merges the shared upper into + /// the workdir file-by-file, so a crash (or `ENOSPC`) *mid-commit* can leave + /// the workdir partially merged. The all-or-nothing guarantee holds for a + /// clean stage failure or timeout (nothing is written until the commit + /// starts); durable crash-atomic commit is a later phase. + pub async fn run_transactional( + self, + timeout: Option, + ) -> Result { + validate_txn_stages(&self.stages)?; + run_pipeline_txn(self.stages, timeout).await + } } impl std::ops::BitOr for Pipeline { @@ -118,6 +149,222 @@ impl std::ops::BitOr for Pipeline { } } +// ============================================================ +// Transactional pipeline +// ============================================================ + +/// Outcome of [`Pipeline::run_transactional`]. +#[derive(Debug, Clone)] +pub struct TxnOutcome { + /// True if every stage exited 0 and the shared upper was committed to the + /// workdir. False if any stage failed (or the pipeline timed out) and the + /// upper was discarded, leaving the workdir byte-identical. + pub committed: bool, + /// Per-stage results in execution order. On a stage-failure abort this holds + /// the stages that ran, up to and including the one that failed. On a timeout + /// or driver-error abort it is empty: the in-flight stage-driver future is + /// cancelled and its accumulated results are dropped (`committed` and + /// `abort_reason` still report the outcome). + pub stages: Vec, + /// Human-readable reason the transaction aborted; `None` when committed. + pub abort_reason: Option, +} + +/// Reject stage configurations that can't participate in a transaction. The +/// pipeline owns the single shared upper and the single commit/abort, so each +/// stage must set the same workdir, keep the supervisor, and not carry its own +/// branch action. +fn validate_txn_stages(stages: &[Stage]) -> Result<(), SandlockError> { + fn reject(msg: impl Into) -> SandlockError { + SandlockError::Runtime(SandboxRuntimeError::Child(msg.into())) + } + + // `Pipeline::new` enforces >= 2, but the struct is constructible directly; + // check here so `run_transactional` never indexes `stages[0]` out of bounds. + if stages.len() < 2 { + return Err(reject("transactional pipeline requires at least 2 stages")); + } + + let base = &stages[0].sandbox; + let base_wd = base.workdir.as_ref().ok_or_else(|| { + reject("transactional pipeline: stage 0 has no workdir; every stage must set the shared transaction workdir") + })?; + let base_max_disk = base.max_disk.map(|b| b.0).unwrap_or(0); + + for (i, stage) in stages.iter().enumerate() { + let sb = &stage.sandbox; + let wd = sb.workdir.as_ref().ok_or_else(|| { + reject(format!("transactional pipeline: stage {i} has no workdir; every stage must set the shared transaction workdir")) + })?; + if wd != base_wd { + return Err(reject(format!( + "transactional pipeline: stages must share one workdir (stage 0 = {}, stage {i} = {})", + base_wd.display(), + wd.display() + ))); + } + if sb.no_supervisor { + return Err(reject(format!( + "transactional pipeline: stage {i} has no_supervisor=true; transactions require the COW supervisor" + ))); + } + // All stages overlay ONE shared upper, so per-stage COW storage/quota and + // chroot can't each take effect — reject a stage that sets them differently + // (or at all, for chroot) rather than silently using only stage 0's. + if sb.chroot.is_some() { + return Err(reject(format!( + "transactional pipeline: stage {i} sets chroot, which is unsupported with a shared COW workdir" + ))); + } + if sb.fs_storage != base.fs_storage || sb.max_disk.map(|b| b.0).unwrap_or(0) != base_max_disk { + return Err(reject(format!( + "transactional pipeline: stage {i} sets a different fs_storage/max_disk; all stages share one COW upper, so these must match stage 0" + ))); + } + // The builder leaves both actions at `BranchAction::Commit` by default + // (`unwrap_or_default()`); anything else is an explicit per-stage choice + // that conflicts with the pipeline owning commit/abort. + if sb.on_exit != crate::sandbox::BranchAction::Commit + || sb.on_error != crate::sandbox::BranchAction::Commit + { + return Err(reject(format!( + "transactional pipeline: stage {i} sets on_exit/on_error, which conflicts with a transactional pipeline (the pipeline owns commit/abort) — leave them at their defaults" + ))); + } + } + Ok(()) +} + +/// Create the shared COW branch, drive the stages sequentially over it, then +/// commit-all or abort-all. The branch lives here (outside the driven future) +/// so a timeout that cancels the stage loop can still abort cleanly. +async fn run_pipeline_txn( + stages: Vec, + timeout: Option, +) -> Result { + fn child_err(msg: String) -> SandlockError { + SandlockError::Runtime(SandboxRuntimeError::Child(msg)) + } + + // All stages share the validated workdir; take COW storage/quota from the + // first stage (they overlay the same lower). + let base = &stages[0].sandbox; + let workdir = base.workdir.clone().expect("validated: stage 0 has a workdir"); + let storage = base.fs_storage.clone(); + let max_disk = base.max_disk.map(|b| b.0).unwrap_or(0); + + let branch = crate::cow::seccomp::SeccompCowBranch::create(&workdir, storage.as_deref(), max_disk) + .map_err(|e| child_err(format!("transactional pipeline: failed to create COW branch: {e}")))?; + let upper_dir = branch.upper_dir().to_path_buf(); + + let mut cow_state = crate::seccomp::state::CowState::new(); + cow_state.branch = Some(branch); + let state = std::sync::Arc::new(tokio::sync::Mutex::new(cow_state)); + let shared = crate::sandbox::SharedCow { state: std::sync::Arc::clone(&state), upper_dir }; + + let drive = drive_txn_stages(stages, shared); + let driven: Result<(bool, Vec, Option), SandlockError> = match timeout { + Some(dur) => match tokio::time::timeout(dur, drive).await { + Ok(r) => r, + Err(_) => Ok((false, Vec::new(), Some("transactional pipeline timed out".to_string()))), + }, + None => drive.await, + }; + + // Finalize the shared upper in EVERY case — commit only on a clean full run, + // otherwise discard — before propagating any driver error, so a mid-loop + // failure never leaves the upper dangling. (`SeccompCowBranch`'s Drop is a + // further backstop for panics/cancellation; this is the deterministic path.) + // Take the branch out from under the async mutex first, then commit/abort the + // owned value so the sync merge doesn't run while holding the guard. + let taken = { state.lock().await.branch.take() }; + let (all_ok, results, mut reason, drive_err) = match driven { + Ok((ok, res, rsn)) => (ok, res, rsn, None), + Err(e) => (false, Vec::new(), Some(format!("{e}")), Some(e)), + }; + let committed = match taken { + Some(mut branch) if all_ok => { + // First-committer-win: take an exclusive, non-blocking lock on the + // shared workdir for the duration of the merge. commit() rewrites the + // workdir file-by-file, so two concurrent transactional pipelines + // committing into one workdir would interleave and corrupt it. A + // pipeline that finds the lock held LOSES the race — a benign + // non-commit outcome (committed=false + abort_reason), not an error, + // and its upper is aborted. (The lock is scoped to the transactional + // commit here, not the single-sandbox commit() path.) + let lock = std::fs::File::open(&workdir).map_err(|e| { + child_err(format!("transactional pipeline: open workdir for commit lock: {e}")) + })?; + if unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 { + let err = std::io::Error::last_os_error(); + // EWOULDBLOCK (== EAGAIN on Linux) means another commit holds the + // lock: lose the race benignly. Any other errno is a real failure. + if err.raw_os_error() == Some(libc::EWOULDBLOCK) { + let _ = branch.abort(); + reason = Some( + "transactional pipeline: lost the first-committer-win race \ + (another commit is in progress on this workdir)" + .to_string(), + ); + false + } else { + return Err(child_err(format!( + "transactional pipeline: commit lock: {err}" + ))); + } + } else { + branch.commit().map_err(|e| { + child_err(format!("transactional pipeline: commit failed: {e}")) + })?; + drop(lock); // release the workdir lock after the merge + true + } + } + Some(mut branch) => { + let _ = branch.abort(); + false + } + None => all_ok, + }; + if let Some(e) = drive_err { + return Err(e); + } + + Ok(TxnOutcome { + committed, + stages: results, + abort_reason: if committed { None } else { reason }, + }) +} + +/// Run each stage to completion in order over the shared upper, with no +/// inter-stage pipes (all stdio inherited). Stops at the first non-zero exit. +async fn drive_txn_stages( + stages: Vec, + shared: crate::sandbox::SharedCow, +) -> Result<(bool, Vec, Option), SandlockError> { + let mut results: Vec = Vec::with_capacity(stages.len()); + for (i, stage) in stages.into_iter().enumerate() { + let cmd_refs: Vec<&str> = stage.args.iter().map(|s| s.as_str()).collect(); + let mut sb = stage.sandbox.with_name(format!("txn-stage-{i}")); + sb.set_shared_cow(shared.clone())?; + sb.create_with_io(&cmd_refs, None, None, None).await?; + sb.start()?; + let result = sb.wait().await?; + + let status = result.exit_status.clone(); + results.push(result); + if !matches!(status, ExitStatus::Code(0)) { + return Ok(( + false, + results, + Some(format!("stage {i} did not exit cleanly: {status:?}")), + )); + } + } + Ok((true, results, None)) +} + // ============================================================ // Internal: run_pipeline // ============================================================ diff --git a/crates/sandlock-core/src/sandbox.rs b/crates/sandlock-core/src/sandbox.rs index 5f46a33a..c62c9b86 100644 --- a/crates/sandlock-core/src/sandbox.rs +++ b/crates/sandlock-core/src/sandbox.rs @@ -271,6 +271,26 @@ struct Runtime { on_bind: Option) + Send + Sync>>, handlers: Vec<(i64, Arc)>, ready_w: Option, + /// Set when this sandbox is a stage of a transactional pipeline that shares + /// one COW upper across all stages. When present, `do_create_stdio` reuses + /// this `CowState` (instead of building its own branch), and neither `wait()` + /// nor `Drop` take/commit/abort the branch — the pipeline coordinator owns + /// the single commit/abort. See [`SharedCow`] and `Pipeline::run_transactional`. + shared_cow: Option, +} + +/// A COW branch (one `upper` over the workdir) shared by every stage of a +/// transactional pipeline. Cloned into each stage's `Runtime` so sequential +/// stages accumulate writes in the same upper (read-committed), while the +/// coordinator retains the original to commit/abort once at the end. +#[derive(Clone)] +pub(crate) struct SharedCow { + /// The shared supervisor COW state (holds the single `SeccompCowBranch`). + pub(crate) state: Arc>, + /// The branch's upper dir, granted to each stage's Landlock ruleset so a + /// binary created inside the workdir (materialized into the upper) is + /// executable. Cached here to avoid locking `state` just to read it. + pub(crate) upper_dir: PathBuf, } /// Lifecycle state for the runtime. @@ -782,9 +802,15 @@ impl Sandbox { if let Some(h) = rt.throttle_handle.take() { h.abort(); } if let Some(h) = rt.loadavg_handle.take() { h.abort(); } - if let Some(ref cow_state) = self.rt().supervisor_cow.clone() { - let mut cow = cow_state.lock().await; - self.rt_mut().seccomp_cow = cow.branch.take(); + // A transactional-pipeline stage leaves the branch in the shared COW + // state for the next stage / the coordinator's single commit — don't + // take it out (that would strip the upper from later stages) and don't + // let Drop commit/abort it (`seccomp_cow` stays None). + if self.rt().shared_cow.is_none() { + if let Some(ref cow_state) = self.rt().supervisor_cow.clone() { + let mut cow = cow_state.lock().await; + self.rt_mut().seccomp_cow = cow.branch.take(); + } } let stdout = self.rt_mut()._stdout_read.take().map(sandbox_read_fd_to_end); @@ -1328,6 +1354,7 @@ impl Sandbox { on_bind: None, handlers: Vec::new(), ready_w: None, + shared_cow: None, })); clones.push(clone_sb); } @@ -1422,10 +1449,21 @@ impl Sandbox { on_bind: None, handlers: Vec::new(), ready_w: None, + shared_cow: None, })); Ok(()) } + /// Attach a shared transactional-pipeline COW branch to this sandbox before + /// `create`. The stage reuses the shared upper instead of building its own, + /// and leaves commit/abort to the pipeline coordinator. Internal — used only + /// by `Pipeline::run_transactional`. + pub(crate) fn set_shared_cow(&mut self, shared: SharedCow) -> Result<(), crate::error::SandlockError> { + self.ensure_runtime()?; + self.rt_mut().shared_cow = Some(shared); + Ok(()) + } + // ================================================================ // Internal: collect_changes / do_abort // ================================================================ @@ -1575,7 +1613,14 @@ impl Sandbox { // workdir live in the upper dir, and Landlock checks EXECUTE on the // file's real path at execve time — so the upper dir must be granted // read+execute (READ_ACCESS) or `./created-binary` fails with EACCES. - let seccomp_cow_branch = if !no_supervisor && self.workdir.is_some() { + // A transactional-pipeline stage reuses one shared upper across stages; + // it must not build its own branch. Grant Landlock read+exec on the + // shared upper so a binary created in the workdir stays executable. + let shared_cow = self.rt().shared_cow.clone(); + let seccomp_cow_branch = if let Some(ref shared) = shared_cow { + self.fs_readable.push(shared.upper_dir.clone()); + None + } else if !no_supervisor && self.workdir.is_some() { let workdir = self.workdir.as_ref().unwrap().clone(); let storage = self.fs_storage.clone(); let max_disk = self.max_disk.map(|b| b.0).unwrap_or(0); @@ -1847,6 +1892,10 @@ impl Sandbox { let mut cow_state = CowState::new(); cow_state.branch = seccomp_cow_branch; + // A shared transactional-pipeline branch overrides the per-stage one: + // reuse the coordinator's single COW state so every stage writes into + // (and reads through) the same upper. + let shared_cow_state = shared_cow.as_ref().map(|s| Arc::clone(&s.state)); let mut policy_fn_state = PolicyFnState::new(); @@ -1895,7 +1944,10 @@ impl Sandbox { let res_state = Arc::new(tokio::sync::Mutex::new(res_state)); self.rt_mut().supervisor_resource = Some(Arc::clone(&res_state)); - let cow_state = Arc::new(tokio::sync::Mutex::new(cow_state)); + let cow_state = match shared_cow_state { + Some(shared) => shared, + None => Arc::new(tokio::sync::Mutex::new(cow_state)), + }; self.rt_mut().supervisor_cow = Some(Arc::clone(&cow_state)); let net_state = Arc::new(tokio::sync::Mutex::new(net_state)); @@ -2091,7 +2143,9 @@ impl Drop for Sandbox { match action { BranchAction::Commit => { let _ = cow.commit(); } BranchAction::Abort => { let _ = cow.abort(); } - BranchAction::Keep => {} + // Mark kept so the branch's Drop backstop preserves the upper + // instead of cleaning it as an undisposed leak. + BranchAction::Keep => cow.keep(), } } } diff --git a/crates/sandlock-core/tests/integration.rs b/crates/sandlock-core/tests/integration.rs index bb93f880..b9b30097 100644 --- a/crates/sandlock-core/tests/integration.rs +++ b/crates/sandlock-core/tests/integration.rs @@ -31,6 +31,9 @@ mod test_landlock; #[path = "integration/test_pipeline.rs"] mod test_pipeline; +#[path = "integration/test_pipeline_txn.rs"] +mod test_pipeline_txn; + #[path = "integration/test_network.rs"] mod test_network; diff --git a/crates/sandlock-core/tests/integration/test_pipeline_txn.rs b/crates/sandlock-core/tests/integration/test_pipeline_txn.rs new file mode 100644 index 00000000..cfd292a2 --- /dev/null +++ b/crates/sandlock-core/tests/integration/test_pipeline_txn.rs @@ -0,0 +1,436 @@ +//! Transactional pipeline tests (RFC #65 Phase 1). +//! +//! Sequential stages share one COW upper over a common workdir: a later stage +//! sees an earlier stage's writes (read-committed), and the whole pipeline +//! commits all-or-nothing. Data is exchanged through the shared workspace, not +//! inter-stage pipes. + +use sandlock_core::sandbox::BranchAction; +use sandlock_core::{Sandbox, Stage}; +use std::fs; +use std::os::unix::io::AsRawFd; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +fn temp_dir(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("sandlock-test-txn-{}-{}", name, std::process::id())); + let _ = fs::remove_dir_all(&dir); + let _ = fs::create_dir_all(&dir); + dir +} + +/// Base policy shared by every stage: read the system, write+COW the workdir, +/// and run with the workdir as cwd so relative paths resolve into the upper. +/// `on_exit`/`on_error` are left at their defaults (the pipeline owns commit). +fn stage_policy(workdir: &Path) -> Sandbox { + Sandbox::builder() + .fs_read("/usr").fs_read("/lib").fs_read_if_exists("/lib64").fs_read("/bin").fs_read("/etc") + .fs_read("/proc") + .fs_write(workdir) + .workdir(workdir) + .cwd(workdir) + .build() + .unwrap() +} + +/// Whether this environment can actually run a sandbox (Landlock + seccomp). Used +/// to skip the behavioral tests EXPLICITLY, so a real regression in the +/// transaction logic hard-fails instead of hiding behind a tolerated error. +async fn sandbox_available() -> bool { + let mut sb = Sandbox::builder().fs_read("/usr").fs_read("/bin").build().unwrap(); + matches!(sb.run(&["true"]).await, Ok(r) if r.success()) +} + +/// Number of branch subdirectories under a `fs_storage` dir. Zero means every COW +/// branch's upper has been reclaimed (committed or aborted or dropped). +fn branch_count(storage: &Path) -> usize { + fs::read_dir(storage).map(|rd| rd.count()).unwrap_or(0) +} + +/// Full success: stage 1 writes `a.txt`, stage 2 reads it (proving read-committed) +/// and writes `b.txt`, stage 3 reads both. On commit both files land in workdir. +#[tokio::test] +async fn test_txn_pipeline_commits_on_success() { + if !sandbox_available().await { + eprintln!("commit test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("commit"); + let policy = stage_policy(&workdir); + + let pipeline = Stage::new(&policy, &["sh", "-c", "echo plan > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "cat a.txt && echo built > b.txt"]) + | Stage::new(&policy, &["sh", "-c", "cat a.txt b.txt"]); + + let outcome = pipeline.run_transactional(None).await.expect("transaction should run"); + assert!(outcome.committed, "pipeline should commit; abort_reason: {:?}", outcome.abort_reason); + assert_eq!(outcome.stages.len(), 3, "all three stages should have run"); + assert!(workdir.join("a.txt").exists(), "a.txt must be committed to workdir"); + assert!(workdir.join("b.txt").exists(), "b.txt must be committed to workdir"); + assert_eq!(fs::read_to_string(workdir.join("a.txt")).unwrap(), "plan\n"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// First-committer-win: while another commit holds the workdir lock, a +/// transaction that ran every stage to success still does NOT commit — it loses +/// the race benignly (committed=false + abort_reason) instead of erroring or +/// racing the file-by-file merge and corrupting the workdir. +#[tokio::test] +async fn test_txn_pipeline_loses_first_committer_win_race() { + if !sandbox_available().await { + eprintln!("first-committer-win test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("first-committer-win"); + let policy = stage_policy(&workdir); + + // Simulate a concurrent transaction mid-commit by holding the workdir lock. + let held = std::fs::File::open(&workdir).unwrap(); + assert_eq!( + unsafe { libc::flock(held.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }, + 0, + "test setup: could not take the workdir lock" + ); + + // Every stage succeeds, so the pipeline reaches commit; the held lock makes it + // lose the race. run_transactional must return Ok — a lost race is not an error. + let pipeline = Stage::new(&policy, &["sh", "-c", "echo plan > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "cat a.txt && echo built > b.txt"]); + let outcome = pipeline + .run_transactional(None) + .await + .expect("a lost first-committer-win race is a benign outcome, not an error"); + + assert!( + !outcome.committed, + "must not commit while another commit holds the workdir lock" + ); + assert!( + outcome + .abort_reason + .as_deref() + .unwrap_or("") + .contains("first-committer-win"), + "abort_reason should name the lost race, got: {:?}", + outcome.abort_reason + ); + // Nothing may have been merged into the workdir. + assert!( + !workdir.join("a.txt").exists() && !workdir.join("b.txt").exists(), + "no stage output may be committed when the race is lost" + ); + + drop(held); + let _ = fs::remove_dir_all(&workdir); +} + +/// Any stage failing aborts the whole transaction: earlier stages' writes are +/// discarded and the workdir is byte-identical to before the run. +#[tokio::test] +async fn test_txn_pipeline_aborts_on_stage_failure() { + if !sandbox_available().await { + eprintln!("abort test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("abort"); + fs::write(workdir.join("existing.txt"), "original\n").unwrap(); + let policy = stage_policy(&workdir); + + // Stage 2 writes b.txt but the final stage exits non-zero → abort all. + let pipeline = Stage::new(&policy, &["sh", "-c", "echo plan > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "cat a.txt && echo built > b.txt"]) + | Stage::new(&policy, &["sh", "-c", "exit 1"]); + + let outcome = pipeline.run_transactional(None).await.expect("transaction should run"); + assert!(!outcome.committed, "a failing stage must abort the transaction"); + assert!(outcome.abort_reason.is_some(), "abort must carry a reason"); + assert!(!workdir.join("a.txt").exists(), "a.txt must NOT leak after abort"); + assert!(!workdir.join("b.txt").exists(), "b.txt must NOT leak after abort"); + assert_eq!(fs::read_to_string(workdir.join("existing.txt")).unwrap(), "original\n"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// The shared upper is reclaimed from disk after BOTH abort and commit — the +/// end-to-end check that a failed/completed transaction never orphans its upper. +#[tokio::test] +async fn test_txn_pipeline_reclaims_upper() { + if !sandbox_available().await { + eprintln!("reclaim test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("reclaim-wd"); + let storage = temp_dir("reclaim-st"); + let policy = Sandbox::builder() + .fs_read("/usr").fs_read("/lib").fs_read_if_exists("/lib64").fs_read("/bin").fs_read("/etc") + .fs_read("/proc") + .fs_write(&workdir).workdir(&workdir).cwd(&workdir) + .fs_storage(&storage) + .build() + .unwrap(); + + // Abort path: the upper must be gone from the storage dir. + let aborted = (Stage::new(&policy, &["sh", "-c", "echo x > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "exit 1"])) + .run_transactional(None).await.expect("transaction should run"); + assert!(!aborted.committed); + assert_eq!(branch_count(&storage), 0, "aborted pipeline must reclaim its upper from the storage dir"); + + // Commit path: also reclaimed after the merge. + let committed = (Stage::new(&policy, &["sh", "-c", "echo y > b.txt"]) + | Stage::new(&policy, &["sh", "-c", "cat b.txt"])) + .run_transactional(None).await.expect("transaction should run"); + assert!(committed.committed); + assert_eq!(branch_count(&storage), 0, "committed pipeline must reclaim its upper from the storage dir"); + + let _ = fs::remove_dir_all(&workdir); + let _ = fs::remove_dir_all(&storage); +} + +/// A pipeline timeout aborts the whole transaction without leaking earlier +/// stages' writes into the workdir. +#[tokio::test] +async fn test_txn_pipeline_timeout_aborts() { + if !sandbox_available().await { + eprintln!("timeout test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("timeout"); + let policy = stage_policy(&workdir); + + let pipeline = Stage::new(&policy, &["sh", "-c", "echo plan > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "sleep 30"]); + + let outcome = pipeline + .run_transactional(Some(Duration::from_millis(600))) + .await + .expect("transaction should run"); + assert!(!outcome.committed, "a timed-out pipeline must abort"); + assert!( + outcome.abort_reason.as_deref().unwrap_or("").contains("timed out"), + "abort reason should mention the timeout, got: {:?}", + outcome.abort_reason + ); + assert!(!workdir.join("a.txt").exists(), "a.txt must NOT leak after a timeout abort"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Guardrail: a non-default `on_exit`/`on_error` conflicts with the pipeline +/// owning commit/abort, and is rejected before anything runs. (No sandbox needed.) +#[tokio::test] +async fn test_txn_pipeline_rejects_branch_action() { + let workdir = temp_dir("guard-action"); + let plain = stage_policy(&workdir); + let with_action = Sandbox::builder() + .fs_read("/usr").fs_write(&workdir).workdir(&workdir) + .on_exit(BranchAction::Keep) + .build() + .unwrap(); + + let pipeline = Stage::new(&plain, &["true"]) | Stage::new(&with_action, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("on_exit/on_error"), "expected the on_exit guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Guardrail: every stage must set a workdir (the shared transaction dir). +#[tokio::test] +async fn test_txn_pipeline_rejects_missing_workdir() { + let workdir = temp_dir("guard-workdir"); + let with_wd = stage_policy(&workdir); + let no_wd = Sandbox::builder().fs_read("/usr").build().unwrap(); + + let pipeline = Stage::new(&with_wd, &["true"]) | Stage::new(&no_wd, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("no workdir"), "expected the workdir guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Guardrail: fewer than two stages is rejected, not a panic (the `pub stages` +/// field bypasses `Pipeline::new`'s check). +#[tokio::test] +async fn test_txn_pipeline_rejects_too_few_stages() { + let workdir = temp_dir("guard-count"); + let policy = stage_policy(&workdir); + let single = sandlock_core::Pipeline { stages: vec![Stage::new(&policy, &["true"])] }; + let err = single.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("at least 2 stages"), "expected the stage-count guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Guardrail: stages that each set a workdir but a DIFFERENT one are rejected — +/// distinct from the missing-workdir case (they share one COW upper). +#[tokio::test] +async fn test_txn_pipeline_rejects_mismatched_workdir() { + let wd1 = temp_dir("guard-wd-a"); + let wd2 = temp_dir("guard-wd-b"); + let s0 = stage_policy(&wd1); + let s1 = stage_policy(&wd2); // valid workdir, but not the same one + let pipeline = Stage::new(&s0, &["true"]) | Stage::new(&s1, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("share one workdir"), "expected the shared-workdir guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&wd1); + let _ = fs::remove_dir_all(&wd2); +} + +/// Guardrail: a stage running without the supervisor cannot participate in a COW +/// transaction (no notif path to build/commit the shared upper). +#[tokio::test] +async fn test_txn_pipeline_rejects_no_supervisor() { + let workdir = temp_dir("guard-nosup"); + let ok = stage_policy(&workdir); + // Same workdir (so the workdir guardrail doesn't fire first) but no supervisor. + let nosup = Sandbox::builder() + .fs_read("/usr").fs_write(&workdir).workdir(&workdir).cwd(&workdir) + .no_supervisor(true) + .build() + .unwrap(); + let pipeline = Stage::new(&ok, &["true"]) | Stage::new(&nosup, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("no_supervisor"), "expected the no_supervisor guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Guardrail: chroot is unsupported with a shared COW workdir (the workdir path +/// can't resolve the same across differing roots). +#[tokio::test] +async fn test_txn_pipeline_rejects_chroot() { + let workdir = temp_dir("guard-chroot"); + let rootfs = temp_dir("guard-chroot-root"); + let ok = stage_policy(&workdir); + let with_chroot = Sandbox::builder() + .fs_read("/usr").fs_write(&workdir).workdir(&workdir).cwd(&workdir) + .chroot(&rootfs) + .build() + .unwrap(); + let pipeline = Stage::new(&ok, &["true"]) | Stage::new(&with_chroot, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("chroot"), "expected the chroot guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); + let _ = fs::remove_dir_all(&rootfs); +} + +/// Guardrail: stages must share one COW upper, so differing fs_storage/max_disk +/// (here stage 1 sets fs_storage while stage 0 does not) is rejected. +#[tokio::test] +async fn test_txn_pipeline_rejects_mismatched_fs_storage() { + let workdir = temp_dir("guard-store-wd"); + let storage = temp_dir("guard-store-st"); + let s0 = stage_policy(&workdir); // no fs_storage + let s1 = Sandbox::builder() + .fs_read("/usr").fs_write(&workdir).workdir(&workdir).cwd(&workdir) + .fs_storage(&storage) + .build() + .unwrap(); + let pipeline = Stage::new(&s0, &["true"]) | Stage::new(&s1, &["true"]); + let err = pipeline.run_transactional(None).await.unwrap_err().to_string(); + assert!(err.contains("fs_storage/max_disk"), "expected the fs_storage guardrail, got: {err}"); + + let _ = fs::remove_dir_all(&workdir); + let _ = fs::remove_dir_all(&storage); +} + +/// Boundary: the FIRST stage failing aborts, and the pipeline STOPS — later +/// stages must not run (outcome.stages holds only the failed stage). Distinct +/// from the last-stage-failure case. +#[tokio::test] +async fn test_txn_pipeline_aborts_on_first_stage_failure() { + if !sandbox_available().await { + eprintln!("first-fail test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("first-fail"); + let policy = stage_policy(&workdir); + // Stage 0 writes a.txt then exits non-zero; stages 1 and 2 must NOT run. + let pipeline = Stage::new(&policy, &["sh", "-c", "echo a > a.txt; exit 1"]) + | Stage::new(&policy, &["sh", "-c", "echo b > b.txt"]) + | Stage::new(&policy, &["sh", "-c", "echo c > c.txt"]); + + let outcome = pipeline.run_transactional(None).await.expect("transaction should run"); + assert!(!outcome.committed, "first-stage failure must abort"); + assert_eq!(outcome.stages.len(), 1, "pipeline must stop at the failed stage — later stages must not run"); + assert!(!workdir.join("a.txt").exists(), "a.txt must NOT leak after abort"); + assert!(!workdir.join("b.txt").exists(), "stage 2 must not have run"); + assert!(!workdir.join("c.txt").exists(), "stage 3 must not have run"); + + let _ = fs::remove_dir_all(&workdir); +} + +/// Combination: a timeout aborts AND reclaims the shared upper (no orphan on the +/// timeout path — the reclaim test only covered clean abort/commit). +#[tokio::test] +async fn test_txn_pipeline_timeout_reclaims_upper() { + if !sandbox_available().await { + eprintln!("timeout-reclaim test skipped: sandbox unavailable"); + return; + } + let workdir = temp_dir("to-reclaim-wd"); + let storage = temp_dir("to-reclaim-st"); + let policy = Sandbox::builder() + .fs_read("/usr").fs_read("/lib").fs_read_if_exists("/lib64").fs_read("/bin").fs_read("/etc") + .fs_read("/proc") + .fs_write(&workdir).workdir(&workdir).cwd(&workdir) + .fs_storage(&storage) + .build() + .unwrap(); + + let outcome = (Stage::new(&policy, &["sh", "-c", "echo x > a.txt"]) + | Stage::new(&policy, &["sh", "-c", "sleep 30"])) + .run_transactional(Some(Duration::from_millis(600))) + .await + .expect("transaction should run"); + assert!(!outcome.committed, "timed-out pipeline must abort"); + assert_eq!(branch_count(&storage), 0, "timed-out pipeline must reclaim its upper from the storage dir"); + + let _ = fs::remove_dir_all(&workdir); + let _ = fs::remove_dir_all(&storage); +} + +/// COW deletion (whiteout) semantics through commit AND abort: +/// - commit: a stage deleting a pre-existing workdir file removes it from the +/// workdir, and a later stage sees the deletion (read-committed); +/// - abort: the deletion is discarded — the file survives byte-identical. +#[tokio::test] +async fn test_txn_pipeline_deletion_commit_applies_abort_preserves() { + if !sandbox_available().await { + eprintln!("deletion test skipped: sandbox unavailable"); + return; + } + // Commit path. + let wd_c = temp_dir("del-commit"); + fs::write(wd_c.join("keep.txt"), "orig\n").unwrap(); + let p_c = stage_policy(&wd_c); + let committed = (Stage::new(&p_c, &["sh", "-c", "rm keep.txt"]) + | Stage::new(&p_c, &["sh", "-c", "test ! -e keep.txt"])) // stage 2 must SEE the deletion + .run_transactional(None) + .await + .expect("transaction should run"); + assert!(committed.committed, "commit expected; abort_reason: {:?}", committed.abort_reason); + assert!(!wd_c.join("keep.txt").exists(), "committed deletion must remove keep.txt from the workdir"); + + // Abort path: same deletion, but the pipeline aborts → deletion discarded. + let wd_a = temp_dir("del-abort"); + fs::write(wd_a.join("keep.txt"), "orig\n").unwrap(); + let p_a = stage_policy(&wd_a); + let aborted = (Stage::new(&p_a, &["sh", "-c", "rm keep.txt"]) + | Stage::new(&p_a, &["sh", "-c", "exit 1"])) + .run_transactional(None) + .await + .expect("transaction should run"); + assert!(!aborted.committed, "abort expected"); + assert_eq!( + fs::read_to_string(wd_a.join("keep.txt")).unwrap(), "orig\n", + "aborted deletion must leave keep.txt intact in the workdir", + ); + + let _ = fs::remove_dir_all(&wd_c); + let _ = fs::remove_dir_all(&wd_a); +}