diff --git a/PROTOCOL.md b/PROTOCOL.md index 1a39292..5e96d29 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -448,9 +448,23 @@ builds `/plugins/notes/source/`: 3. It builds the release `.wasm` for `wasm32-wasip2`. The tests and the build share the compiled dependencies. 4. It returns `{"wasm": "/state/cargo/builds//notes.wasm", "tests": - {"passed", "failed", "output"}, "warnings"}`. If a step fails, it returns + {"passed", "failed", "output"}, "warnings", "logs"}`. If a step fails, it returns an error with the compiler or test output. +- Before running commands, it creates `/logs/cargo//.stdout.log` + and `.stderr.log` (UTC date, unique build ID). Both test and release phases + append their raw bytes directly to these files, before any parsing, progress + filtering or UTF-8 conversion. This includes successful JSON build output. + `/logs` is a top-level world folder, created if missing, not under `/state`. +- Success results and error -32004 `data` include `logs`: + `{"stdout":{"path":"/logs/cargo/...stdout.log","bytes":123}, + "stderr":{"path":"/logs/cargo/...stderr.log","bytes":45},"total_bytes":168}`. + Counts are raw bytes, not characters. `logs` is null only if log creation + itself fails. Preparation failures otherwise include empty log files. + Inline diagnostics may be filtered or shortened; the files are the full truth. +- Logs remain after failure, total/test timeout, and cancellation. Cancellation + need not return a result; its files remain discoverable in `/logs/cargo/`. + Pruning numeric `/state/cargo/builds/` artifact folders does not prune logs. - Optional `timeout_secs` is an integer of at least 1, default **900** (not null or unlimited). The wall-clock budget starts when the build acquires its mutex, excluding queue time, and covers tests and release compilation together. diff --git a/cli/src/cargo.rs b/cli/src/cargo.rs index af3bf5d..717f6c6 100644 --- a/cli/src/cargo.rs +++ b/cli/src/cargo.rs @@ -9,7 +9,7 @@ use std::{ path::{Path, PathBuf}, sync::Arc, }; -use tokio::{io::AsyncReadExt, process::Command, time::Instant}; +use tokio::{process::Command, time::Instant}; /// Default wall-clock budget after acquiring the build mutex (not queue time). pub const DEFAULT_TIMEOUT_SECS: u64 = 900; @@ -58,9 +58,78 @@ pub struct TestResults { pub failed: u64, pub output: String, } +/// Raw, unfiltered output shared by the test and release phases of one build. +#[derive(Clone, Debug, serde::Serialize)] +pub struct BuildLogs { + pub stdout: StreamLog, + pub stderr: StreamLog, + pub total_bytes: u64, +} +#[derive(Clone, Debug, serde::Serialize)] +pub struct StreamLog { + pub path: String, + pub bytes: u64, +} +/// Persistent descriptors are inherited by cargo, so cancellation cannot lose +/// bytes already written by cargo or its descendants. No buffered writer is used. +struct LogFiles { + stdout: std::fs::File, + stderr: std::fs::File, + stdout_path: PathBuf, + stderr_path: PathBuf, +} +impl LogFiles { + fn create(world: &Path) -> Result { + let directory = world + .join("logs/cargo") + .join(chrono::Utc::now().format("%Y-%m-%d").to_string()); + std::fs::create_dir_all(&directory).context("create cargo log directory")?; + let id = uuid::Uuid::now_v7(); + let stdout_path = directory.join(format!("{id}.stdout.log")); + let stderr_path = directory.join(format!("{id}.stderr.log")); + let create = |path: &Path| { + std::fs::OpenOptions::new() + .create_new(true) + .append(true) + .open(path) + .with_context(|| format!("create {}", path.display())) + }; + Ok(Self { + stdout: create(&stdout_path)?, + stderr: create(&stderr_path)?, + stdout_path, + stderr_path, + }) + } + fn info(&self, world: &Path) -> Result { + let stream = |file: &std::fs::File, path: &Path| -> Result { + Ok(StreamLog { + path: format!("/{}", path.strip_prefix(world)?.display()), + bytes: file.metadata()?.len(), + }) + }; + let stdout = stream(&self.stdout, &self.stdout_path)?; + let stderr = stream(&self.stderr, &self.stderr_path)?; + Ok(BuildLogs { + total_bytes: stdout.bytes + stderr.bytes, + stdout, + stderr, + }) + } + fn read(&self, path: &Path, offset: u64) -> Result> { + use std::io::{Read, Seek, SeekFrom}; + let mut file = + std::fs::File::open(path).with_context(|| format!("read {}", path.display()))?; + file.seek(SeekFrom::Start(offset))?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes)?; + Ok(bytes) + } +} /// A successful build, usable without a running host. #[derive(Debug)] pub struct BuildOutput { + pub logs: Option, pub wasm: PathBuf, pub warnings: String, pub tests: TestResults, @@ -68,6 +137,7 @@ pub struct BuildOutput { /// A failed compiler or test phase, with its complete diagnostics. #[derive(Debug)] pub struct BuildFailure { + pub logs: Option, pub reason: String, pub tests: TestResults, pub output: String, @@ -83,6 +153,7 @@ impl std::error::Error for BuildFailure {} impl From for BuildFailure { fn from(error: anyhow::Error) -> Self { Self { + logs: None, reason: "build preparation failed".into(), output: format!("{error:#}"), tests: TestResults::default(), @@ -179,7 +250,7 @@ impl Cargo { host.register_native(ID,Arc::new(move |_context,method,params| { let builder=builder.clone(); Box::pin(async move { - if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); } + if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings","logs"],"properties":{"logs":{"type":"object","required":["stdout","stderr","total_bytes"],"properties":{"stdout":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"stderr":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"total_bytes":{"type":"integer","minimum":0}}},"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); } if method!="build" { return Err(crate::error(-32601,"cargo method not found")); } let plugin=params["plugin"].as_str().ok_or_else(||crate::error(-32602,"plugin must be a string"))?; let options = BuildOptions { @@ -187,8 +258,8 @@ impl Cargo { test_timeout_secs: seconds(¶ms, "test_timeout_secs", DEFAULT_TEST_TIMEOUT_SECS)?, }; match builder.build_with_options(&builder.world.plugin_dir(plugin).join("source"), options).await { - Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings})), - Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step}})), + Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings,"logs":output.logs})), + Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step,"logs":failure.logs}})), } }) })); @@ -203,6 +274,26 @@ impl Cargo { &self, source: &Path, options: BuildOptions, + ) -> std::result::Result { + let logs = LogFiles::create(&self.world.root)?; + let result = self.build_logged(source, options, &logs).await; + let info = logs.info(&self.world.root)?; + match result { + Ok(mut output) => { + output.logs = Some(info); + Ok(output) + } + Err(mut failure) => { + failure.logs = Some(info); + Err(failure) + } + } + } + async fn build_logged( + &self, + source: &Path, + options: BuildOptions, + logs: &LogFiles, ) -> std::result::Result { if options.timeout_secs == 0 || options.test_timeout_secs == 0 { return Err(anyhow::anyhow!("timeouts must be at least 1 second").into()); @@ -285,7 +376,15 @@ impl Cargo { "target.wasm32-wasip2.runner={}", json!([self.runner, "test-runner"]) )); - let output = run_command(test, started, deadline, "tests", TestResults::default()).await?; + let output = run_command( + test, + started, + deadline, + "tests", + TestResults::default(), + logs, + ) + .await?; let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); let tests = TestResults { @@ -299,6 +398,7 @@ impl Cargo { }; if !output.status.success() { return Err(BuildFailure { + logs: None, reason: "WASI tests failed".into(), output: tests.output.clone(), tests, @@ -318,7 +418,15 @@ impl Cargo { ]) .arg(&target) .current_dir(&work); - let output = run_command(build, started, deadline, "release build", tests.clone()).await?; + let output = run_command( + build, + started, + deadline, + "release build", + tests.clone(), + logs, + ) + .await?; let result = async { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); @@ -352,6 +460,7 @@ impl Cargo { }) .await??; Ok(BuildOutput { + logs: None, wasm: destination, warnings: strip_progress(&diagnostics), tests: tests.clone(), @@ -359,6 +468,7 @@ impl Cargo { } .await; result.map_err(|error: anyhow::Error| BuildFailure { + logs: None, reason: "release build failed".into(), output: format!("{error:#}"), tests, @@ -384,28 +494,23 @@ async fn run_command( deadline: Instant, step: &'static str, mut tests: TestResults, + logs: &LogFiles, ) -> std::result::Result { command .process_group(0) .kill_on_drop(true) .stdin(std::process::Stdio::null()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); + // Child writes go straight to persistent files: no pipe buffers or reader + // tasks can be dropped on cancellation. Parsing happens only afterwards. + .stdout(logs.stdout.try_clone().context("clone stdout log")?) + .stderr(logs.stderr.try_clone().context("clone stderr log")?); + let out_start = logs.stdout.metadata().context("stat stdout log")?.len(); + let err_start = logs.stderr.metadata().context("stat stderr log")?.len(); let mut child = command.spawn().context("start cargo")?; let group = ProcessGroup(child.id().unwrap()); - let mut stdout = child.stdout.take().unwrap(); - let mut stderr = child.stderr.take().unwrap(); - let mut out = Vec::new(); - let mut err = Vec::new(); + let (status, timed_out) = { - let collect = async { - let (status, _, _) = tokio::try_join!( - child.wait(), - stdout.read_to_end(&mut out), - stderr.read_to_end(&mut err) - )?; - Ok::<_, std::io::Error>(status) - }; + let collect = child.wait(); tokio::pin!(collect); tokio::select! { biased; @@ -419,6 +524,10 @@ async fn run_command( } }; let status = status.map_err(anyhow::Error::from)?; + // Also stop any descendants that kept the streams open after cargo exited. + drop(group); + let out = logs.read(&logs.stdout_path, out_start)?; + let err = logs.read(&logs.stderr_path, err_start)?; if timed_out { let output = format!( "{}\n{}", @@ -441,6 +550,7 @@ async fn run_command( tests.output = output.clone(); } return Err(BuildFailure { + logs: None, reason: "timeout".into(), tests, output, @@ -599,6 +709,103 @@ fn strip_progress(output: &str) -> String { mod timeout_tests { use super::*; + /// Streams retain exact bytes, including non-UTF8, progress and JSON; phases append. + #[tokio::test] + async fn raw_streams_append_before_parsing() { + let directory = tempfile::tempdir().unwrap(); + let logs = LogFiles::create(directory.path()).unwrap(); + for _ in 0..2 { + let mut command = Command::new("sh"); + command.args(["-c", "printf '\\377raw\\r'; printf ' Compiling progress\\n{\"reason\":\"compiler-artifact\"}\\n' >&2; exit 3"]); + let now = Instant::now(); + let output = tokio::time::timeout( + Duration::from_secs(5), + run_command( + command, + now, + now + Duration::from_secs(3), + "tests", + TestResults::default(), + &logs, + ), + ) + .await + .unwrap() + .unwrap(); + assert!(!output.status.success()); + assert_eq!(output.stdout, b"\xffraw\r"); + assert!(String::from_utf8_lossy(&output.stderr).contains("Compiling progress")); + } + assert_eq!( + std::fs::read(&logs.stdout_path).unwrap(), + b"\xffraw\r\xffraw\r" + ); + let info = logs.info(directory.path()).unwrap(); + assert_eq!(info.stdout.bytes, 10); + assert_eq!(info.total_bytes, info.stdout.bytes + info.stderr.bytes); + assert!(info.stdout.path.starts_with("/logs/cargo/")); + assert!(!directory.path().join("state/cargo").exists()); + } + + /// Cancellation leaves kernel-written data intact, even without a newline. + #[tokio::test] + async fn cancelled_command_keeps_logs() { + use notify::Watcher; + let directory = tempfile::tempdir().unwrap(); + let logs = LogFiles::create(directory.path()).unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let mut watcher = + notify::recommended_watcher(move |event: notify::Result| { + if let Ok(event) = event { + if event + .paths + .iter() + .any(|path| path.file_name().is_some_and(|name| name == "ready")) + { + let _ = tx.send(()); + } + } + }) + .unwrap(); + watcher + .watch(directory.path(), notify::RecursiveMode::NonRecursive) + .unwrap(); + let mut command = Command::new("sh"); + command.args(["-c", "printf 'partial\\377'; printf 'stderr\\r' >&2; touch \"$1/ready\"; exec sleep 1000", "sh"]).arg(directory.path()); + let now = Instant::now(); + let task = tokio::spawn(async move { + run_command( + command, + now, + now + Duration::from_secs(60), + "tests", + TestResults::default(), + &logs, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .unwrap() + .unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + let paths: Vec<_> = std::fs::read_dir(directory.path().join("logs/cargo")) + .unwrap() + .flat_map(|day| std::fs::read_dir(day.unwrap().path()).unwrap()) + .map(|file| file.unwrap().path()) + .collect(); + assert_eq!(paths.len(), 2); + for path in paths { + let expected: &[u8] = if path.to_string_lossy().ends_with(".stdout.log") { + b"partial\xff" + } else { + b"stderr\r" + }; + assert_eq!(std::fs::read(path).unwrap(), expected); + } + } + /// Deadline errors retain the shared start time, phase, test totals, and output tail. #[tokio::test] async fn timeout_diagnostics() { @@ -607,6 +814,8 @@ mod timeout_tests { "-c", "i=0; while [ $i -lt 100 ]; do echo diagnostic-$i; i=$((i+1)); done; exec sleep 1000", ]); + let directory = tempfile::tempdir().unwrap(); + let logs = LogFiles::create(directory.path()).unwrap(); let now = Instant::now(); let failure = run_command( command, @@ -618,6 +827,7 @@ mod timeout_tests { failed: 0, output: "successful tests".into(), }, + &logs, ) .await .unwrap_err(); @@ -629,5 +839,8 @@ mod timeout_tests { assert_eq!(failure.output.lines().count(), 50); assert!(failure.output.contains("diagnostic-99")); assert!(!failure.output.contains("diagnostic-0\n")); + let raw = std::fs::read_to_string(&logs.stdout_path).unwrap(); + assert_eq!(raw.lines().count(), 100); + assert!(raw.starts_with("diagnostic-0\n")); } } diff --git a/cli/tests/host/cargo.rs b/cli/tests/host/cargo.rs index a99c1b6..e860322 100644 --- a/cli/tests/host/cargo.rs +++ b/cli/tests/host/cargo.rs @@ -73,8 +73,28 @@ async fn build(world: &TestWorld, name: &str) -> Result { .register(&host); let result = host.call("cargo.build", json!({"plugin":name})).await; host.stop().await; + let logs = match &result { + Ok(output) => &output["logs"], + Err(error) => &error["data"]["logs"], + }; + check_logs(world, logs); result } +/// RPC paths point at full raw streams, outside pruned build folders. +fn check_logs(world: &TestWorld, logs: &Value) { + let mut total = 0; + for stream in ["stdout", "stderr"] { + let path = logs[stream]["path"].as_str().expect("log path"); + assert!(path.starts_with("/logs/cargo/")); + assert!(path.ends_with(&format!(".{stream}.log"))); + let bytes = std::fs::metadata(world.0.join(path.trim_start_matches('/'))) + .unwrap() + .len(); + assert_eq!(logs[stream]["bytes"], bytes); + total += bytes; + } + assert_eq!(logs["total_bytes"], total); +} struct Valid { world: TestWorld, result: Value, @@ -206,6 +226,20 @@ async fn build_output_and_warnings() { .unwrap() .contains("fixture build warning") ); + let logs = &valid.result["logs"]; + let raw = |stream: &str| { + std::fs::read_to_string( + valid.world.0.join( + logs[stream]["path"] + .as_str() + .unwrap() + .trim_start_matches('/'), + ), + ) + .unwrap() + }; + assert!(raw("stdout").contains("\"reason\":\"compiler-artifact\"")); + assert!(raw("stderr").contains("Finished ")); let output = valid.result["tests"]["output"].as_str().unwrap(); assert!(output.contains("INTEGRATION_CASE")); assert!(!output.contains("Compiling ")); @@ -277,6 +311,7 @@ async fn native_discovery_and_error_paths() { .await .unwrap_err(); assert_eq!(error["code"], -32004); + check_logs(&world, &error["data"]["logs"]); assert!( error["data"]["output"] .as_str() @@ -721,6 +756,7 @@ async fn build_timeout_and_queued_build() { assert_eq!(error["code"], -32004); assert_eq!(error["message"], "timeout"); assert_eq!(error["data"]["reason"], "timeout"); + check_logs(&world, &error["data"]["logs"]); assert_eq!(error["data"]["step"], "tests"); assert!(error["data"]["elapsed_secs"].as_f64().unwrap() >= 5.0); assert!(!error["data"]["output"].as_str().unwrap().is_empty()); @@ -776,6 +812,7 @@ fn main() { .unwrap_err(); assert_eq!(error["code"], -32004); assert_eq!(error["data"]["reason"], "timeout"); + check_logs(&world, &error["data"]["logs"]); assert_eq!(error["data"]["step"], "release build"); assert_eq!(error["data"]["tests"]["passed"], 1); assert!(error["data"]["elapsed_secs"].as_f64().unwrap() >= 5.0);