diff --git a/Cargo.lock b/Cargo.lock index 7359c40..692e774 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5076,6 +5076,9 @@ dependencies = [ "getrandom 0.3.4", "proptest", "rusqlite", + "serde", + "serde_json", + "sha2", "tempfile", "tokio", "wasip2", diff --git a/host/src/api/mod.rs b/host/src/api/mod.rs index 310ee52..2585c86 100644 --- a/host/src/api/mod.rs +++ b/host/src/api/mod.rs @@ -4,6 +4,7 @@ mod auth; mod pages; mod server; mod ssh; +mod undo; pub use server::{Listener, Options, run_with_userland}; use crate::cred::{self, Caller, Cred, entry::EntryKey, webauthn::Passkeys}; @@ -23,6 +24,7 @@ pub struct Api { pub passkeys: Mutex>, pub entry: EntryKey, requests: Mutex>, + journal: Mutex>>, } impl Api { /// Own the connection and ceremony state shared by both transports. @@ -51,8 +53,12 @@ impl Api { passkeys: Mutex::new(passkeys), entry, requests: Mutex::new(HashMap::new()), + journal: Mutex::new(None), } } + pub(crate) fn attach_journal(&self, journal: Option>) { + *self.journal.lock().unwrap() = journal; + } fn rate_limit(&self, caller: &Caller) -> Result<()> { let now = Instant::now(); let mut requests = self.requests.lock().map_err(|_| Error(500, "storage"))?; @@ -132,6 +138,9 @@ fn request_view(request: cred::requests::Request) -> Result { /// Handle a request without a network transport. Errors contain only static error kinds. pub async fn handle(ctx: &Api, caller: Caller, request: Request) -> Response { + if request.uri().path() == "/undo" || request.uri().path().starts_with("/undo/") { + return undo::handle(ctx, caller, request).await; + } respond(ctx, caller, request) } fn respond(ctx: &Api, caller: Caller, request: Request) -> Response { diff --git a/host/src/api/server.rs b/host/src/api/server.rs index ecb182f..6ae4534 100644 --- a/host/src/api/server.rs +++ b/host/src/api/server.rs @@ -1,5 +1,5 @@ //! The host HTTP listener and console startup share one owned API and credentials store. -use super::{Api, respond}; +use super::Api; use crate::cred::{Caller, Cred}; use std::{net::SocketAddr, path::Path, sync::Arc}; @@ -37,6 +37,10 @@ impl Listener { ); let serving = server.clone(); let thread = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("API runtime"); for mut request in serving.incoming_requests() { let response = (|| -> anyhow::Result<_> { let mut body = Vec::new(); @@ -45,7 +49,7 @@ impl Listener { .method(request.method().as_str()) .uri(request.url()) .body(bytes::Bytes::from(body))?; - Ok(respond( + Ok(runtime.block_on(super::handle( &api, Caller { program: "/host/browser".into(), @@ -53,7 +57,7 @@ impl Listener { conversations: Vec::new(), }, request, - )) + ))) })(); let response = response.unwrap_or_else(|_| { http::Response::builder() diff --git a/host/src/api/undo.rs b/host/src/api/undo.rs new file mode 100644 index 0000000..f19d657 --- /dev/null +++ b/host/src/api/undo.rs @@ -0,0 +1,60 @@ +//! Undo routes share the world's journal. They require no credential approval. +use super::*; +use sausage_vfs::journal::{Tag, Undo}; +pub(super) async fn handle(ctx: &Api, caller: Caller, request: Request) -> Response { + let journal = ctx.journal.lock().unwrap().clone(); + let result = async { + let journal = journal.ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::NotFound, "journal unavailable") + })?; + journal.retain().await?; + let query: HashMap = + url::form_urlencoded::parse(request.uri().query().unwrap_or("").as_bytes()) + .into_owned() + .collect(); + match (request.method().as_str(), request.uri().path()) { + ("GET", "/undo" | "/undo/list") => Ok(( + 200, + serde_json::to_value(journal.list(query.get("conversation").cloned()).await?)?, + )), + ("GET", "/undo/show") => { + let tag = query.get("tag").ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "tag required") + })?; + Ok((200, serde_json::to_value(journal.show(tag.clone()).await?)?)) + } + ("POST", "/undo") => { + let request: Undo = serde_json::from_slice(request.body())?; + let result = journal + .undo( + request, + Tag::process(caller.pid, caller.program, caller.conversations)?, + ) + .await?; + let status = if result.tag.is_none() { 409 } else { 200 }; + Ok((status, serde_json::to_value(result)?)) + } + _ => Ok((404, json!({"error":"not_found"}))), + } + } + .await; + let (status, body) = match result { + Ok(result) => result, + Err(error) => { + let error: Box = error; + let status = error + .downcast_ref::() + .map_or(400, |e| match e.kind() { + std::io::ErrorKind::NotFound => 404, + std::io::ErrorKind::InvalidInput => 400, + _ => 500, + }); + (status, json!({"error":error.to_string()})) + } + }; + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Bytes::from(serde_json::to_vec(&body).expect("JSON value"))) + .expect("valid response") +} diff --git a/host/src/exec.rs b/host/src/exec.rs index 21870ef..6711926 100644 --- a/host/src/exec.rs +++ b/host/src/exec.rs @@ -17,6 +17,8 @@ pub struct Options { pub env: Vec<(String, String)>, pub timeout: Option, pub conversations: Vec, + /// Undo tag shared by this tool call and all child processes. + pub tool_call: Option, } /// The stream that one output chunk came from. diff --git a/host/src/process/mod.rs b/host/src/process/mod.rs index 5397a81..c305c6d 100644 --- a/host/src/process/mod.rs +++ b/host/src/process/mod.rs @@ -83,6 +83,7 @@ struct Entry { command: u32, group: u32, cancelled: Arc, + tag: sausage_vfs::journal::Tag, abort: AbortHandle, } /// One runtime keeps compiled components warm and shares the world's namespace across stores. @@ -191,6 +192,7 @@ impl Runtime { "http://localhost:8081", )?), }; + router.attach_journal(world.root.journal()); let network = Arc::new(net::NetPolicy::open(&world.host_dir)?); let engine = compiler.engine.clone(); let alive = Arc::new(AtomicBool::new(true)); @@ -313,6 +315,7 @@ impl Runtime { self.kill_group(group); Ok(()) } + #[allow(clippy::too_many_arguments)] async fn start( self: &Arc, mut request: Request, @@ -321,6 +324,7 @@ impl Runtime { inherited_group: u32, conversations: Vec, inherited_io: Option, + tool_call: Option, ) -> anyhow::Result { anyhow::ensure!(!request.argv.is_empty(), "empty argv"); let cancelled = if parent == 0 { @@ -361,6 +365,27 @@ impl Runtime { let pre = self.compiler.get(bytes.clone()).await?; self.cache.lock().unwrap().insert(bytes); let pid = self.next.fetch_add(1, Ordering::Relaxed); + let tag = if parent == 0 { + let mut tag = + sausage_vfs::journal::Tag::process(pid, executable.clone(), conversations.clone())?; + if let Some(id) = tool_call { + tag.id = id; + } + tag + } else { + self.entries + .lock() + .unwrap() + .get(&parent) + .ok_or_else(|| anyhow::anyhow!("parent exited"))? + .tag + .clone() + }; + let root = self + .world + .root + .tagged(tag.clone()) + .unwrap_or_else(|| self.world.root.clone()); let command = if command == 0 { pid } else { command }; let group = match request.group { Some(0) => pid, @@ -384,7 +409,7 @@ impl Runtime { }; let preview1 = if matches!(pre, Compiled::Module(_)) { Some(sausage_vfs::preview1::State::new( - self.world.root.clone(), + root.clone(), &request.cwd, request.argv.clone(), request.env.clone(), @@ -396,7 +421,7 @@ impl Runtime { None }; let mut mounts = Mounts::default(); - mounts.mount("/", self.world.root.clone(), false); + mounts.mount("/", root, false); let mut wasi = WasiCtxBuilder::new(); wasi.args(&request.argv) .envs(&request.env) @@ -499,6 +524,7 @@ impl Runtime { command, group, cancelled: cancelled.clone(), + tag, abort: task.abort_handle(), }, ); @@ -552,6 +578,7 @@ impl Host for State { self.group, self.conversations.clone(), Some(self.stdio.clone()), + None, ) .await { @@ -706,6 +733,7 @@ impl Exec for Arc { 0, options.conversations, None, + options.tool_call, ); let mut child = if let Some(deadline) = deadline { tokio::time::timeout_at(deadline, start) @@ -737,13 +765,17 @@ impl Exec for Arc { child.code.changed().await?; } }; - if let Some(deadline) = deadline { + let result = if let Some(deadline) = deadline { tokio::time::timeout_at(deadline, run) .await .map_err(|_| anyhow::anyhow!("command timed out"))? } else { run.await + }; + if let Some(journal) = self.world.root.journal() { + journal.retain().await?; } + result } } /// Construct the harness's shared executor over the same open world. diff --git a/host/src/process/services.rs b/host/src/process/services.rs index f7b06f6..9ddf59a 100644 --- a/host/src/process/services.rs +++ b/host/src/process/services.rs @@ -33,6 +33,7 @@ pub fn supervise(runtime: Arc) -> Supervisor { 0, Vec::new(), None, + None, ); let child = tokio::select! { _ = &mut stop => return, child = start => child }; if let Ok(mut child) = child { diff --git a/host/src/world.rs b/host/src/world.rs index b77674e..811744f 100644 --- a/host/src/world.rs +++ b/host/src/world.rs @@ -77,7 +77,13 @@ impl World { }); mounts.mount("/proc", Arc::new(proc), true); mounts.mount("/run", Arc::new(Memory::default()), false); - let root: Arc = Arc::new(mounts); + let journal = + sausage_vfs::journal::Journal::open(&host_dir.join("undo"), Arc::new(mounts))?; + let root = journal.layer(sausage_vfs::journal::Tag::process( + std::process::id(), + "/host".into(), + Vec::new(), + )?); Ok(World { root, host_dir, diff --git a/host/tests/init.rs b/host/tests/init.rs index 61fe19e..be5772e 100644 --- a/host/tests/init.rs +++ b/host/tests/init.rs @@ -14,7 +14,7 @@ static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); fn guests() -> &'static Path { static BUILT: OnceLock = OnceLock::new(); BUILT.get_or_init(|| { - let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-guests"); + let target = PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap()); let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland/Cargo.toml"); assert!( std::process::Command::new("cargo") diff --git a/host/tests/p1.rs b/host/tests/p1.rs index 65fd7f9..ef4c22d 100644 --- a/host/tests/p1.rs +++ b/host/tests/p1.rs @@ -6,12 +6,12 @@ use std::path::PathBuf; #[tokio::test(flavor = "current_thread")] async fn bundle_copy_and_real_shell() -> anyhow::Result<()> { let repo = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".."); - let bundle = repo.join("target/userland/bundle"); + let bundle = PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap()).join("bundle"); if !bundle.join("bin/sh.wasm").exists() { assert!( std::process::Command::new("nu") - .env_remove("CARGO_TARGET_DIR") .arg(repo.join("userland/bundle.nu")) + .arg(&bundle) .status()? .success() ); diff --git a/host/tests/process.rs b/host/tests/process.rs index 517de26..e659392 100644 --- a/host/tests/process.rs +++ b/host/tests/process.rs @@ -357,8 +357,6 @@ fn build() { assert!( std::process::Command::new(args[0]) .args(&args[1..]) - .env_remove("CARGO_TARGET_DIR") - .env("CARGO_HOME", cwd.join(".cargo-home")) .current_dir(cwd) .status() .unwrap() @@ -403,16 +401,7 @@ fn build() { &root.join("userland"), ); run( - &[ - "cargo", - "build", - "--release", - "-p", - "git", - "--target-dir", - "../target/process-userland", - "--locked", - ], + &["cargo", "build", "--release", "-p", "git", "--locked"], &root.join("userland"), ); }); @@ -442,7 +431,8 @@ fn world() -> (tempfile::TempDir, Arc, sausage::process::Processes) { let processes = world.processes.clone(); fixture(&temp.path().join("world")); - let artifacts = repo().join("userland/target/wasm32-wasip2/release"); + let artifacts = + PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap()).join("wasm32-wasip2/release"); let bin = temp.path().join("world/bin"); for tool in ["cat", "tr", "ls", "wc", "yes", "head", "sleep"] { std::fs::copy( @@ -455,21 +445,9 @@ fn world() -> (tempfile::TempDir, Arc, sausage::process::Processes) { std::fs::copy(artifacts.join("sausage-procps.wasm"), bin.join(tool)).unwrap(); } std::fs::copy(artifacts.join("sausage-cwd-probe.wasm"), bin.join("probe")).unwrap(); - std::fs::copy( - repo().join("userland/brush/source/target/wasm32-wasip2/release/brush.wasm"), - bin.join("sh"), - ) - .unwrap(); - std::fs::copy( - repo().join("userland/brush/probe/target/wasm32-wasip2/release/process-probe.wasm"), - bin.join("launcher"), - ) - .unwrap(); - std::fs::copy( - repo().join("target/process-userland/wasm32-wasip2/release/git.wasm"), - bin.join("git"), - ) - .unwrap(); + std::fs::copy(artifacts.join("brush.wasm"), bin.join("sh")).unwrap(); + std::fs::copy(artifacts.join("process-probe.wasm"), bin.join("launcher")).unwrap(); + std::fs::copy(artifacts.join("git.wasm"), bin.join("git")).unwrap(); ( temp, Runtime::new_with_cache(world, &cache).unwrap(), diff --git a/host/tests/undo.rs b/host/tests/undo.rs new file mode 100644 index 0000000..34a7d6a --- /dev/null +++ b/host/tests/undo.rs @@ -0,0 +1,112 @@ +//! Undo snapshots and reversible namespace changes through a real world. +use sausage::world::World; +use sausage_vfs::{ + Layer, Open, + journal::{Tag, Undo}, +}; +use std::sync::Arc; +fn tag(id: &str) -> Tag { + Tag { + id: id.into(), + pid: 1, + program: "/bin/test".into(), + conversations: vec!["conversation".into()], + } +} +async fn write(root: &Arc, path: &str, bytes: &[u8]) { + root.open( + path, + Open { + write: true, + create: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap() + .write(0, bytes.to_vec()) + .await + .unwrap(); +} +async fn read(root: &Arc, path: &str) -> Vec { + root.open(path, Open::default()) + .await + .unwrap() + .read(0, 100) + .await + .unwrap() +} +#[tokio::test] +async fn conflicts_force_and_undo_of_undo() { + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + write(&world.root, "/a", b"old").await; + let journal = world.root.journal().unwrap(); + let a = journal.layer(tag("A")); + let b = journal.layer(tag("B")); + write(&a, "/a", b"A").await; + write(&a, "/only-a", b"A").await; + write(&b, "/a", b"B").await; + write(&b, "/only-b", b"B").await; + let request = Undo { + tag: Some("A".into()), + ..Default::default() + }; + let blocked = journal.undo(request.clone(), tag("blocked")).await.unwrap(); + assert_eq!(blocked.conflicts, ["/a"]); + assert!(blocked.tag.is_none()); + assert_eq!(read(&world.root, "/a").await, b"B"); + let result = journal + .undo( + Undo { + force: true, + ..request + }, + tag("restore"), + ) + .await + .unwrap(); + assert_eq!(read(&world.root, "/a").await, b"old"); + assert!(world.root.stat("/only-a", false).await.is_err()); + assert_eq!(read(&world.root, "/only-b").await, b"B"); + journal + .undo( + Undo { + tag: result.tag, + ..Default::default() + }, + tag("redo"), + ) + .await + .unwrap(); + assert_eq!(read(&world.root, "/a").await, b"B"); + assert_eq!(read(&world.root, "/only-a").await, b"A"); +} +#[tokio::test] +async fn remove_rename_folder_and_inclusive_since() { + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + write(&world.root, "/old", b"old").await; + write(&world.root, "/removed", b"removed").await; + let journal = world.root.journal().unwrap(); + let a = journal.layer(tag("A")); + a.mkdir("/new").await.unwrap(); + a.rename("/old", "/new/moved").await.unwrap(); + a.remove_file("/removed").await.unwrap(); + let entry = journal.list(None).await.unwrap()[0].entry; + write(&journal.layer(tag("B")), "/new/child", b"child").await; + journal + .undo( + Undo { + since: Some(entry), + ..Default::default() + }, + tag("restore"), + ) + .await + .unwrap(); + assert_eq!(read(&world.root, "/old").await, b"old"); + assert_eq!(read(&world.root, "/removed").await, b"removed"); + assert!(world.root.stat("/new", false).await.is_err()); +} diff --git a/host/tests/undo_api.rs b/host/tests/undo_api.rs new file mode 100644 index 0000000..1d606c7 --- /dev/null +++ b/host/tests/undo_api.rs @@ -0,0 +1,73 @@ +//! The host API shares the journal that tags world mutations. +use sausage::{ + api::{Api, handle}, + cred::{Caller, Cred}, + process::{Runtime, default_script}, + world::World, +}; +use sausage_vfs::{Open, journal::Tag}; +use std::sync::Arc; +#[tokio::test] +async fn list_show_and_restore() { + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + let journal = world.root.journal().unwrap(); + let tag = Tag { + id: "A".into(), + pid: 1, + program: "/bin/test".into(), + conversations: vec!["c".into()], + }; + journal + .layer(tag) + .open( + "/file", + Open { + create: true, + write: true, + ..Default::default() + }, + ) + .await + .unwrap() + .write(0, b"data".to_vec()) + .await + .unwrap(); + let cred = Arc::new(Cred::open(&world.host_dir).unwrap()); + let api = Arc::new( + Api::open( + cred.clone(), + &world.host_dir, + "localhost", + "http://localhost:8081", + ) + .unwrap(), + ); + let _runtime = + Runtime::new_with_hooks(world, cred, default_script(), Some(api.clone())).unwrap(); + let request = |method, path, body: &'static str| { + http::Request::builder() + .method(method) + .uri(path) + .body(bytes::Bytes::from_static(body.as_bytes())) + .unwrap() + }; + let caller = || Caller { + pid: 1, + program: "/bin/undo".into(), + conversations: vec!["c".into()], + }; + let response = handle(&api, caller(), request("GET", "/undo?conversation=c", "")).await; + assert_eq!(response.status(), 200); + let entries: serde_json::Value = serde_json::from_slice(response.body()).unwrap(); + assert_eq!(entries[0]["tag"], "A"); + let response = handle(&api, caller(), request("GET", "/undo/show?tag=A", "")).await; + assert_eq!(response.status(), 200); + let snapshots: serde_json::Value = serde_json::from_slice(response.body()).unwrap(); + assert_eq!(snapshots[0]["path"], "/file"); + let response = handle(&api, caller(), request("POST", "/undo", r#"{"tag":"A"}"#)).await; + assert_eq!(response.status(), 200); + assert!(!temp.path().join("world/file").exists()); + let response = handle(&api, caller(), request("POST", "/undo", "{}")).await; + assert_eq!(response.status(), 400); +} diff --git a/host/tests/undo_cost.rs b/host/tests/undo_cost.rs new file mode 100644 index 0000000..5579350 --- /dev/null +++ b/host/tests/undo_cost.rs @@ -0,0 +1,171 @@ +//! Compare identical Git status calls with and without the journal in fresh worlds. +use sausage::{ + exec::{Command, Exec, Options}, + process::Runtime, + world::World, +}; +use sausage_vfs::{Host, Mounts, Process}; +use std::{path::PathBuf, sync::Arc, time::Instant}; +#[tokio::test(flavor = "multi_thread")] +async fn status_cost() { + let target = PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap()); + if !target.join("wasm32-wasip2/release/git.wasm").is_file() { + let userland = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../userland"); + assert!( + std::process::Command::new("nu") + .args(["--no-config-file", "cwd-init/build.nu"]) + .current_dir(&userland) + .status() + .unwrap() + .success() + ); + assert!( + std::process::Command::new("cargo") + .args(["build", "--release", "-p", "git"]) + .current_dir(&userland) + .status() + .unwrap() + .success() + ); + } + let temp = tempfile::tempdir().unwrap(); + let source = temp.path().join("source"); + std::fs::create_dir_all(&source).unwrap(); + let archive = std::process::Command::new("git") + .args([ + "archive", + "HEAD", + "host", + "vfs", + "userland", + "Cargo.toml", + "Cargo.lock", + "flake.nix", + "flake.lock", + "README.md", + "AGENTS.md", + ".gitignore", + ]) + .current_dir(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("..")) + .output() + .unwrap(); + assert!(archive.status.success()); + let tar = temp.path().join("source.tar"); + std::fs::write(&tar, archive.stdout).unwrap(); + assert!( + std::process::Command::new("tar") + .arg("-xf") + .arg(&tar) + .arg("-C") + .arg(&source) + .status() + .unwrap() + .success() + ); + for args in [ + vec!["init"], + vec!["add", "."], + vec![ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "--no-gpg-sign", + "-m", + "fixture", + ], + ] { + let output = std::process::Command::new("git") + .args(args) + .current_dir(&source) + .output() + .unwrap(); + assert!(output.status.success()); + } + let mut times = [Vec::new(), Vec::new()]; + for round in 0..3 { + for enabled in [false, true] { + let instance = temp.path().join(format!("{round}-{enabled}")); + let mut world = World::open(&instance).unwrap(); + let root = instance.join("world"); + std::fs::create_dir_all(root.join("bin")).unwrap(); + std::fs::create_dir_all(root.join("src")).unwrap(); + std::fs::copy( + target.join("wasm32-wasip2/release/git.wasm"), + root.join("bin/git"), + ) + .unwrap(); + assert!( + std::process::Command::new("cp") + .arg("-R") + .arg(&source) + .arg(root.join("src/sausage")) + .status() + .unwrap() + .success() + ); + // Same content, new timestamp: status must refresh its index. + assert!( + std::process::Command::new("touch") + .arg(root.join("src/sausage/README.md")) + .status() + .unwrap() + .success() + ); + if !enabled { + let mut mounts = Mounts::default(); + mounts.mount( + "/", + Arc::new(Host::new(&root, Process::default()).unwrap()), + false, + ); + world.root = Arc::new(mounts); + } + let runtime = Runtime::new(world).unwrap(); + let options = Options { + cwd: "/src/sausage".into(), + ..Default::default() + }; + assert_eq!( + runtime + .exec( + Command::Argv(vec!["/bin/git".into(), "--version".into()]), + options.clone(), + &mut |_, _| {} + ) + .await + .unwrap(), + 0 + ); + let start = Instant::now(); + assert_eq!( + runtime + .exec( + Command::Argv(vec!["/bin/git".into(), "status".into(), "--short".into()]), + options, + &mut |_, bytes| assert!( + bytes.is_empty(), + "{}", + String::from_utf8_lossy(bytes) + ) + ) + .await + .unwrap(), + 0 + ); + times[usize::from(enabled)].push(start.elapsed().as_secs_f64()); + } + } + for values in &mut times { + values.sort_by(f64::total_cmp); + } + println!( + "status without journal: {:.6}s; with journal: {:.6}s", + times[0][1], times[1][1] + ); + assert!( + times[1][1] <= times[0][1] * 1.10, + "journal exceeds ten percent" + ); +} diff --git a/host/tests/undo_exec.rs b/host/tests/undo_exec.rs new file mode 100644 index 0000000..0b7aaca --- /dev/null +++ b/host/tests/undo_exec.rs @@ -0,0 +1,140 @@ +//! Real guest mutations share tool-call tags and restore a committed repository. +use sausage::{ + exec::{Command, Exec, Options, Stream}, + process::Runtime, + world::World, +}; +use sausage_vfs::journal::Outcome; +use std::{path::PathBuf, sync::Arc}; +async fn run(runtime: &Arc, id: &str, argv: &[&str]) -> String { + let mut out = Vec::new(); + let mut err = Vec::new(); + let code = runtime + .exec( + Command::Argv(argv.iter().map(|s| s.to_string()).collect()), + Options { + cwd: "/repo".into(), + tool_call: Some(id.into()), + conversations: vec!["conversation".into()], + ..Default::default() + }, + &mut |stream, bytes| match stream { + Stream::Stdout => out.extend_from_slice(bytes), + Stream::Stderr => err.extend_from_slice(bytes), + }, + ) + .await + .unwrap(); + assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err)); + String::from_utf8(out).unwrap() +} +#[tokio::test(flavor = "multi_thread")] +async fn tool_call_commit_and_independent_change() { + let target = PathBuf::from( + std::env::var("CARGO_TARGET_DIR").expect("set the executor target directory"), + ); + let guest_dir = target.join("wasm32-wasip2/release"); + if !guest_dir.join("git.wasm").is_file() + || !guest_dir.join("cp.wasm").is_file() + || !guest_dir.join("touch.wasm").is_file() + || !guest_dir.join("undo.wasm").is_file() + { + assert!( + std::process::Command::new("nu") + .args(["--no-config-file", "cwd-init/build.nu"]) + .current_dir(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../userland")) + .status() + .unwrap() + .success() + ); + assert!( + std::process::Command::new("cargo") + .args([ + "build", + "--release", + "-p", + "git", + "-p", + "sausage-coreutils", + "-p", + "undo" + ]) + .current_dir(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../userland")) + .status() + .unwrap() + .success() + ); + } + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + let root = temp.path().join("world"); + std::fs::create_dir_all(root.join("bin")).unwrap(); + std::fs::create_dir_all(root.join("repo")).unwrap(); + for (binary, guest) in [ + ("git", "git"), + ("cp", "cp"), + ("touch", "touch"), + ("undo", "undo"), + ] { + std::fs::copy( + target.join(format!("wasm32-wasip2/release/{binary}.wasm")), + root.join(format!("bin/{guest}")), + ) + .unwrap(); + } + let git = |args: &[&str]| { + let output = std::process::Command::new("git") + .args(args) + .current_dir(root.join("repo")) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() + }; + git(&["init", "-b", "main"]); + git(&["config", "user.name", "Test"]); + git(&["config", "user.email", "test@example.com"]); + std::fs::write(root.join("repo/file"), "old\n").unwrap(); + git(&["add", "file"]); + git(&["commit", "--no-gpg-sign", "-m", "initial"]); + let before = git(&["log", "--oneline"]); + std::fs::write(root.join("input"), "new\n").unwrap(); + let runtime = Runtime::new(world).unwrap(); + run(&runtime, "A", &["/bin/cp", "/input", "file"]).await; + run(&runtime, "A", &["/bin/git", "add", "file"]).await; + run(&runtime, "A", &["/bin/git", "commit", "-m", "change"]).await; + run(&runtime, "B", &["/bin/touch", "independent"]).await; + let entries = run( + &runtime, + "list", + &["/bin/undo", "list", "--conversation", "conversation"], + ) + .await; + let entries: Vec = serde_json::from_str(&entries).unwrap(); + assert!(entries.iter().any(|entry| entry.tag == "A")); + let snapshots = run(&runtime, "show", &["/bin/undo", "show", "A"]).await; + let snapshots: Vec = serde_json::from_str(&snapshots).unwrap(); + assert!( + snapshots + .iter() + .any(|snapshot| snapshot.path == "/repo/file") + ); + let output = run(&runtime, "undo-call", &["/bin/undo", "A"]).await; + let outcome: Outcome = serde_json::from_str(&output).unwrap(); + assert!(outcome.conflicts.is_empty(), "{:?}", outcome); + assert!(outcome.tag.is_some()); + assert_eq!(std::fs::read(root.join("repo/file")).unwrap(), b"old\n"); + assert!(root.join("repo/independent").is_file()); + assert_eq!( + run(&runtime, "inspect", &["/bin/git", "log", "--oneline"]).await, + before + ); + assert_eq!( + run(&runtime, "inspect", &["/bin/git", "status", "--short"]).await, + "?? independent\n" + ); +} diff --git a/userland/Cargo.lock b/userland/Cargo.lock index 70d4431..87cbcd3 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -5074,6 +5074,16 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "undo" +version = "0.1.0" +dependencies = [ + "clap", + "curl", + "serde_json", + "url", +] + [[package]] name = "unic-langid" version = "0.9.6" diff --git a/userland/Cargo.toml b/userland/Cargo.toml index cca634f..63002de 100644 --- a/userland/Cargo.toml +++ b/userland/Cargo.toml @@ -1,7 +1,7 @@ # The default CLIs of a world, built for wasm32-wasip2. Its own workspace, outside the host's. [workspace] resolver = "3" -members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "ssh-client", "ssh", "curl", "init"] +members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "ssh-client", "ssh", "curl", "init", "undo"] # WASI compatibility patches: git/vendor/PATCHES.md. [patch.crates-io] diff --git a/userland/undo/Cargo.toml b/userland/undo/Cargo.toml new file mode 100644 index 0000000..6452939 --- /dev/null +++ b/userland/undo/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "undo" +version = "0.1.0" +edition = "2024" + +[dependencies] +clap = { version = "4", features = ["derive"] } +curl = { path = "../curl" } +serde_json = "1" +url = "2" diff --git a/userland/undo/README.md b/userland/undo/README.md new file mode 100644 index 0000000..ce0fbb4 --- /dev/null +++ b/userland/undo/README.md @@ -0,0 +1,24 @@ +# Undo + +The CLI calls `http://host.sausage/undo` through standard WASI HTTP. +The API returns JSON. + +- `undo list [--conversation ID]` lists tags, newest first. +- `undo show TAG` lists the old states for a tag. +- `undo TAG [--force]` restores one tag. +- `undo --since ENTRY [--force]` restores all tags from the named entry, inclusive. + +A later tag that changes the same path causes a conflict. +Undo stops before changes when conflicts or unavailable snapshots exist. +`--force` overrides conflicts and skips unavailable snapshots. +The response lists both conditions. +An undo creates its own tag, which permits another undo. + +The journal records content, directories, and absence, not mode bits. +WASI exposes no mode bits. +Files over 64 MiB and symlinks have unavailable snapshots. +The journal excludes `/proc`, `/run`, and `/tmp`. +The store deduplicates content by SHA-256 outside the world, under `host/undo`. +Retention removes the oldest tags after seven days or above 4 GiB of unique content. +The host applies retention after each command and before each API request. +Undo does not reverse network effects or native builds. diff --git a/userland/undo/src/main.rs b/userland/undo/src/main.rs new file mode 100644 index 0000000..962fc1d --- /dev/null +++ b/userland/undo/src/main.rs @@ -0,0 +1,45 @@ +//! The undo CLI selects journal entries through the reserved host HTTP API. +use clap::Parser; +#[derive(Parser)] +struct Args { + /// A tag, or the list/show command. + #[arg(required_unless_present = "since")] + action: Option, + #[arg(required_if_eq("action", "show"))] + tag: Option, + #[arg(long)] + conversation: Option, + #[arg(long, conflicts_with = "action")] + since: Option, + #[arg(long)] + force: bool, +} +fn main() { + let args = Args::parse(); + let mut url = url::Url::parse("http://host.sausage/undo").unwrap(); + let mut transfer = vec!["undo-http".to_owned()]; + match args.action.as_deref() { + Some("list") => { + if let Some(conversation) = args.conversation { + url.query_pairs_mut() + .append_pair("conversation", &conversation); + } + } + Some("show") => { + let tag = args.tag.expect("required show tag"); + url.set_path("/undo/show"); + url.query_pairs_mut().append_pair("tag", &tag); + } + tag => { + let body = serde_json::json!({"tag":tag,"since":args.since,"force":args.force}); + transfer.extend(["--json".into(), body.to_string()]); + } + } + transfer.extend(["--fail-with-body".into(), url.to_string()]); + let mut request = curl::Args::parse_from(transfer); + request.parts = request.json.iter().map(|s| (s.clone(), 3)).collect(); + if let Err(error) = curl::run(&request) { + eprintln!("undo: {}", error.message); + std::process::exit(error.code); + } +} diff --git a/vfs/Cargo.toml b/vfs/Cargo.toml index 633b14a..adcee7e 100644 --- a/vfs/Cargo.toml +++ b/vfs/Cargo.toml @@ -14,6 +14,10 @@ bytes = "1" tokio = { version = "1", features = ["rt-multi-thread", "sync", "macros", "io-util", "time"] } wiggle = "49" getrandom = "0.3" +rusqlite = { version = "0.38", features = ["bundled"] } +sha2 = "0.10" +serde_json = "1" +serde = { version = "1", features = ["derive"] } [dev-dependencies] wasip2 = "1" diff --git a/vfs/src/journal.rs b/vfs/src/journal.rs new file mode 100644 index 0000000..ad4d98c --- /dev/null +++ b/vfs/src/journal.rs @@ -0,0 +1,245 @@ +//! First-change snapshots per tag, shared across views of one virtual namespace. +//! The SQLite index and deduplicated blobs live outside that namespace. +mod layer; +mod restore; +#[cfg(test)] +mod tests; +use crate::{Layer, Open, Result, observe::under}; +use rusqlite::{Connection, OptionalExtension, params}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashSet, + io, + path::{Path, PathBuf}, + sync::{Arc, Mutex}, + time::{SystemTime, UNIX_EPOCH}, +}; + +const MAX_FILE: u64 = 64 * 1024 * 1024; +const MAX_BYTES: u64 = 4 * 1024 * 1024 * 1024; +const MAX_AGE: u64 = 7 * 24 * 60 * 60; +fn err(e: impl std::fmt::Display) -> io::Error { + io::Error::other(e.to_string()) +} +fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +/// A tool call, or the originating process and its inherited conversation chain. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Tag { + pub id: String, + pub pid: u32, + pub program: String, + pub conversations: Vec, +} +impl Tag { + pub fn process(pid: u32, program: String, conversations: Vec) -> Result { + let mut random = [0; 16]; + getrandom::fill(&mut random).map_err(err)?; + Ok(Self { + id: format!("process:{:032x}", u128::from_le_bytes(random)), + pid, + program, + conversations, + }) + } +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Entry { + pub entry: i64, + pub tag: String, + pub pid: u32, + pub program: String, + pub conversations: Vec, + pub created: u64, + pub paths: usize, +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Snapshot { + pub entry: i64, + pub tag: String, + pub path: String, + /// absent, directory, file, or not_undoable. + pub kind: String, + pub hash: Option, + pub size: u64, +} +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct Undo { + pub tag: Option, + pub since: Option, + #[serde(default)] + pub force: bool, +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Outcome { + pub tag: Option, + pub conflicts: Vec, + pub skipped: Vec, + pub restored: Vec, +} +struct Storage { + db: Connection, + directory: PathBuf, +} +/// Shared journal. Reads of the world do not access its database or mutation lock. +pub struct Journal { + root: Arc, + storage: Arc>, + seen: Mutex>, + gate: tokio::sync::Mutex<()>, + paths: Mutex>>>>, +} +impl Journal { + /// Open the host-owned store outside an asynchronous worker. + pub fn open(directory: &Path, root: Arc) -> Result> { + std::fs::create_dir_all(directory.join("blobs"))?; + let db = Connection::open(directory.join("index.sqlite")).map_err(err)?; + db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; + CREATE TABLE IF NOT EXISTS tags(entry INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, pid INTEGER NOT NULL, program TEXT NOT NULL, conversations TEXT NOT NULL, created INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS snapshots(entry INTEGER PRIMARY KEY AUTOINCREMENT, tag TEXT NOT NULL REFERENCES tags(id) ON DELETE CASCADE, path TEXT NOT NULL, kind TEXT NOT NULL, hash TEXT, size INTEGER NOT NULL, UNIQUE(tag,path)); + CREATE INDEX IF NOT EXISTS snapshots_path ON snapshots(path);").map_err(err)?; + let seen = { + let mut statement = db.prepare("SELECT tag,path FROM snapshots").map_err(err)?; + statement + .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .map_err(err)? + .collect::>() + .map_err(err)? + }; + Ok(Arc::new(Self { + root, + storage: Arc::new(Mutex::new(Storage { + db, + directory: directory.into(), + })), + seen: Mutex::new(seen), + gate: tokio::sync::Mutex::new(()), + paths: Mutex::new(Vec::new()), + })) + } + pub fn layer(self: &Arc, tag: Tag) -> Arc { + Arc::new(layer::View { + journal: self.clone(), + tag, + }) + } + async fn db( + &self, + f: impl FnOnce(&mut Storage) -> Result + Send + 'static, + ) -> Result { + let storage = self.storage.clone(); + tokio::task::spawn_blocking(move || f(&mut storage.lock().unwrap())) + .await + .map_err(err)? + } + fn excluded(path: &str) -> bool { + ["/proc", "/run", "/tmp"] + .iter() + .any(|base| under(path, base)) + } + /// Caller holds the mutation gate until the actual mutation completes. + async fn capture(&self, tag: &Tag, path: &str) -> Result<()> { + if Self::excluded(path) + || self + .seen + .lock() + .unwrap() + .contains(&(tag.id.clone(), path.into())) + { + return Ok(()); + } + let (kind, bytes, size) = match self.root.stat(path, false).await { + Err(e) if e.kind() == io::ErrorKind::NotFound => ("absent", None, 0), + Err(e) => return Err(e), + Ok(stat) if stat.directory => ("directory", None, 0), + Ok(stat) if stat.symlink || stat.size > MAX_FILE => ("not_undoable", None, stat.size), + Ok(stat) => { + let file = self.root.open(path, Open::default()).await?; + let mut bytes = Vec::new(); + while (bytes.len() as u64) < stat.size { + let chunk = file + .read(bytes.len() as u64, (stat.size as usize) - bytes.len()) + .await?; + if chunk.is_empty() { + return Err(io::ErrorKind::UnexpectedEof.into()); + } + bytes.extend(chunk); + } + ("file", Some(bytes), stat.size) + } + }; + let (kind, bytes) = if bytes.as_ref().is_some_and(|b| b.len() as u64 > MAX_FILE) { + ("not_undoable", None) + } else { + (kind, bytes) + }; + let tag = tag.clone(); + let key = (tag.id.clone(), path.to_owned()); + let path = path.to_owned(); + self.db(move |s| { + let hash = if let Some(bytes) = bytes { + let hash = format!("{:x}", Sha256::digest(&bytes)); + let blob = s.directory.join("blobs").join(&hash); + if !blob.exists() { std::fs::write(blob, bytes)?; } + Some(hash) + } else { None }; + let transaction = s.db.transaction().map_err(err)?; + transaction.execute("INSERT OR IGNORE INTO tags(id,pid,program,conversations,created) VALUES (?1,?2,?3,?4,?5)", params![tag.id, tag.pid, tag.program, serde_json::to_string(&tag.conversations).map_err(err)?, now() as i64]).map_err(err)?; + transaction.execute("INSERT OR IGNORE INTO snapshots(tag,path,kind,hash,size) VALUES (?1,?2,?3,?4,?5)", params![tag.id,path,kind,hash,size as i64]).map_err(err)?; + transaction.commit().map_err(err)?; + Ok(()) + }).await?; + self.seen.lock().unwrap().insert(key); + Ok(()) + } + pub async fn list(&self, conversation: Option) -> Result> { + self.db(move |s| { + let mut statement = s.db.prepare("SELECT t.entry,t.id,t.pid,t.program,t.conversations,t.created,COUNT(s.entry) FROM tags t JOIN snapshots s ON s.tag=t.id GROUP BY t.id ORDER BY t.entry DESC").map_err(err)?; + let entries = statement.query_map([], |r| Ok(Entry { entry:r.get(0)?, tag:r.get(1)?, pid:r.get(2)?, program:r.get(3)?, conversations:serde_json::from_str(&r.get::<_,String>(4)?).unwrap_or_default(), created:r.get::<_, i64>(5)? as u64, paths:r.get::<_, i64>(6)? as usize })).map_err(err)?.collect::>>().map_err(err)?; + Ok(entries.into_iter().filter(|e| conversation.as_ref().is_none_or(|c| e.conversations.contains(c))).collect()) + }).await + } + pub async fn show(&self, tag: String) -> Result> { + self.db(move |s| snapshots(&s.db, Some(&tag))).await + } + /// Drop old tags until both retention limits hold. Content accounting deduplicates hashes. + pub async fn retain(&self) -> Result<()> { + let _gate = self.gate.lock().await; + let remaining = self.db(|s| { + s.db.execute("DELETE FROM tags WHERE created < ?1", [now().saturating_sub(MAX_AGE) as i64]).map_err(err)?; + loop { + let size: u64 = s.db.query_row("SELECT COALESCE(SUM(size),0) FROM (SELECT hash,MAX(size) size FROM snapshots WHERE hash IS NOT NULL GROUP BY hash)", [], |r| Ok(r.get::<_, i64>(0)? as u64)).map_err(err)?; + if size <= MAX_BYTES { break; } + s.db.execute("DELETE FROM tags WHERE entry=(SELECT MIN(entry) FROM tags)", []).map_err(err)?; + } + let hashes = { let mut q = s.db.prepare("SELECT DISTINCT hash FROM snapshots WHERE hash IS NOT NULL").map_err(err)?; q.query_map([], |r| r.get::<_,String>(0)).map_err(err)?.collect::>>().map_err(err)? }; + for file in std::fs::read_dir(s.directory.join("blobs"))? { let file=file?; if !hashes.contains(&file.file_name().to_string_lossy().into_owned()) { std::fs::remove_file(file.path())?; } } + let mut q=s.db.prepare("SELECT tag,path FROM snapshots").map_err(err)?; + q.query_map([], |r| Ok((r.get(0)?,r.get(1)?))).map_err(err)?.collect::>>().map_err(err) + }).await?; + *self.seen.lock().unwrap() = remaining; + Ok(()) + } +} +fn snapshots(db: &Connection, tag: Option<&str>) -> Result> { + let mut q = db.prepare("SELECT entry,tag,path,kind,hash,size FROM snapshots WHERE ?1 IS NULL OR tag=?1 ORDER BY entry").map_err(err)?; + q.query_map([tag], |r| { + Ok(Snapshot { + entry: r.get(0)?, + tag: r.get(1)?, + path: r.get(2)?, + kind: r.get(3)?, + hash: r.get(4)?, + size: r.get::<_, i64>(5)? as u64, + }) + }) + .map_err(err)? + .collect::>() + .map_err(err) +} diff --git a/vfs/src/journal/layer.rs b/vfs/src/journal/layer.rs new file mode 100644 index 0000000..7c2669c --- /dev/null +++ b/vfs/src/journal/layer.rs @@ -0,0 +1,202 @@ +//! Tagged mutations and writable descriptors. Read-only calls directly reach the inner layer. +use super::*; +use crate::{Entry as DirEntry, File, Process, Stat, async_trait}; +pub(super) struct View { + pub journal: Arc, + pub tag: Tag, +} +impl View { + async fn mutate( + &self, + paths: Vec, + operation: impl Future> + Send + 'static, + ) -> Result { + let (journal, tag) = (self.journal.clone(), self.tag.clone()); + Process::default() + .operation(async move { + let _gate = journal.gate.lock().await; + for path in paths { + journal.capture(&tag, &path).await?; + } + operation.await + }) + .await + } +} +#[async_trait] +impl Layer for View { + fn observer(&self) -> Option { + self.journal.root.observer() + } + fn journal(&self) -> Option> { + Some(self.journal.clone()) + } + fn tagged(&self, tag: Tag) -> Option> { + Some(self.journal.layer(tag)) + } + async fn stat(&self, p: &str, f: bool) -> Result { + self.journal.root.stat(p, f).await + } + async fn readlink(&self, p: &str) -> Result { + self.journal.root.readlink(p).await + } + async fn list(&self, p: &str) -> Result> { + self.journal.root.list(p).await + } + async fn open(&self, p: &str, o: Open) -> Result> { + if !(o.write || o.create || o.truncate) || Journal::excluded(p) { + return self.journal.root.open(p, o).await; + } + let (journal, tag, path) = (self.journal.clone(), self.tag.clone(), p.to_owned()); + Process::default() + .operation(async move { + let _gate = journal.gate.lock().await; + if o.create || o.truncate { + journal.capture(&tag, &path).await?; + } + let file = journal.root.open(&path, o).await?; + let path = Arc::new(Mutex::new(Some(path))); + let mut paths = journal.paths.lock().unwrap(); + paths.retain(|p| p.strong_count() != 0); + paths.push(Arc::downgrade(&path)); + Ok(Arc::new(Handle { + file, + path, + view: View { + journal: journal.clone(), + tag, + }, + }) as Arc) + }) + .await + } + async fn mkdir(&self, p: &str) -> Result<()> { + let (root, path) = (self.journal.root.clone(), p.to_owned()); + self.mutate(vec![path.clone()], async move { root.mkdir(&path).await }) + .await + } + async fn remove_file(&self, p: &str) -> Result<()> { + self.remove(p, false).await + } + async fn remove_dir(&self, p: &str) -> Result<()> { + self.remove(p, true).await + } + async fn rename(&self, a: &str, b: &str) -> Result<()> { + let (journal, tag, a, b) = ( + self.journal.clone(), + self.tag.clone(), + a.to_owned(), + b.to_owned(), + ); + Process::default() + .operation(async move { + let _gate = journal.gate.lock().await; + let mut pending = vec![(a.clone(), b.clone())]; + while let Some((from, to)) = pending.pop() { + journal.capture(&tag, &from).await?; + journal.capture(&tag, &to).await?; + if journal.root.stat(&from, false).await?.directory { + for entry in journal.root.list(&from).await? { + pending.push(( + format!("{from}/{}", entry.name), + format!("{to}/{}", entry.name), + )); + } + } + } + journal.root.rename(&a, &b).await?; + if a != b { + journal.moved(&a, Some(&b)); + } + Ok(()) + }) + .await + } +} +impl View { + async fn remove(&self, p: &str, directory: bool) -> Result<()> { + let (journal, path) = (self.journal.clone(), p.to_owned()); + self.mutate(vec![path.clone()], async move { + if directory { + journal.root.remove_dir(&path).await?; + } else { + journal.root.remove_file(&path).await?; + } + journal.moved(&path, None); + Ok(()) + }) + .await + } +} +struct Handle { + file: Arc, + path: Arc>>, + view: View, +} +impl Handle { + async fn mutate( + &self, + operation: impl Future> + Send + 'static, + ) -> Result { + let (journal, tag, path) = ( + self.view.journal.clone(), + self.view.tag.clone(), + self.path.clone(), + ); + Process::default() + .operation(async move { + let _gate = journal.gate.lock().await; + let path = path.lock().unwrap().clone(); + if let Some(path) = path { + journal.capture(&tag, &path).await?; + } + operation.await + }) + .await + } +} +#[async_trait] +impl File for Handle { + async fn stat(&self) -> Result { + self.file.stat().await + } + async fn read(&self, o: u64, n: usize) -> Result> { + self.file.read(o, n).await + } + async fn write(&self, o: u64, b: Vec) -> Result { + let f = self.file.clone(); + self.mutate(async move { f.write(o, b).await }).await + } + async fn append(&self, b: Vec) -> Result { + let f = self.file.clone(); + self.mutate(async move { f.append(b).await }).await + } + async fn set_size(&self, n: u64) -> Result<()> { + let f = self.file.clone(); + self.mutate(async move { f.set_size(n).await }).await + } + async fn sync(&self) -> Result<()> { + self.file.sync().await + } + async fn set_times(&self, a: Option, m: Option) -> Result<()> { + let f = self.file.clone(); + self.mutate(async move { f.set_times(a, m).await }).await + } +} +impl Journal { + pub(super) fn moved(&self, from: &str, to: Option<&str>) { + let mut paths = self.paths.lock().unwrap(); + paths.retain(|p| { + let Some(p) = p.upgrade() else { return false }; + let mut path = p.lock().unwrap(); + if let Some(old) = path.as_deref() { + if under(old, from) { + *path = to.map(|to| format!("{to}{}", &old[from.len()..])); + } else if to.is_some_and(|to| under(old, to)) { + *path = None; + } + } + true + }); + } +} diff --git a/vfs/src/journal/restore.rs b/vfs/src/journal/restore.rs new file mode 100644 index 0000000..44d8649 --- /dev/null +++ b/vfs/src/journal/restore.rs @@ -0,0 +1,196 @@ +//! Conflict checks and reversible restoration. The mutation gate covers selection and restore. +use super::*; +use std::collections::{BTreeMap, BTreeSet}; +impl Journal { + pub async fn undo(&self, request: Undo, mut tag: Tag) -> Result { + if request.tag.is_some() == request.since.is_some() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "select one tag or entry", + )); + } + let _gate = self.gate.lock().await; + let (selected, all) = self + .db(move |s| { + let ids: HashSet = if let Some(id) = request.tag { + let exists = + s.db.query_row("SELECT id FROM tags WHERE id=?1", [&id], |r| { + r.get::<_, String>(0) + }) + .optional() + .map_err(err)?; + if exists.is_none() { + return Err(io::ErrorKind::NotFound.into()); + } + HashSet::from([id]) + } else { + let since = request.since.unwrap(); + let exists = + s.db.query_row("SELECT entry FROM tags WHERE entry=?1", [since], |r| { + r.get::<_, i64>(0) + }) + .optional() + .map_err(err)?; + if exists.is_none() { + return Err(io::ErrorKind::NotFound.into()); + } + let mut q = + s.db.prepare("SELECT id FROM tags WHERE entry>=?1") + .map_err(err)?; + q.query_map([since], |r| r.get(0)) + .map_err(err)? + .collect::>() + .map_err(err)? + }; + Ok((ids, snapshots(&s.db, None)?)) + }) + .await?; + let mut target = BTreeMap::new(); + for snapshot in &all { + if selected.contains(&snapshot.tag) { + target + .entry(snapshot.path.clone()) + .or_insert_with(|| snapshot.clone()); + } + } + let mut conflicts = BTreeSet::new(); + for snapshot in target.values() { + for later in &all { + if later.entry > snapshot.entry + && !selected.contains(&later.tag) + && (under(&later.path, &snapshot.path) || under(&snapshot.path, &later.path)) + { + conflicts.insert(later.path.clone()); + } + } + } + let paths = target.values().cloned().collect::>(); + let missing = self + .db(move |s| { + Ok(paths + .iter() + .filter(|p| { + p.kind == "not_undoable" + || p.hash + .as_ref() + .is_some_and(|h| !s.directory.join("blobs").join(h).exists()) + }) + .map(|p| p.path.clone()) + .collect::>()) + }) + .await?; + let mut outcome = Outcome { + tag: None, + conflicts: conflicts.into_iter().collect(), + skipped: missing, + restored: Vec::new(), + }; + if !request.force && (!outcome.conflicts.is_empty() || !outcome.skipped.is_empty()) { + return Ok(outcome); + } + tag.id = format!("undo:{}", tag.id); + target.retain(|path, _| !outcome.skipped.contains(path)); + if target.is_empty() { + return Ok(outcome); + } + // Capture every affected current descendant before force removes a directory. + for snapshot in target.values() { + let mut pending = vec![snapshot.path.clone()]; + while let Some(path) = pending.pop() { + self.capture(&tag, &path).await?; + if self + .root + .stat(&path, false) + .await + .is_ok_and(|s| s.directory) + { + for entry in self.root.list(&path).await? { + pending.push(format!("{path}/{}", entry.name)); + } + } + } + } + let mut paths = target.into_values().collect::>(); + // Remove unwanted nodes deepest first, then create parents before children. + paths.sort_by_key(|s| std::cmp::Reverse(s.path.matches('/').count())); + for snapshot in &paths { + if snapshot.kind == "absent" { + self.remove_tree(&snapshot.path, &outcome.skipped).await?; + } else if let Ok(stat) = self.root.stat(&snapshot.path, false).await + && (stat.directory != (snapshot.kind == "directory") || stat.symlink) + { + self.remove_tree(&snapshot.path, &outcome.skipped).await?; + } + } + paths.sort_by_key(|s| s.path.matches('/').count()); + for snapshot in &paths { + match snapshot.kind.as_str() { + "directory" => { + if self.root.stat(&snapshot.path, false).await.is_err() { + self.root.mkdir(&snapshot.path).await?; + } + } + "file" => { + let hash = snapshot.hash.clone().unwrap(); + let bytes = self + .db(move |s| std::fs::read(s.directory.join("blobs").join(hash))) + .await?; + let file = self + .root + .open( + &snapshot.path, + Open { + write: true, + create: true, + truncate: true, + ..Default::default() + }, + ) + .await?; + let mut offset = 0; + while offset < bytes.len() { + let n = file.write(offset as u64, bytes[offset..].to_vec()).await?; + if n == 0 { + return Err(io::ErrorKind::WriteZero.into()); + } + offset += n; + } + } + _ => {} + } + outcome.restored.push(snapshot.path.clone()); + } + outcome.tag = Some(tag.id); + Ok(outcome) + } + async fn remove_tree(&self, path: &str, skipped: &[String]) -> Result<()> { + let mut pending = vec![(path.to_owned(), false)]; + while let Some((path, visited)) = pending.pop() { + if skipped.iter().any(|skip| under(&path, skip)) { + continue; + } + let stat = match self.root.stat(&path, false).await { + Ok(s) => s, + Err(e) if e.kind() == io::ErrorKind::NotFound => continue, + Err(e) => return Err(e), + }; + if stat.directory && !visited { + pending.push((path.clone(), true)); + for entry in self.root.list(&path).await? { + pending.push((format!("{path}/{}", entry.name), false)); + } + } else { + if stat.directory { + if skipped.iter().any(|skip| under(skip, &path)) { + continue; + } + self.root.remove_dir(&path).await?; + } else { + self.root.remove_file(&path).await?; + } + self.moved(&path, None); + } + } + Ok(()) + } +} diff --git a/vfs/src/journal/tests.rs b/vfs/src/journal/tests.rs new file mode 100644 index 0000000..525557d --- /dev/null +++ b/vfs/src/journal/tests.rs @@ -0,0 +1,193 @@ +//! Journal persistence, first-change deduplication, exclusions, and unavailable snapshots. +use super::*; +use crate::{Host, Process}; +#[tokio::test] +async fn persistence_dedup_exclusions_and_missing_blob() { + let temp = tempfile::tempdir().unwrap(); + let root_dir = temp.path().join("world"); + std::fs::create_dir_all(root_dir.join("tmp")).unwrap(); + std::fs::write(root_dir.join("a"), b"same").unwrap(); + std::fs::write(root_dir.join("b"), b"same").unwrap(); + let root: Arc = Arc::new(Host::new(&root_dir, Process::default()).unwrap()); + let directory = temp.path().join("undo"); + let journal = Journal::open(&directory, root.clone()).unwrap(); + let tag = Tag::process(1, "/test".into(), vec!["c".into()]).unwrap(); + let view = journal.layer(tag.clone()); + for path in ["/a", "/b", "/tmp/excluded"] { + let file = view + .open( + path, + Open { + write: true, + create: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, b"changed".to_vec()).await.unwrap(); + file.set_size(2).await.unwrap(); + } + assert_eq!(journal.show(tag.id.clone()).await.unwrap().len(), 2); + assert_eq!( + std::fs::read_dir(directory.join("blobs")).unwrap().count(), + 1 + ); + assert_eq!(journal.list(Some("c".into())).await.unwrap().len(), 1); + drop(view); + drop(journal); + let journal = Journal::open(&directory, root).unwrap(); + let blob = std::fs::read_dir(directory.join("blobs")) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + std::fs::remove_file(blob).unwrap(); + let request = Undo { + tag: Some(tag.id), + ..Default::default() + }; + let blocked = journal + .undo( + request.clone(), + Tag::process(2, "/undo".into(), vec![]).unwrap(), + ) + .await + .unwrap(); + assert_eq!(blocked.skipped, ["/a", "/b"]); + assert!(blocked.tag.is_none()); + assert_eq!(std::fs::read(root_dir.join("a")).unwrap(), b"ch"); + let forced = journal + .undo( + Undo { + force: true, + ..request + }, + Tag::process(2, "/undo".into(), vec![]).unwrap(), + ) + .await + .unwrap(); + assert_eq!(forced.skipped, ["/a", "/b"]); + assert!(forced.restored.is_empty()); +} + +#[tokio::test] +async fn retention_removes_oldest_tags_and_unreferenced_content() { + let temp = tempfile::tempdir().unwrap(); + let root: Arc = Arc::new(crate::Memory::default()); + let journal = Journal::open(temp.path(), root).unwrap(); + let mut ids = Vec::new(); + for bytes in [b"old".to_vec(), b"new".to_vec()] { + let tag = Tag::process(1, "/test".into(), vec![]).unwrap(); + let view = journal.layer(tag.clone()); + view.open( + "/file", + Open { + write: true, + create: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap() + .write(0, bytes) + .await + .unwrap(); + ids.push(tag.id); + } + let expired = ids[0].clone(); + journal + .db(move |s| { + s.db.execute("UPDATE tags SET created=0 WHERE id=?1", [expired]) + .map_err(err)?; + Ok(()) + }) + .await + .unwrap(); + journal.retain().await.unwrap(); + assert!(journal.show(ids[0].clone()).await.unwrap().is_empty()); + assert_eq!(journal.list(None).await.unwrap().len(), 1); + // Use index accounting to test the size limit without a multi-GiB fixture. + journal + .db(|s| { + s.db.execute( + "UPDATE snapshots SET size=?1 WHERE hash IS NOT NULL", + [(MAX_BYTES + 1) as i64], + ) + .map_err(err)?; + Ok(()) + }) + .await + .unwrap(); + journal.retain().await.unwrap(); + assert!(journal.list(None).await.unwrap().is_empty()); + assert_eq!( + std::fs::read_dir(temp.path().join("blobs")) + .unwrap() + .count(), + 0 + ); +} + +#[tokio::test] +async fn force_preserves_unavailable_descendants() { + let temp = tempfile::tempdir().unwrap(); + let directory = temp.path().join("world"); + std::fs::create_dir_all(&directory).unwrap(); + let root: Arc = Arc::new(Host::new(&directory, Process::default()).unwrap()); + let journal = Journal::open(&temp.path().join("undo"), root).unwrap(); + let a = Tag::process(1, "/a".into(), vec![]).unwrap(); + journal.layer(a.clone()).mkdir("/new").await.unwrap(); + let entry = journal.list(None).await.unwrap()[0].entry; + // A sparse fixture exercises the size cutoff without allocating its content. + std::fs::File::create(directory.join("new/large")) + .unwrap() + .set_len(MAX_FILE + 1) + .unwrap(); + std::fs::write(directory.join("new/small"), b"old").unwrap(); + let b = journal.layer(Tag::process(2, "/b".into(), vec![]).unwrap()); + for path in ["/new/large", "/new/small"] { + b.open( + path, + Open { + write: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap() + .write(0, b"new".to_vec()) + .await + .unwrap(); + } + let request = Undo { + since: Some(entry), + ..Default::default() + }; + let blocked = journal + .undo( + request.clone(), + Tag::process(3, "/undo".into(), vec![]).unwrap(), + ) + .await + .unwrap(); + assert_eq!(blocked.skipped, ["/new/large"]); + assert!(blocked.tag.is_none()); + let outcome = journal + .undo( + Undo { + force: true, + ..request + }, + Tag::process(3, "/undo".into(), vec![]).unwrap(), + ) + .await + .unwrap(); + assert_eq!(outcome.skipped, ["/new/large"]); + assert_eq!(std::fs::read(directory.join("new/large")).unwrap(), b"new"); + assert_eq!(std::fs::read(directory.join("new/small")).unwrap(), b"old"); +} diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index 0902c87..4ef66b0 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -3,6 +3,7 @@ pub mod adapter; pub mod generated; pub mod host; +pub mod journal; pub mod memory; pub mod mounts; pub mod observe; @@ -67,6 +68,14 @@ pub trait File: Send + Sync { /// Implementations must retain open file identity and return portable IO errors. #[async_trait] pub trait Layer: Send + Sync { + /// Shared undo journal, when this namespace has one. + fn journal(&self) -> Option> { + None + } + /// Select a mutation tag without adding another journal layer. + fn tagged(&self, _tag: journal::Tag) -> Option> { + None + } /// Shared observation of mutations in this namespace, when available. fn observer(&self) -> Option { None