pipeline: transactional sequential pipelines over a shared COW workdir (RFC #65 Phase 1)

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();

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).

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<Duration>,
+    ) -> Result<TxnOutcome, SandlockError> {
+        validate_txn_stages(&self.stages)?;
+        run_pipeline_txn(self.stages, timeout).await
+    }
 }
 
 impl std::ops::BitOr<Stage> for Pipeline {
@@ -118,6 +149,222 @@ impl std::ops::BitOr<Stage> 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<RunResult>,
+    /// Human-readable reason the transaction aborted; `None` when committed.
+    pub abort_reason: Option<String>,
+}
+
+/// 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<String>) -> 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<Stage>,
+    timeout: Option<Duration>,
+) -> Result<TxnOutcome, SandlockError> {
+    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<RunResult>, Option<String>), 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<Stage>,
+    shared: crate::sandbox::SharedCow,
+) -> Result<(bool, Vec<RunResult>, Option<String>), SandlockError> {
+    let mut results: Vec<RunResult> = 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
 // ============================================================