From 4b09b8f8fec0c9146d7465234e999634a7a71a18 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 26 Apr 2026 14:53:03 -0400 Subject: [PATCH 1/2] feat(server): add per-run worker logs Mirror worker tracing into run-scoped runtime/server.log files, expose them through the run logs API, and include run.log in dump exports when available. --- docs-internal/logging-strategy.md | 10 +- docs-internal/run-directory-keys.md | 2 +- docs/api-reference/fabro-api.yaml | 22 ++ docs/changelog/2026-04-26.mdx | 14 + docs/docs.json | 1 + lib/crates/fabro-cli/src/commands/dump.rs | 4 + lib/crates/fabro-cli/src/commands/run/mod.rs | 14 +- lib/crates/fabro-cli/src/logging.rs | 76 ++--- lib/crates/fabro-cli/src/main.rs | 57 +++- lib/crates/fabro-cli/tests/it/cmd/dump.rs | 33 ++- lib/crates/fabro-cli/tests/it/cmd/runner.rs | 25 ++ lib/crates/fabro-client/src/client.rs | 32 +++ lib/crates/fabro-server/src/server.rs | 187 ++++++++++++- lib/crates/fabro-server/src/spawn_env.rs | 3 + lib/crates/fabro-static/src/env_vars.rs | 2 + lib/crates/fabro-util/src/run_log.rs | 260 +++++++----------- lib/crates/fabro-workflow/src/run_dump.rs | 4 + .../src/api/run-internals-api.ts | 74 +++++ 18 files changed, 616 insertions(+), 204 deletions(-) create mode 100644 docs/changelog/2026-04-26.mdx diff --git a/docs-internal/logging-strategy.md b/docs-internal/logging-strategy.md index 07a2a099c..308ae6f17 100644 --- a/docs-internal/logging-strategy.md +++ b/docs-internal/logging-strategy.md @@ -1,9 +1,17 @@ # Fabro Logging Strategy -Fabro uses the `tracing` crate for structured, file-based logging. Logs write to `~/.fabro/logs/{prefix}.YYYY-MM-DD.log` (e.g. `cli.2026-04-06.log`, `server.2026-04-06.log`), rotated daily by `tracing-appender`. Logs older than 7 days are cleaned up on startup. Controlled by the `FABRO_LOG` env var (default: `info`). Logs are for **developers debugging issues after the fact** — they are not user-facing output. +Fabro uses the `tracing` crate for structured, file-based logging. CLI logs write to `~/.fabro/logs/cli.YYYY-MM-DD.log`, rotated daily by `tracing-appender`; logs older than 7 days are cleaned up on startup. The server writes one main log at `/logs/server.log`, and worker subprocesses append their tracing events to that same file. + +Each worker also writes its tracing events to the run-scoped log at `/runtime/server.log`. This per-run file is worker tracing only: parent-side scheduling/cancel/delete events stay in the main server log, and unstructured worker stderr is still drained by the parent into `/logs/server.log`. + +Log level is controlled by the `FABRO_LOG` env var (default: `info`). Server `[server.logging] level = "debug"` is propagated to worker subprocesses when `FABRO_LOG` is not already set in the parent process. Logs are for **developers debugging issues after the fact** — they are not user-facing output. Production runs at INFO level. INFO should be low-volume and high-signal — the summary of what happened. When something goes wrong, developers enable `FABRO_LOG=debug` to get the full picture. DEBUG can be as verbose as needed since it's only turned on temporarily. +## File Appenders + +Fixed server and per-run logs use a per-event-buffered writer opened with `O_APPEND`. Each tracing event buffers formatting writes in memory, then flushes that event to the shared file under a mutex with one `write_all()` call. This is intended to keep normal tracing-sized lines contiguous across concurrent tasks and worker processes; tests validate the event-size range used by Fabro, but the code does not claim a strict syscall-level atomicity guarantee for all possible filesystems and buffer sizes. + ## When to Log **Log at INFO (always on in production):** diff --git a/docs-internal/run-directory-keys.md b/docs-internal/run-directory-keys.md index 990e57bfe..db534a51f 100644 --- a/docs-internal/run-directory-keys.md +++ b/docs-internal/run-directory-keys.md @@ -24,8 +24,8 @@ These paths are local runtime state, not canonical event projections. | Path | Purpose | |---|---| | `worktree/` | Git worktree used by checkpointed runs | +| `runtime/server.log` | Worker tracing log for the run | | `runtime/blobs/` | Materialized local blob payloads for file-backed `fabro+blob://` references | -| `runtime/worker.stderr.log` | Server-managed worker stderr capture | | `nodes/{manager_node}_{visit}/child/` | Nested scratch root for manager-loop child workflows | ## Reconstructed / Exported Files diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index ffd692e62..2e5d8078c 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -1050,6 +1050,28 @@ paths: schema: $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/logs: + get: + operationId: getRunLogs + tags: [Run Internals] + summary: Get Run Logs + description: Returns the worker tracing log for a run when it is available. + parameters: + - $ref: "#/components/parameters/RunId" + responses: + "200": + description: Per-run worker tracing log + content: + text/plain; charset=utf-8: + schema: + type: string + "404": + description: Run not found, or no run log has been written yet + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/pull_request: post: operationId: createRunPullRequest diff --git a/docs/changelog/2026-04-26.mdx b/docs/changelog/2026-04-26.mdx new file mode 100644 index 000000000..45649b843 --- /dev/null +++ b/docs/changelog/2026-04-26.mdx @@ -0,0 +1,14 @@ +--- +title: "Per-run worker tracing logs" +date: "2026-04-26" +--- + +## Run logs + +Each server-dispatched run now gets its own worker tracing log at `runtime/server.log`. The server also exposes the log through `GET /api/v1/runs/{id}/logs`, so tools can fetch execution-time tracing for a single run without grepping the process-wide server log. + +`fabro dump` now includes the same log as `run.log` when it is available. Dumps still succeed for older runs or submitted runs that do not have a worker log yet. + +## Operators + +Worker subprocesses continue writing to the main `/logs/server.log`, and also mirror worker tracing into the per-run file. `FABRO_LOG` still controls log verbosity; if a server config sets `[server.logging] level`, that level is now propagated to workers unless the parent process already set `FABRO_LOG`. diff --git a/docs/docs.json b/docs/docs.json index 67e18331a..0516c2af7 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -252,6 +252,7 @@ "group": "April 2026", "icon": "clock-rotate-left", "pages": [ + "changelog/2026-04-26", "changelog/2026-04-25", "changelog/2026-04-24", "changelog/2026-04-23", diff --git a/lib/crates/fabro-cli/src/commands/dump.rs b/lib/crates/fabro-cli/src/commands/dump.rs index b6920c3ef..a9515a1c5 100644 --- a/lib/crates/fabro-cli/src/commands/dump.rs +++ b/lib/crates/fabro-cli/src/commands/dump.rs @@ -82,6 +82,10 @@ async fn write_run_dump( let events = client.list_run_events(run_id, None, None).await?; let mut dump = RunDump::from_store_state_and_events(state, &events)?; + if let Some(log) = client.get_run_logs(run_id).await? { + dump.add_file_bytes("run.log", log.into_bytes()); + } + dump.hydrate_referenced_blobs_with_reader(|blob_id| { Box::pin(async move { client.read_run_blob(run_id, &blob_id).await }) }) diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 48dedee9b..e17859b8c 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -1,5 +1,6 @@ use anyhow::{Result, anyhow}; use fabro_util::terminal::Styles; +use tracing::Instrument as _; use crate::args::{AttachArgs, RunCommands, RunWorkerArgs, StartArgs}; use crate::command_context::CommandContext; @@ -89,14 +90,11 @@ pub(crate) async fn dispatch( .ok_or_else(|| { anyhow!("FABRO_WORKER_TOKEN is required for worker subprocess auth") })?; - Box::pin(runner::execute( - run_id, - server, - storage_dir, - run_dir, - mode, - &worker_token, - )) + let run_span = tracing::info_span!("run", run_id = %run_id); + Box::pin( + runner::execute(run_id, server, storage_dir, run_dir, mode, &worker_token) + .instrument(run_span), + ) .await } RunCommands::Diff(args) => diff::run(args, base_ctx).await, diff --git a/lib/crates/fabro-cli/src/logging.rs b/lib/crates/fabro-cli/src/logging.rs index d0ec8b9b1..2a46d4406 100644 --- a/lib/crates/fabro-cli/src/logging.rs +++ b/lib/crates/fabro-cli/src/logging.rs @@ -2,11 +2,11 @@ clippy::disallowed_methods, reason = "CLI logging setup: sync directory scan during startup" )] -use std::fs::{File, OpenOptions}; use std::path::Path; use anyhow::{Context, Result}; -use fabro_util::run_log; +use fabro_static::EnvVars; +use fabro_util::run_log::BufferedFileAppender; use tracing_appender::rolling; use tracing_subscriber::fmt::writer::MakeWriter; use tracing_subscriber::layer::SubscriberExt; @@ -18,7 +18,13 @@ const LOG_RETENTION_DAYS: u32 = 7; #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum InternalLogSink { Cli, - Server { path: std::path::PathBuf }, + Server { + path: std::path::PathBuf, + }, + Worker { + server_log_path: std::path::PathBuf, + per_run_log_path: std::path::PathBuf, + }, } pub(crate) fn init_tracing( @@ -31,8 +37,8 @@ pub(crate) fn init_tracing( } else { config_log_level.unwrap_or("info") }; - let filter = - EnvFilter::try_from_env("FABRO_LOG").unwrap_or_else(|_| EnvFilter::new(default_level)); + let filter = EnvFilter::try_from_env(EnvVars::FABRO_LOG) + .unwrap_or_else(|_| EnvFilter::new(default_level)); match sink { InternalLogSink::Cli => { @@ -53,7 +59,17 @@ pub(crate) fn init_tracing( init_subscriber(filter, file_appender); } InternalLogSink::Server { path } => { - init_subscriber(filter, FixedFileAppender::open(path)?); + init_subscriber(filter, open_buffered_appender(path)?); + } + InternalLogSink::Worker { + server_log_path, + per_run_log_path, + } => { + init_worker_subscriber( + filter, + open_buffered_appender(server_log_path)?, + open_buffered_appender(per_run_log_path)?, + ); } } @@ -96,8 +112,6 @@ fn init_subscriber(filter: EnvFilter, file_writer: W) where W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, { - let run_log_writer = run_log::init(); - tracing_subscriber::registry() .with(filter) .with( @@ -106,35 +120,35 @@ where .with_target(true) .with_ansi(false), ) + .init(); +} + +fn init_worker_subscriber( + filter: EnvFilter, + server_writer: ServerWriter, + run_writer: RunWriter, +) where + ServerWriter: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, + RunWriter: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, +{ + tracing_subscriber::registry() + .with(filter) .with( fmt::layer() - .with_writer(run_log_writer) + .with_writer(server_writer) + .with_target(true) + .with_ansi(false), + ) + .with( + fmt::layer() + .with_writer(run_writer) .with_target(true) .with_ansi(false), ) .init(); } -struct FixedFileAppender { - file: File, -} - -impl FixedFileAppender { - fn open(path: &Path) -> Result { - let file = OpenOptions::new() - .append(true) - .open(path) - .with_context(|| format!("Failed to open server log file: {}", path.display()))?; - Ok(Self { file }) - } -} - -impl<'writer> MakeWriter<'writer> for FixedFileAppender { - type Writer = File; - - fn make_writer(&'writer self) -> Self::Writer { - self.file - .try_clone() - .expect("fixed log file handle should be cloneable") - } +fn open_buffered_appender(path: &Path) -> Result { + BufferedFileAppender::open(path) + .with_context(|| format!("Failed to open log file: {}", path.display())) } diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 949ab7b90..54819264e 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -209,7 +209,9 @@ async fn main_inner(worker_token: Option) -> (String, Result<()>) { let config_log_level = match &pre_tracing_bootstrap.sink { logging::InternalLogSink::Cli => base_ctx.user_settings().cli.logging.level.clone(), - logging::InternalLogSink::Server { .. } => pre_tracing_bootstrap.config_log_level.clone(), + logging::InternalLogSink::Server { .. } | logging::InternalLogSink::Worker { .. } => { + pre_tracing_bootstrap.config_log_level.clone() + } }; if let Err(err) = logging::init_tracing( globals.debug, @@ -449,6 +451,9 @@ async fn pre_tracing_bootstrap(command: &Commands) -> Result { + prepare_run_worker_bootstrap(args.storage_dir.as_deref(), &args.run_dir) + } _ => Ok(PreTracingBootstrap::cli()), } } @@ -476,6 +481,23 @@ async fn prepare_server_bootstrap( }) } +fn prepare_run_worker_bootstrap( + storage_dir: Option<&std::path::Path>, + run_dir: &std::path::Path, +) -> Result { + let local_config = local_server::LocalServerConfig::load_with_storage_dir(storage_dir)?; + let runtime_directory = fabro_config::RuntimeDirectory::new(local_config.storage_dir()); + + Ok(PreTracingBootstrap { + sink: logging::InternalLogSink::Worker { + server_log_path: runtime_directory.log_path(), + per_run_log_path: run_dir.join("runtime").join("server.log"), + }, + config_log_level: None, + foreground_server_log_bootstrap: None, + }) +} + #[cfg(test)] #[expect( clippy::disallowed_methods, @@ -731,6 +753,39 @@ level = "warn" assert!(bootstrap.foreground_server_log_bootstrap.is_none()); } + #[test] + fn pre_tracing_bootstrap_uses_worker_sink_for_run_worker() { + let storage_dir = tempfile::tempdir().unwrap(); + let run_dir = tempfile::tempdir().unwrap(); + let cli = Cli::try_parse_from([ + "fabro", + "__run-worker", + "--server", + "/tmp/fabro.sock", + "--storage-dir", + storage_dir.path().to_str().unwrap(), + "--run-dir", + run_dir.path().to_str().unwrap(), + "--run-id", + "01ARZ3NDEKTSV4RRFFQ69G5FAV", + "--mode", + "start", + ]) + .expect("should parse"); + let command = cli.command.as_deref().unwrap(); + + let bootstrap = runtime() + .block_on(pre_tracing_bootstrap(command)) + .expect("bootstrap should resolve"); + + assert_eq!(bootstrap.sink, logging::InternalLogSink::Worker { + server_log_path: storage_dir.path().join("logs").join("server.log"), + per_run_log_path: run_dir.path().join("runtime").join("server.log"), + }); + assert!(bootstrap.config_log_level.is_none()); + assert!(bootstrap.foreground_server_log_bootstrap.is_none()); + } + #[test] fn pre_tracing_bootstrap_uses_cli_sink_for_server_start_daemon_wrapper() { let storage_dir = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-cli/tests/it/cmd/dump.rs b/lib/crates/fabro-cli/tests/it/cmd/dump.rs index ee0de5671..9224ab842 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/dump.rs @@ -9,7 +9,9 @@ use std::time::Duration; use fabro_test::{fabro_snapshot, test_context}; use insta::assert_snapshot; -use super::support::{local_dev_token, server_target, setup_completed_dry_run}; +use super::support::{ + local_dev_token, server_target, setup_completed_dry_run, setup_created_dry_run, +}; use crate::support::{LightweightCli, unique_run_id}; #[test] @@ -263,7 +265,7 @@ fn dump_exports_completed_run_snapshot() { success: true exit_code: 0 ----- stdout ----- - Exported 12 files for run [ULID] to [TEMP_DIR]/export + Exported 13 files for run [ULID] to [TEMP_DIR]/export ----- stderr ----- "); @@ -274,6 +276,7 @@ fn dump_exports_completed_run_snapshot() { events.jsonl graph.fabro run.json + run.log stages/exit@1/status.json stages/report@1/response.md stages/report@1/status.json @@ -283,6 +286,32 @@ fn dump_exports_completed_run_snapshot() { "); } +#[test] +fn dump_succeeds_when_run_log_is_missing() { + let context = test_context!(); + let run = setup_created_dry_run(&context); + let output_dir = context.temp_dir.join("export-missing-log"); + + let mut cmd = context.command(); + cmd.args([ + "dump", + "--output", + output_dir.to_str().unwrap(), + &run.run_id, + ]); + let output = cmd.output().expect("dump should execute"); + assert!( + output.status.success(), + "dump failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !output_dir.join("run.log").exists(), + "dump should skip run.log when the server has no run log" + ); +} + #[test] fn dump_rejects_non_empty_output_dir() { let context = test_context!(); diff --git a/lib/crates/fabro-cli/tests/it/cmd/runner.rs b/lib/crates/fabro-cli/tests/it/cmd/runner.rs index 38e9d5245..3f9cfb707 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/runner.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/runner.rs @@ -571,6 +571,31 @@ methods = ["dev-token"] let server_log = std::fs::read_to_string(storage_dir.join("logs/server.log")).unwrap_or_default(); assert_no_worker_env_leak("server log", &server_log); + assert!( + server_log.contains("Workflow run started"), + "main server log should include worker tracing, got:\n{server_log}" + ); + assert!( + server_log.contains(&run_id), + "main server log should include the run id, got:\n{server_log}" + ); + + let run_log_path = run_dir.join("runtime/server.log"); + assert!( + run_log_path.is_file(), + "run log should be written at {}", + run_log_path.display() + ); + let run_log = std::fs::read_to_string(&run_log_path).expect("run log should be readable"); + assert!( + run_log.contains("Workflow run started"), + "per-run log should include worker tracing, got:\n{run_log}" + ); + assert!( + run_log.contains(&run_id), + "per-run log should include the run id, got:\n{run_log}" + ); + assert_no_worker_env_leak("per-run log", &run_log); } #[test] diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index bf37165aa..969af62e8 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -873,6 +873,38 @@ impl Client { convert_type(response.into_inner()) } + #[expect( + clippy::disallowed_types, + reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries." + )] + pub async fn get_run_logs(&self, run_id: &RunId) -> Result> { + let base_url = self.base_url(); + let mut url = fabro_http::Url::parse(&base_url) + .with_context(|| format!("invalid server base URL {base_url}"))?; + url.path_segments_mut() + .map_err(|()| anyhow!("server base URL cannot accept path segments"))? + .extend(["api", "v1", "runs", &run_id.to_string(), "logs"]); + let request_url = url.clone(); + + let response = self + .send_http_response(move |client| { + let url = request_url.clone(); + async move { client.get(url).send().await } + }) + .await?; + match response { + Ok(response) => { + let bytes = response + .bytes() + .await + .context("failed to read run logs response body")?; + Ok(Some(String::from_utf8_lossy(&bytes).into_owned())) + } + Err(failure) if failure.status == fabro_http::StatusCode::NOT_FOUND => Ok(None), + Err(failure) => Err(raw_response_failure_error(&failure)), + } + } + pub async fn create_run_pull_request( &self, run_id: &RunId, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index d5f06b501..eebe155c9 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, HashSet}; +use std::io::ErrorKind; use std::path::PathBuf; use std::process::Stdio; use std::str::FromStr; @@ -107,7 +108,7 @@ use tokio_stream::StreamExt; use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream}; use tower::{ServiceExt, service_fn}; use tower_http::trace::TraceLayer; -use tracing::{debug, error, info, warn}; +use tracing::{Instrument, debug, error, info, warn}; use ulid::Ulid; use crate::auth::{self, GithubEndpoints, auth_translation_middleware, demo_routing_middleware}; @@ -1111,6 +1112,7 @@ fn demo_routes() -> Router> { .route("/runs/{id}/questions", get(demo::get_questions_stub)) .route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub)) .route("/runs/{id}/state", get(not_implemented)) + .route("/runs/{id}/logs", get(not_implemented)) .route( "/runs/{id}/events", get(not_implemented).post(not_implemented), @@ -1193,6 +1195,7 @@ fn real_routes() -> Router> { .route("/runs/{id}/questions", get(get_questions)) .route("/runs/{id}/questions/{qid}/answer", post(submit_answer)) .route("/runs/{id}/state", get(get_run_state)) + .route("/runs/{id}/logs", get(get_run_logs)) .route( "/runs/{id}/pull_request", get(get_run_pull_request).post(create_run_pull_request), @@ -3929,6 +3932,11 @@ fn worker_command( .stderr(Stdio::piped()); apply_worker_env(&mut cmd); + if (state.env_lookup)(EnvVars::FABRO_LOG).is_none() { + if let Some(level) = state.server_settings().server.logging.level.as_deref() { + cmd.env(EnvVars::FABRO_LOG, level); + } + } cmd.env_remove(EnvVars::FABRO_WORKER_TOKEN); cmd.env(EnvVars::FABRO_WORKER_TOKEN, worker_token); @@ -5122,7 +5130,10 @@ pub fn spawn_scheduler(state: Arc) { match run_to_start { Some(id) => { let state_clone = Arc::clone(&state); - tokio::spawn(execute_run(state_clone, id)); + tokio::spawn( + execute_run(state_clone, id) + .instrument(tracing::info_span!("run", run_id = %id)), + ); } None => break, } @@ -5261,6 +5272,32 @@ async fn get_run_state( } } +async fn get_run_logs( + AuthorizeRunScoped(id): AuthorizeRunScoped, + State(state): State>, +) -> Response { + if state.store.open_run_reader(&id).await.is_err() { + return ApiError::not_found("Run not found.").into_response(); + } + + let path = Storage::new(state.server_storage_dir()) + .run_scratch(&id) + .runtime_dir() + .join("server.log"); + match fs::read(&path).await { + Ok(bytes) => { + let body = String::from_utf8_lossy(&bytes).into_owned(); + ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], body).into_response() + } + Err(err) if err.kind() == ErrorKind::NotFound => { + ApiError::not_found("Run log not available.").into_response() + } + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } + } +} + #[expect( clippy::disallowed_types, reason = "Pull-request API validates public github.com URLs; these raw URLs are not credential-bearing log output." @@ -8039,6 +8076,29 @@ mod tests { build_router(state, AuthMode::Disabled) } + fn create_app_state_with_isolated_storage() -> Arc { + let storage_dir = std::env::temp_dir().join(format!("fabro-server-test-{}", Ulid::new())); + std::fs::create_dir_all(&storage_dir).expect("test storage dir should be creatable"); + let source = format!( + r#" +_version = 1 + +[server.storage] +root = "{}" + +[server.auth] +methods = ["dev-token"] +"#, + storage_dir.display() + ); + + create_app_state_with_options( + server_settings_from_toml(&source), + manifest_run_defaults_from_toml(&source), + 5, + ) + } + async fn body_json(body: Body) -> serde_json::Value { let bytes = to_bytes(body, usize::MAX).await.unwrap(); serde_json::from_slice(&bytes).unwrap() @@ -9069,6 +9129,35 @@ provider = "invalid-provider" assert_eq!(dev_claims.run_id, dev_token_run_id.to_string()); } + #[cfg(unix)] + #[test] + fn worker_command_sets_fabro_log_from_server_logging_config() { + let storage_dir = tempfile::tempdir().unwrap(); + let state = worker_command_test_state_with_extra_config( + storage_dir.path(), + &["dev-token"], + Some(TEST_DEV_TOKEN), + r#" +[server.logging] +level = "debug" +"#, + ); + let run_id = RunId::new(); + + let cmd = worker_command( + state.as_ref(), + run_id, + RunExecutionMode::Start, + storage_dir.path(), + ) + .unwrap(); + + assert_eq!( + command_env_value(&cmd, EnvVars::FABRO_LOG), + EnvOverride::Set("debug".to_string()) + ); + } + #[test] fn build_app_state_requires_session_secret_for_worker_tokens() { let server_settings = server_settings_from_toml( @@ -9111,6 +9200,15 @@ methods = ["dev-token"] storage_dir: &Path, methods: &[&str], dev_token: Option<&str>, + ) -> Arc { + worker_command_test_state_with_extra_config(storage_dir, methods, dev_token, "") + } + + fn worker_command_test_state_with_extra_config( + storage_dir: &Path, + methods: &[&str], + dev_token: Option<&str>, + extra_config: &str, ) -> Arc { let dev_token = dev_token.map(str::to_owned); std::fs::create_dir_all(storage_dir).unwrap(); @@ -9126,6 +9224,7 @@ methods = [{}] [server.auth.github] allowed_usernames = ["octocat"] +{extra_config} "#, storage_dir.display(), methods @@ -10125,6 +10224,80 @@ slug = "fabro" assert!(body["nodes"].is_object()); } + #[tokio::test] + async fn get_run_logs_returns_per_run_log_file() { + let state = create_app_state_with_isolated_storage(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + let run_id = RunId::new(); + create_durable_run_with_events(&state, run_id, &[workflow_event::Event::RunSubmitted { + definition_blob: None, + }]) + .await; + let log_path = Storage::new(state.server_storage_dir()) + .run_scratch(&run_id) + .runtime_dir() + .join("server.log"); + tokio::fs::create_dir_all(log_path.parent().unwrap()) + .await + .unwrap(); + tokio::fs::write(&log_path, b"worker log line\nsecond line\n") + .await + .unwrap(); + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/logs"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + let content_type = response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let body = response_bytes!(response, StatusCode::OK).await; + + assert_eq!(content_type.as_deref(), Some("text/plain; charset=utf-8")); + assert_eq!(&body[..], b"worker log line\nsecond line\n"); + } + + #[tokio::test] + async fn get_run_logs_returns_not_found_for_missing_run() { + let state = create_app_state_with_isolated_storage(); + let app = build_router(state, AuthMode::Disabled); + let missing_run_id = RunId::new(); + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{missing_run_id}/logs"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::NOT_FOUND).await; + } + + #[tokio::test] + async fn get_run_logs_returns_not_found_when_log_file_is_missing() { + let state = create_app_state_with_isolated_storage(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + let run_id = RunId::new(); + create_durable_run_with_events(&state, run_id, &[workflow_event::Event::RunSubmitted { + definition_blob: None, + }]) + .await; + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/logs"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::NOT_FOUND).await; + } + #[tokio::test] async fn get_run_pull_request_returns_live_detail_from_github() { let github = MockServer::start(); @@ -12946,7 +13119,10 @@ timeout = "30s" let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await; let run_id = run_id_str.parse::().unwrap(); - let runner = tokio::spawn(execute_run(Arc::clone(&state), run_id)); + let runner = tokio::spawn( + execute_run(Arc::clone(&state), run_id) + .instrument(tracing::info_span!("run", run_id = %run_id)), + ); let mut live_status_before_cancel = None; for _ in 0..50 { live_status_before_cancel = { @@ -13042,7 +13218,10 @@ timeout = "30s" let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await; let run_id = run_id_str.parse::().unwrap(); - let runner = tokio::spawn(execute_run(Arc::clone(&state), run_id)); + let runner = tokio::spawn( + execute_run(Arc::clone(&state), run_id) + .instrument(tracing::info_span!("run", run_id = %run_id)), + ); tokio::time::sleep(std::time::Duration::from_millis(50)).await; let req = Request::builder() diff --git a/lib/crates/fabro-server/src/spawn_env.rs b/lib/crates/fabro-server/src/spawn_env.rs index 54ab7bd45..c60e0f719 100644 --- a/lib/crates/fabro-server/src/spawn_env.rs +++ b/lib/crates/fabro-server/src/spawn_env.rs @@ -10,6 +10,7 @@ const WORKER_ENV_ALLOWLIST: &[&str] = &[ EnvVars::USER, EnvVars::RUST_LOG, EnvVars::RUST_BACKTRACE, + EnvVars::FABRO_LOG, EnvVars::FABRO_HOME, EnvVars::FABRO_STORAGE_ROOT, ]; @@ -75,6 +76,7 @@ mod tests { ("TMPDIR".to_string(), "/tmp".to_string()), ("USER".to_string(), "alice".to_string()), ("RUST_LOG".to_string(), "debug".to_string()), + ("FABRO_LOG".to_string(), "debug".to_string()), ("FABRO_HOME".to_string(), "/tmp/fabro-home".to_string()), ( "FABRO_STORAGE_ROOT".to_string(), @@ -102,6 +104,7 @@ mod tests { assert_eq!(actual.get("PATH").map(String::as_str), Some("/bin")); assert_eq!(actual.get("HOME").map(String::as_str), Some("/tmp/home")); + assert_eq!(actual.get("FABRO_LOG").map(String::as_str), Some("debug")); assert_eq!( actual.get("FABRO_DEV_TOKEN").map(String::as_str), Some("fabro_dev_abababababababababababababababababababababababababababababababab") diff --git a/lib/crates/fabro-static/src/env_vars.rs b/lib/crates/fabro-static/src/env_vars.rs index 30e8594a0..5c95a13bc 100644 --- a/lib/crates/fabro-static/src/env_vars.rs +++ b/lib/crates/fabro-static/src/env_vars.rs @@ -19,6 +19,7 @@ impl EnvVars { pub const FABRO_HOME: &'static str = "FABRO_HOME"; pub const FABRO_HTTP_PROXY_POLICY: &'static str = "FABRO_HTTP_PROXY_POLICY"; pub const FABRO_JSON: &'static str = "FABRO_JSON"; + pub const FABRO_LOG: &'static str = "FABRO_LOG"; pub const FABRO_NO_UPGRADE_CHECK: &'static str = "FABRO_NO_UPGRADE_CHECK"; pub const FABRO_QUIET: &'static str = "FABRO_QUIET"; pub const FABRO_SERVER: &'static str = "FABRO_SERVER"; @@ -151,6 +152,7 @@ mod tests { EnvVars::FABRO_HOME, EnvVars::FABRO_HTTP_PROXY_POLICY, EnvVars::FABRO_JSON, + EnvVars::FABRO_LOG, EnvVars::FABRO_NO_UPGRADE_CHECK, EnvVars::FABRO_QUIET, EnvVars::FABRO_SERVER, diff --git a/lib/crates/fabro-util/src/run_log.rs b/lib/crates/fabro-util/src/run_log.rs index 8c1ac3063..17ba17cbc 100644 --- a/lib/crates/fabro-util/src/run_log.rs +++ b/lib/crates/fabro-util/src/run_log.rs @@ -1,72 +1,64 @@ #![expect( clippy::disallowed_types, - reason = "file-backed tracing sink: sync BufWriter is intentional; writes happen on a \ - dedicated per-event guard and are not in an async hot path" + reason = "file-backed tracing sink: sync File is intentional; writes happen on a dedicated \ + per-event guard and are not in an async hot path" )] #![expect( clippy::disallowed_methods, - reason = "sync File::create and read_to_string for the on-disk run-log file; not on Tokio path" + reason = "sync directory creation and OpenOptions for tracing file appender setup; not on \ + Tokio path" )] -use std::io::{self, BufWriter, Write}; +use std::fs::{File, OpenOptions}; +use std::io::{self, Write}; use std::path::Path; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex, OnceLock}; +use std::sync::{Arc, Mutex}; use tracing_subscriber::fmt::MakeWriter; -static RUN_LOG: OnceLock = OnceLock::new(); - -/// A switchable `MakeWriter` that can be activated/deactivated at runtime. -/// -/// When inactive, writes are silently discarded with no allocation. -/// When active, each event is buffered per-guard and flushed atomically on -/// drop. +/// File-backed tracing writer that buffers each event and appends it as one +/// contiguous write under a shared lock. #[derive(Clone, Debug)] -pub struct RunLogWriter { - active: Arc, - file: Arc>>>, +pub struct BufferedFileAppender { + file: Arc>, } -impl RunLogWriter { - fn new() -> Self { - Self { - active: Arc::new(AtomicBool::new(false)), - file: Arc::new(Mutex::new(None)), +impl BufferedFileAppender { + pub fn open(path: &Path) -> io::Result { + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + std::fs::create_dir_all(parent)?; } + let file = OpenOptions::new().append(true).create(true).open(path)?; + Ok(Self { + file: Arc::new(Mutex::new(file)), + }) } } -impl<'a> MakeWriter<'a> for RunLogWriter { - type Writer = RunLogGuard; +impl<'a> MakeWriter<'a> for BufferedFileAppender { + type Writer = BufferedFileGuard; fn make_writer(&'a self) -> Self::Writer { - if self.active.load(Ordering::Relaxed) { - RunLogGuard::Active { - buf: Vec::new(), - file: self.file.clone(), - } - } else { - RunLogGuard::Inactive + BufferedFileGuard { + buf: Vec::new(), + file: self.file.clone(), } } } -/// Per-event write guard. Buffers writes and flushes atomically on drop. -pub enum RunLogGuard { - Inactive, - Active { - buf: Vec, - file: Arc>>>, - }, +/// Per-event write guard. Buffers tracing's write calls and flushes the whole +/// event when the guard is dropped. +pub struct BufferedFileGuard { + buf: Vec, + file: Arc>, } -impl Write for RunLogGuard { +impl Write for BufferedFileGuard { fn write(&mut self, data: &[u8]) -> io::Result { - match self { - Self::Inactive => Ok(data.len()), - Self::Active { buf, .. } => buf.write(data), - } + self.buf.write(data) } fn flush(&mut self) -> io::Result<()> { @@ -74,145 +66,51 @@ impl Write for RunLogGuard { } } -impl Drop for RunLogGuard { +impl Drop for BufferedFileGuard { fn drop(&mut self) { - if let Self::Active { buf, file } = self { - if buf.is_empty() { - return; - } - if let Ok(mut guard) = file.lock() { - if let Some(writer) = guard.as_mut() { - let _ = writer.write_all(buf); - let _ = writer.flush(); - } - } + if self.buf.is_empty() { + return; + } + if let Ok(mut file) = self.file.lock() { + let _ = file.write_all(&self.buf); + let _ = file.flush(); } - } -} - -/// Initialize the global run log writer. Returns a clone for use as a tracing -/// layer writer. -/// -/// Must be called exactly once (typically from logging init). Panics on second -/// call. -pub fn init() -> RunLogWriter { - let writer = RunLogWriter::new(); - let clone = writer.clone(); - RUN_LOG - .set(writer) - .expect("run_log::init() called more than once"); - clone -} - -/// Activate per-run logging, directing tracing output to `path`. -pub fn activate(path: &Path) -> io::Result<()> { - let writer = RUN_LOG - .get() - .expect("run_log::activate() called before init()"); - let file = std::fs::File::create(path)?; - let mut guard = writer.file.lock().expect("run log lock poisoned"); - *guard = Some(BufWriter::new(file)); - writer.active.store(true, Ordering::Release); - Ok(()) -} - -/// Deactivate per-run logging. Flushes and closes the current log file. -pub fn deactivate() { - let Some(writer) = RUN_LOG.get() else { - return; - }; - writer.active.store(false, Ordering::Release); - let mut guard = writer.file.lock().expect("run log lock poisoned"); - if let Some(mut w) = guard.take() { - let _ = w.flush(); } } #[cfg(test)] mod tests { use std::io::Write; + use std::thread; use super::*; - /// Helper: create a standalone `RunLogWriter` (not the global singleton) - /// for isolated tests. - fn test_writer() -> RunLogWriter { - RunLogWriter::new() - } - #[test] - fn inactive_writer_discards_writes() { - let w = test_writer(); - let mut guard = w.make_writer(); - guard.write_all(b"should be discarded").unwrap(); - drop(guard); - // No file, no panic — writes silently dropped - } - - #[test] - fn activate_writes_to_file() { - let w = test_writer(); + fn buffered_file_appender_creates_parent_dir() { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("test.log"); + let path = dir + .path() + .join("missing") + .join("runtime") + .join("server.log"); + let appender = BufferedFileAppender::open(&path).unwrap(); - // Activate - let file = std::fs::File::create(&path).unwrap(); - *w.file.lock().unwrap() = Some(BufWriter::new(file)); - w.active.store(true, Ordering::Release); - - // Write via guard - let mut guard = w.make_writer(); + let mut guard = appender.make_writer(); guard.write_all(b"hello world").unwrap(); drop(guard); + assert!(path.is_file()); let contents = std::fs::read_to_string(&path).unwrap(); assert_eq!(contents, "hello world"); } #[test] - fn deactivate_stops_writing() { - let w = test_writer(); + fn buffered_file_appender_uses_one_contiguous_flush_per_event() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.log"); + let appender = BufferedFileAppender::open(&path).unwrap(); - // Activate - let file = std::fs::File::create(&path).unwrap(); - *w.file.lock().unwrap() = Some(BufWriter::new(file)); - w.active.store(true, Ordering::Release); - - // Write while active - let mut guard = w.make_writer(); - guard.write_all(b"before").unwrap(); - drop(guard); - - // Deactivate - w.active.store(false, Ordering::Release); - if let Some(mut bw) = w.file.lock().unwrap().take() { - let _ = bw.flush(); - } - - // Write while inactive — should be discarded - let mut guard = w.make_writer(); - guard.write_all(b"after").unwrap(); - drop(guard); - - let contents = std::fs::read_to_string(&path).unwrap(); - assert_eq!(contents, "before"); - } - - #[test] - fn atomic_writes_no_interleaving() { - let w = test_writer(); - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("test.log"); - - // Activate - let file = std::fs::File::create(&path).unwrap(); - *w.file.lock().unwrap() = Some(BufWriter::new(file)); - w.active.store(true, Ordering::Release); - - // Multiple write() calls on the same guard should produce one contiguous block - let mut guard = w.make_writer(); + let mut guard = appender.make_writer(); guard.write_all(b"part1").unwrap(); guard.write_all(b"part2").unwrap(); drop(guard); @@ -220,4 +118,54 @@ mod tests { let contents = std::fs::read_to_string(&path).unwrap(); assert_eq!(contents, "part1part2"); } + + #[test] + fn buffered_file_appender_does_not_tear_tracing_sized_lines() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.log"); + let appender = BufferedFileAppender::open(&path).unwrap(); + let thread_count = 8; + let lines_per_thread = 100; + + let handles = (0..thread_count) + .map(|thread_idx| { + let appender = appender.clone(); + thread::spawn(move || { + let marker = (b'a' + u8::try_from(thread_idx).unwrap()) as char; + for line_idx in 0..lines_per_thread { + let prefix = format!("thread-{thread_idx:02}-line-{line_idx:03}:"); + let payload = marker.to_string().repeat(256 - prefix.len() - 1); + let mut guard = appender.make_writer(); + guard + .write_all(format!("{prefix}{payload}\n").as_bytes()) + .unwrap(); + drop(guard); + } + }) + }) + .collect::>(); + + for handle in handles { + handle.join().unwrap(); + } + + let contents = std::fs::read_to_string(&path).unwrap(); + let lines = contents.lines().collect::>(); + assert_eq!(lines.len(), thread_count * lines_per_thread); + + for line in lines { + assert_eq!(line.len(), 255, "line should not be truncated: {line:?}"); + let (prefix, payload) = line.split_once(':').unwrap(); + let thread_idx = prefix + .strip_prefix("thread-") + .and_then(|rest| rest.split_once("-line-")) + .and_then(|(thread_idx, _)| thread_idx.parse::().ok()) + .unwrap(); + let expected_marker = (b'a' + thread_idx) as char; + assert!( + payload.chars().all(|ch| ch == expected_marker), + "line payload was interleaved: {line:?}" + ); + } + } } diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index dbca35622..92aba6731 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -168,6 +168,10 @@ impl RunDump { Ok(()) } + pub fn add_file_bytes(&mut self, path: impl Into, contents: Vec) { + self.entries.push(RunDumpEntry::bytes(path, contents)); + } + pub async fn hydrate_referenced_blobs_with_reader<'a, F>( &mut self, mut read_blob: F, diff --git a/lib/packages/fabro-api-client/src/api/run-internals-api.ts b/lib/packages/fabro-api-client/src/api/run-internals-api.ts index 5d14b5113..67c93b2ca 100644 --- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts +++ b/lib/packages/fabro-api-client/src/api/run-internals-api.ts @@ -140,6 +140,46 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config options: localVarRequestOptions, }; }, + /** + * Returns the worker tracing log for a run when it is available. + * @summary Get Run Logs + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getRunLogs: async (id: string, options: RawAxiosRequestConfig = {}): Promise => { + // verify required parameter 'id' is not null or undefined + assertParamExists('getRunLogs', 'id', id) + const localVarPath = `/api/v1/runs/{id}/logs` + .replace(`{${"id"}}`, encodeURIComponent(String(id))); + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL); + let baseOptions; + if (configuration) { + baseOptions = configuration.baseOptions; + } + + const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options}; + const localVarHeaderParameter = {} as any; + const localVarQueryParameter = {} as any; + + // authentication SessionCookie required + + // authentication BearerAuth required + // http bearer authentication required + await setBearerAuthToObject(localVarHeaderParameter, configuration) + + localVarHeaderParameter['Accept'] = 'text/plain; charset=utf-8,application/json'; + + setSearchParams(localVarUrlObj, localVarQueryParameter); + let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; + localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + }; + }, /** * Returns the internal event-sourced run projection. This is not a stable public contract. * @summary Get Run State @@ -729,6 +769,19 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.attachRunEvents']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, + /** + * Returns the worker tracing log for a run when it is available. + * @summary Get Run Logs + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async getRunLogs(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.getRunLogs(id, options); + const localVarOperationServerIndex = configuration?.serverIndex ?? 0; + const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.getRunLogs']?.[localVarOperationServerIndex]?.url; + return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); + }, /** * Returns the internal event-sourced run projection. This is not a stable public contract. * @summary Get Run State @@ -931,6 +984,16 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b attachRunEvents(id: string, sinceSeq?: number, options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.attachRunEvents(id, sinceSeq, options).then((request) => request(axios, basePath)); }, + /** + * Returns the worker tracing log for a run when it is available. + * @summary Get Run Logs + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getRunLogs(id: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.getRunLogs(id, options).then((request) => request(axios, basePath)); + }, /** * Returns the internal event-sourced run projection. This is not a stable public contract. * @summary Get Run State @@ -1097,6 +1160,17 @@ export class RunInternalsApi extends BaseAPI { return RunInternalsApiFp(this.configuration).attachRunEvents(id, sinceSeq, options).then((request) => request(this.axios, this.basePath)); } + /** + * Returns the worker tracing log for a run when it is available. + * @summary Get Run Logs + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + public getRunLogs(id: string, options?: RawAxiosRequestConfig) { + return RunInternalsApiFp(this.configuration).getRunLogs(id, options).then((request) => request(this.axios, this.basePath)); + } + /** * Returns the internal event-sourced run projection. This is not a stable public contract. * @summary Get Run State From a97b1515937132b5f369897eb178affd9f61b761 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 26 Apr 2026 16:06:23 -0400 Subject: [PATCH 2/2] refactor: simplify per-run logs client and handler Use the generated progenitor builder for client.get_run_logs, return raw bytes end-to-end, and drop the no-op file.flush() in BufferedFileGuard. Co-Authored-By: Claude Opus 4.7 (1M context) --- lib/crates/fabro-cli/src/commands/dump.rs | 2 +- lib/crates/fabro-client/src/client.rs | 47 +++++++++++------------ lib/crates/fabro-server/src/server.rs | 5 +-- lib/crates/fabro-util/src/run_log.rs | 1 - 4 files changed, 24 insertions(+), 31 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/dump.rs b/lib/crates/fabro-cli/src/commands/dump.rs index a9515a1c5..cbd7ae908 100644 --- a/lib/crates/fabro-cli/src/commands/dump.rs +++ b/lib/crates/fabro-cli/src/commands/dump.rs @@ -83,7 +83,7 @@ async fn write_run_dump( let mut dump = RunDump::from_store_state_and_events(state, &events)?; if let Some(log) = client.get_run_logs(run_id).await? { - dump.add_file_bytes("run.log", log.into_bytes()); + dump.add_file_bytes("run.log", log); } dump.hydrate_referenced_blobs_with_reader(|blob_id| { diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index 969af62e8..4af8d04df 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -873,35 +873,32 @@ impl Client { convert_type(response.into_inner()) } - #[expect( - clippy::disallowed_types, - reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries." - )] - pub async fn get_run_logs(&self, run_id: &RunId) -> Result> { - let base_url = self.base_url(); - let mut url = fabro_http::Url::parse(&base_url) - .with_context(|| format!("invalid server base URL {base_url}"))?; - url.path_segments_mut() - .map_err(|()| anyhow!("server base URL cannot accept path segments"))? - .extend(["api", "v1", "runs", &run_id.to_string(), "logs"]); - let request_url = url.clone(); - + pub async fn get_run_logs(&self, run_id: &RunId) -> Result>> { let response = self - .send_http_response(move |client| { - let url = request_url.clone(); - async move { client.get(url).send().await } - }) - .await?; + .current_state() + .client + .get_run_logs() + .id(run_id.to_string()) + .send() + .await; match response { Ok(response) => { - let bytes = response - .bytes() - .await - .context("failed to read run logs response body")?; - Ok(Some(String::from_utf8_lossy(&bytes).into_owned())) + let mut stream = response.into_inner(); + let mut bytes = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|err| anyhow!("{err}"))?; + bytes.extend_from_slice(&chunk); + } + Ok(Some(bytes)) + } + Err(err) => { + let err = map_api_error(err); + if is_not_found_error(&err) { + Ok(None) + } else { + Err(err) + } } - Err(failure) if failure.status == fabro_http::StatusCode::NOT_FOUND => Ok(None), - Err(failure) => Err(raw_response_failure_error(&failure)), } } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index eebe155c9..dc9662e41 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -5285,10 +5285,7 @@ async fn get_run_logs( .runtime_dir() .join("server.log"); match fs::read(&path).await { - Ok(bytes) => { - let body = String::from_utf8_lossy(&bytes).into_owned(); - ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], body).into_response() - } + Ok(bytes) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], bytes).into_response(), Err(err) if err.kind() == ErrorKind::NotFound => { ApiError::not_found("Run log not available.").into_response() } diff --git a/lib/crates/fabro-util/src/run_log.rs b/lib/crates/fabro-util/src/run_log.rs index 17ba17cbc..f2caf2370 100644 --- a/lib/crates/fabro-util/src/run_log.rs +++ b/lib/crates/fabro-util/src/run_log.rs @@ -73,7 +73,6 @@ impl Drop for BufferedFileGuard { } if let Ok(mut file) = self.file.lock() { let _ = file.write_all(&self.buf); - let _ = file.flush(); } } }