scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge remote-tracking branch 'origin/main'

Browse source 
# Conflicts:
#	docs-internal/logging-strategy.md
#	lib/crates/fabro-cli/src/logging.rs
#	lib/crates/fabro-cli/src/main.rs
#	lib/crates/fabro-static/src/env_vars.rs
This commit is contained in:
Bryan Helmkamp 2026-04-26 16:29:54 -04:00
commit 3a6ef48e56
No known key found for this signature in database
17 changed files with 601 additions and 201 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
docs-internal/logging-strategy.md
View file

@ -1,9 +1,17 @@
# Fabro Logging Strategy
Fabro uses the `tracing` crate for structured logging. By default, 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`, and entries older than 7 days are cleaned up on startup. The server log destination is configurable: set `[server.logging].destination = "stdout"` (or `FABRO_LOG_DESTINATION=stdout`) to stream the server log to stdout instead — required for container deployments where the platform captures stdout. The level is controlled by `FABRO_LOG` (default: `info`). Logs are for **developers debugging issues after the fact** — they are not user-facing output.
Fabro uses the `tracing` crate for structured 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. By default the server writes one main log at `<storage>/logs/server.log`, and worker subprocesses append their tracing events to that same file. Set `[server.logging].destination = "stdout"` (or `FABRO_LOG_DESTINATION=stdout`) to stream the server log to stdout instead — required for container deployments where the platform captures stdout.
Each worker also writes its tracing events to the run-scoped log at `<scratch>/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 `<storage>/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):**

2
docs-internal/run-directory-keys.md
View file

@ -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

22
docs/api-reference/fabro-api.yaml
View file

@ -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

14
docs/changelog/2026-04-26.mdx Normal file
View file

@ -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 `<storage>/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`.

1
docs/docs.json
View file

@ -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",

4
lib/crates/fabro-cli/src/commands/dump.rs
View file

@ -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);
}
dump.hydrate_referenced_blobs_with_reader(|blob_id| {
Box::pin(async move { client.read_run_blob(run_id, &blob_id).await })
})

14
lib/crates/fabro-cli/src/commands/run/mod.rs
View file

@ -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,

67
lib/crates/fabro-cli/src/logging.rs
View file

@ -2,13 +2,12 @@
clippy::disallowed_methods,
reason = "CLI logging setup: sync directory scan during startup"
)]
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use fabro_static::EnvVars;
use fabro_types::settings::server::LogDestination;
use fabro_util::run_log;
use fabro_util::run_log::BufferedFileAppender;
use tracing_appender::rolling;
use tracing_subscriber::fmt::writer::MakeWriter;
use tracing_subscriber::layer::SubscriberExt;
@ -25,6 +24,10 @@ pub(crate) enum InternalLogSink {
Server {
log_path: Option<PathBuf>,
},
Worker {
server_log_path: PathBuf,
per_run_log_path: PathBuf,
},
}
pub(crate) fn init_tracing(
@ -61,11 +64,21 @@ pub(crate) fn init_tracing(
InternalLogSink::Server {
log_path: Some(path),
} => {
init_subscriber(filter, FixedFileAppender::open(path)?);
init_subscriber(filter, open_buffered_appender(path)?);
}
InternalLogSink::Server { log_path: None } => {
init_subscriber(filter, std::io::stdout);
}
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)?,
);
}
}
Ok(())
@ -129,8 +142,6 @@ fn init_subscriber<W>(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(
@ -139,35 +150,35 @@ where
.with_target(true)
.with_ansi(false),
)
.init();
}
fn init_worker_subscriber<ServerWriter, RunWriter>(
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<Self> {
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> {
BufferedFileAppender::open(path)
.with_context(|| format!("Failed to open log file: {}", path.display()))
}

57
lib/crates/fabro-cli/src/main.rs
View file

@ -210,7 +210,9 @@ async fn main_inner(worker_token: Option<String>) -> (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,
@ -450,6 +452,9 @@ async fn pre_tracing_bootstrap(command: &Commands) -> Result<PreTracingBootstrap
)
.await
}
Commands::RunCmd(RunCommands::RunWorker(args)) => {
prepare_run_worker_bootstrap(args.storage_dir.as_deref(), &args.run_dir)
}
_ => Ok(PreTracingBootstrap::cli()),
}
}
@ -485,6 +490,23 @@ async fn prepare_server_bootstrap(
})
}
fn prepare_run_worker_bootstrap(
storage_dir: Option<&std::path::Path>,
run_dir: &std::path::Path,
) -> Result<PreTracingBootstrap> {
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,
@ -868,6 +890,39 @@ destination = "{destination}"
);
}
#[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();

33
lib/crates/fabro-cli/tests/it/cmd/dump.rs
View file

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

25
lib/crates/fabro-cli/tests/it/cmd/runner.rs
View file

@ -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]

29
lib/crates/fabro-client/src/client.rs
View file

@ -873,6 +873,35 @@ impl Client {
convert_type(response.into_inner())
}
pub async fn get_run_logs(&self, run_id: &RunId) -> Result<Option<Vec<u8>>> {
let response = self
.current_state()
.client
.get_run_logs()
.id(run_id.to_string())
.send()
.await;
match response {
Ok(response) => {
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)
}
}
}
}
pub async fn create_run_pull_request(
&self,
run_id: &RunId,

184
lib/crates/fabro-server/src/server.rs
View file

@ -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<Arc<AppState>> {
.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<Arc<AppState>> {
.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<AppState>) {
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,29 @@ async fn get_run_state(
}
}
async fn get_run_logs(
AuthorizeRunScoped(id): AuthorizeRunScoped,
State(state): State<Arc<AppState>>,
) -> 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) => ([(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()
}
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 +8073,29 @@ mod tests {
build_router(state, AuthMode::Disabled)
}
fn create_app_state_with_isolated_storage() -> Arc<AppState> {
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 +9126,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 +9197,15 @@ methods = ["dev-token"]
storage_dir: &Path,
methods: &[&str],
dev_token: Option<&str>,
) -> Arc<AppState> {
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<AppState> {
let dev_token = dev_token.map(str::to_owned);
std::fs::create_dir_all(storage_dir).unwrap();
@ -9126,6 +9221,7 @@ methods = [{}]
[server.auth.github]
allowed_usernames = ["octocat"]
{extra_config}
"#,
storage_dir.display(),
methods
@ -10125,6 +10221,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 +13116,10 @@ timeout = "30s"
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().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 +13215,10 @@ timeout = "30s"
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().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()

3
lib/crates/fabro-server/src/spawn_env.rs
View file

@ -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")

259
lib/crates/fabro-util/src/run_log.rs
View file

@ -1,72 +1,64 @@
#![expect(
clippy::disallowed_types,
reason = "file-backed tracing sink: sync BufWriter<File> 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<RunLogWriter> = 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<AtomicBool>,
file: Arc<Mutex<Option<BufWriter<std::fs::File>>>>,
pub struct BufferedFileAppender {
file: Arc<Mutex<File>>,
}
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<Self> {
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<u8>,
file: Arc<Mutex<Option<BufWriter<std::fs::File>>>>,
},
/// Per-event write guard. Buffers tracing's write calls and flushes the whole
/// event when the guard is dropped.
pub struct BufferedFileGuard {
buf: Vec<u8>,
file: Arc<Mutex<File>>,
}
impl Write for RunLogGuard {
impl Write for BufferedFileGuard {
fn write(&mut self, data: &[u8]) -> io::Result<usize> {
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,50 @@ 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);
}
}
}
/// 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 +117,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::<Vec<_>>();
for handle in handles {
handle.join().unwrap();
}
let contents = std::fs::read_to_string(&path).unwrap();
let lines = contents.lines().collect::<Vec<_>>();
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::<u8>().ok())
.unwrap();
let expected_marker = (b'a' + thread_idx) as char;
assert!(
payload.chars().all(|ch| ch == expected_marker),
"line payload was interleaved: {line:?}"
);
}
}
}

4
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -168,6 +168,10 @@ impl RunDump {
Ok(())
}
pub fn add_file_bytes(&mut self, path: impl Into<String>, contents: Vec<u8>) {
self.entries.push(RunDumpEntry::bytes(path, contents));
}
pub async fn hydrate_referenced_blobs_with_reader<'a, F>(
&mut self,
mut read_blob: F,

74
lib/packages/fabro-api-client/src/api/run-internals-api.ts
View file

@ -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<RequestArgs> => {
// 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<string>> {
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<string> {
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<string> {
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