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

refactor: simplify artifact ZIP download

Browse source 
Path safety now lives in one place. The NUL-byte and drive-letter rules
move from a server-only helper into the store's own filename validation,
so uploads reject those paths at write time instead of only the ZIP read
path catching them. The download still re-checks, because artifacts
stored before the rule existed can still carry an unsafe path, but it
now skips a bad path rather than failing the whole archive.

Promote is_boundary_stage to RunProjection and drop the three identical
private copies. The ZIP download used a node-name match instead, which
would have dropped artifacts from a working node that happened to be
named "start".

Compress the archive. Entries were Stored while the response was also
excluded from transfer compression, so text artifacts moved at full
size. async_zip gains the deflate feature; async-compression and flate2
were already in the lock file.

Log archive failures unconditionally. The send-succeeded guard meant a
client that had already disconnected left no record at all, which is the
case where the log is the only evidence.

Also: collapse the duplicate 500 arms, drop the dead stage-ID tiebreaker
and the cached order in the selection map, name the accessible label
after the visible one, share the run URL prefix between the two download
href builders, and document the mid-stream truncation behavior in the
OpenAPI description.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-08-01 09:09:54 -04:00
parent 980aabc543
commit daaca3f479
No known key found for this signature in database
13 changed files with 145 additions and 132 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -338,6 +338,7 @@ checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac"
dependencies = [
"compression-codecs",
"compression-core",
"futures-io",
"pin-project-lite",
"tokio",
]
@ -408,6 +409,7 @@ version = "0.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6"
dependencies = [
"async-compression",
"crc32fast",
"futures-lite",
"pin-project",

2
Cargo.toml
View file

@ -36,7 +36,7 @@ dotenvy = "0.15"
futures = "0.3"
tokio-stream = "0.1"
async-trait = "0.1"
async_zip = { version = "0.0.18", features = ["tokio"] }
async_zip = { version = "0.0.18", features = ["tokio", "deflate"] }
fs2 = "0.4"
base64 = "0.22"
bytes = "1"

18
apps/fabro-web/app/lib/api-client.ts
View file

@ -408,6 +408,13 @@ export function requestSignalOptions(request?: Request): RawAxiosRequestConfig {
return request?.signal ? { signal: request.signal } : {};
}
/** Absolute href for a path under a run, for links the browser follows itself. */
function runApiPath(id: string, suffix: string): string {
return `${generatedApiConfiguration.basePath ?? ""}/api/v1/runs/${
encodeURIComponent(id)
}${suffix}`;
}
export function stageArtifactDownloadUrl(
id: string,
stageId: string,
@ -418,13 +425,12 @@ export function stageArtifactDownloadUrl(
filename,
retry: String(retry),
});
return `${generatedApiConfiguration.basePath ?? ""}/api/v1/runs/${
encodeURIComponent(id)
}/stages/${encodeURIComponent(stageId)}/artifacts/download?${searchParams}`;
return runApiPath(
id,
`/stages/${encodeURIComponent(stageId)}/artifacts/download?${searchParams}`,
);
}
export function runArtifactsDownloadUrl(id: string): string {
return `${generatedApiConfiguration.basePath ?? ""}/api/v1/runs/${
encodeURIComponent(id)
}/artifacts/download`;
return runApiPath(id, "/artifacts/download");
}

3
apps/fabro-web/app/routes/run-artifacts.tsx
View file

@ -138,8 +138,7 @@ function ArtifactList({ runId, files }: { runId: string; files: readonly Artifac
</span>
<a
href={runArtifactsDownloadUrl(runId)}
download={`fabro-artifacts-${runId}.zip`}
aria-label={`Download the latest version of all ${files.length} ${plural(files.length, "artifact", "artifacts")} as a ZIP file`}
aria-label={`Download all ${files.length} ${plural(files.length, "artifact", "artifacts")} as a ZIP file`}
className="inline-flex min-h-11 shrink-0 items-center gap-1.5 rounded-md bg-overlay px-2.5 py-1 text-xs font-medium text-fg-2 outline-1 -outline-offset-1 outline-line-strong hover:bg-overlay-strong hover:text-fg focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-teal-500 sm:min-h-8"
>
<ArrowDownTrayIcon className="size-3.5 shrink-0" aria-hidden="true" />

6
docs/public/api-reference/fabro-api.yaml
View file

@ -3289,8 +3289,12 @@ paths:
summary: Download Run Artifacts
description: |
Streams a ZIP archive with the latest captured version of each artifact path.
Stage order, retry number, and stage ID determine the latest version, matching the artifacts page.
Stage order and retry number determine the latest version, matching the artifacts page.
Captures from the `start` and `exit` control nodes are excluded.
The archive streams, so the response status is sent before the first artifact is read.
A failure after that point aborts the transfer rather than returning `500`.
The ZIP central directory is written last, so a truncated download does not open as a valid archive.
parameters:
- $ref: "#/components/parameters/RunId"
responses:

142
lib/apps/fabro-server/src/server/handler/artifacts.rs
View file

@ -5,7 +5,9 @@ use async_zip::base::write::ZipFileWriter;
use async_zip::error::ZipError;
use async_zip::{Compression, ZipEntryBuilder};
use axum::http::HeaderValue;
use fabro_store::{ArtifactStore, Error as StoreError, RunProjection};
use fabro_store::{ArtifactStore, Error as StoreError};
use fabro_types::RunProjection;
use fabro_util::error::collect_chain;
use futures_util::SinkExt as _;
use futures_util::io::AsyncWriteExt as _;
use tokio::io::AsyncWrite;
@ -171,61 +173,60 @@ enum ArtifactArchiveError {
Io(#[from] io::Error),
#[error("artifact read failed: {0}")]
Store(#[from] StoreError),
#[error("artifact disappeared while the archive was being created")]
MissingArtifact,
#[error("artifact {0} disappeared while the archive was being created")]
MissingArtifact(String),
#[error("ZIP write failed: {0}")]
Zip(#[from] ZipError),
}
fn validate_artifact_archive_path(path: &str) -> Result<(), StoreError> {
ArtifactStore::validate_relative_path(path)?;
let bytes = path.as_bytes();
let has_windows_drive_prefix =
bytes.first().is_some_and(u8::is_ascii_alphabetic) && bytes.get(1) == Some(&b':');
if path.contains('\0') || has_windows_drive_prefix {
return Err(StoreError::Other(
"artifact path is not safe for a ZIP archive".to_string(),
));
}
Ok(())
}
/// One entry per artifact path, holding the newest capture of that path.
///
/// Newest means latest stage, then latest retry. Stages the projection does not
/// know about sort oldest (`None` < `Some`), which is what the artifacts page
/// does too. Boundary stages are dropped: they run no work, so anything they
/// captured was already in the workspace.
///
/// Paths are re-checked here rather than trusted: a path stored before a
/// validation rule existed would otherwise be written straight into a ZIP that
/// somebody extracts. An unsafe path is skipped, not fatal — one bad path must
/// not cost the caller every other artifact.
fn latest_run_artifacts(
entries: Vec<NodeArtifact>,
projection: &RunProjection,
) -> Result<Vec<NodeArtifact>, StoreError> {
) -> Vec<NodeArtifact> {
let stage_order = projection
.iter_stages()
.enumerate()
.map(|(order, (stage_id, _))| (stage_id.clone(), order))
.collect::<HashMap<_, _>>();
let mut latest_by_path: HashMap<String, (Option<usize>, NodeArtifact)> = HashMap::new();
let capture_rank =
|artifact: &NodeArtifact| (stage_order.get(&artifact.node).copied(), artifact.retry);
let mut latest_by_path: HashMap<String, NodeArtifact> = HashMap::new();
for artifact in entries {
if matches!(artifact.node.node_id(), "start" | "exit") {
if projection.is_boundary_stage(artifact.node.node_id()) {
continue;
}
if let Err(error) = ArtifactStore::validate_relative_path(&artifact.filename) {
warn!(
path = %artifact.filename,
%error,
"skipping artifact with an unsafe path"
);
continue;
}
validate_artifact_archive_path(&artifact.filename)?;
let order = stage_order.get(&artifact.node).copied();
let replace =
latest_by_path
.get(&artifact.filename)
.is_none_or(|(existing_order, existing)| {
(order, artifact.retry, &artifact.node)
> (*existing_order, existing.retry, &existing.node)
});
if replace {
latest_by_path.insert(artifact.filename.clone(), (order, artifact));
match latest_by_path.get(&artifact.filename) {
Some(existing) if capture_rank(existing) >= capture_rank(&artifact) => {}
_ => {
latest_by_path.insert(artifact.filename.clone(), artifact);
}
}
}
let mut latest = latest_by_path
.into_values()
.map(|(_, artifact)| artifact)
.collect::<Vec<_>>();
let mut latest = latest_by_path.into_values().collect::<Vec<_>>();
latest.sort_by(|left, right| left.filename.cmp(&right.filename));
Ok(latest)
latest
}
async fn write_artifact_archive<W>(
@ -241,9 +242,15 @@ where
for artifact in artifacts {
let key = ArtifactKey::new(artifact.node, artifact.retry, artifact.filename.clone());
let Some(mut source) = artifact_store.get_stream(&run_id, &key).await? else {
return Err(ArtifactArchiveError::MissingArtifact);
return Err(ArtifactArchiveError::MissingArtifact(artifact.filename));
};
let entry = ZipEntryBuilder::new(artifact.filename.into(), Compression::Stored);
// Deflate, not Stored: artifacts are mostly logs and reports, and the
// response is excluded from transfer compression precisely because the
// archive already carries its own.
let entry = ZipEntryBuilder::new(artifact.filename.into(), Compression::Deflate);
// `destination` is a futures-io writer from async_zip, while `writer` at
// the end of this function is a tokio one, so both `AsyncWriteExt`
// traits are in scope and each call resolves to a different one.
let mut destination = archive.write_entry_stream(entry).await?;
while let Some(chunk) = source.next().await {
destination.write_all(&chunk?).await?;
@ -260,6 +267,12 @@ fn artifact_archive_body(
run_id: RunId,
artifacts: Vec<NodeArtifact>,
) -> Body {
// A channel of `Result`, rather than `tokio::io::duplex`, so a failure
// partway through can poison the body. Dropping a duplex writer ends the
// response with a clean EOF, which would hand the caller a truncated ZIP
// that looks like a complete download. Sending an `Err` aborts the chunked
// body instead, so the caller sees a transfer error. Do not "simplify" this
// to a duplex without replacing that signal.
let (sender, receiver) =
mpsc::channel::<Result<Bytes, io::Error>>(ARCHIVE_STREAM_CHANNEL_CAPACITY);
let error_sender = sender.clone();
@ -273,10 +286,17 @@ fn artifact_archive_body(
tokio::spawn(async move {
if let Err(error) = write_artifact_archive(writer, artifact_store, run_id, artifacts).await
{
let body_error = io::Error::other("artifact archive stream failed");
if error_sender.send(Err(body_error)).await.is_ok() {
warn!(%run_id, %error, "artifact archive stream failed");
}
// Log before signalling: the send fails when the caller has already
// gone away, and that is exactly when this log is the only record
// that the archive failed. The 200 went out long ago.
warn!(
%run_id,
error = %collect_chain(&error).join(": "),
"artifact archive stream failed"
);
let _ = error_sender
.send(Err(io::Error::other("artifact archive stream failed")))
.await;
}
});
@ -307,19 +327,9 @@ async fn download_run_artifacts(
.into_response();
}
};
let artifacts = match latest_run_artifacts(entries, &cached.projection) {
Ok(artifacts) => artifacts,
Err(error) => {
warn!(run_id = %id, %error, "failed to select artifacts for ZIP download");
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Artifact archive could not be prepared.",
)
.into_response();
}
};
let artifacts = latest_run_artifacts(entries, &cached.projection);
let filename = format!("attachment; filename=\"fabro-artifacts-{id}.zip\"");
let content_disposition = format!("attachment; filename=\"fabro-artifacts-{id}.zip\"");
let body = artifact_archive_body(state.artifact_store.clone(), id, artifacts);
let mut response = body.into_response();
response.headers_mut().insert(
@ -328,7 +338,8 @@ async fn download_run_artifacts(
);
response.headers_mut().insert(
header::CONTENT_DISPOSITION,
HeaderValue::from_str(&filename).expect("run IDs produce valid attachment filenames"),
HeaderValue::from_str(&content_disposition)
.expect("run IDs produce valid attachment filenames"),
);
response.headers_mut().insert(
header::CACHE_CONTROL,
@ -837,26 +848,3 @@ async fn get_stage_artifact(
}
}
}
#[cfg(test)]
mod tests {
use super::validate_artifact_archive_path;
#[test]
fn archive_paths_reject_cross_platform_absolute_forms() {
for path in [
"/escape.txt",
"../escape.txt",
r"..\escape.txt",
"C:/escape.txt",
"c:escape.txt",
"bad\0name",
] {
assert!(
validate_artifact_archive_path(path).is_err(),
"accepted unsafe archive path: {path:?}"
);
}
assert!(validate_artifact_archive_path("nested/result.txt").is_ok());
}
}

11
lib/apps/fabro-server/src/server/handler/billing.rs
View file

@ -170,7 +170,7 @@ fn live_billing_rows(projection: &RunProjection, now: DateTime<Utc>) -> Vec<Live
for (stage_id, stage) in projection.iter_stages() {
let node_id = stage_id.node_id();
if is_boundary_stage(projection, node_id) || !stage_has_billing_row(stage) {
if projection.is_boundary_stage(node_id) || !stage_has_billing_row(stage) {
continue;
}
@ -205,12 +205,3 @@ fn stage_has_billing_row(stage: &StageProjection) -> bool {
|| !stage.usage.is_zero()
|| stage.started_at.is_some()
}
fn is_boundary_stage(projection: &RunProjection, node_id: &str) -> bool {
projection
.spec()
.graph()
.nodes
.get(node_id)
.is_some_and(|node| matches!(node.handler_type(), Some("start" | "exit")))
}

17
lib/apps/fabro-server/src/server/tests.rs
View file

@ -10692,7 +10692,6 @@ async fn run_artifacts_download_streams_latest_files_as_zip() {
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
@ -10717,15 +10716,7 @@ async fn run_artifacts_download_streams_latest_files_as_zip() {
assert!(response.headers().get(header::CONTENT_ENCODING).is_none());
let bytes = response_bytes!(response, StatusCode::OK).await;
let archive = ZipFileReader::new(bytes.clone())
.await
.unwrap_or_else(|error| {
panic!(
"ZIP parse failed for {} bytes (tail {:?}): {error}",
bytes.len(),
&bytes[bytes.len().saturating_sub(48)..]
)
});
let archive = ZipFileReader::new(bytes).await.unwrap();
let names = archive
.file()
.entries()
@ -10743,6 +10734,12 @@ async fn run_artifacts_download_streams_latest_files_as_zip() {
}
assert_eq!(contents_by_name["logs/run.txt"], b"build log");
assert_eq!(contents_by_name["reports/result.txt"], b"latest verify");
}
#[tokio::test]
async fn run_artifacts_download_returns_not_found_for_unknown_run() {
let state = test_app_state();
let app = crate::test_support::build_test_router(Arc::clone(&state));
let response = app
.oneshot(

32
lib/components/fabro-store/src/artifact_store.rs
View file

@ -141,6 +141,11 @@ impl ArtifactStore {
}
}
/// Check a path against the same rules `put` enforces, without writing.
///
/// Readers that hand artifact paths back to a client — the ZIP download,
/// for one — use this to re-check paths that were stored before a rule
/// existed.
pub fn validate_relative_path(relative_path: &str) -> Result<()> {
validate_filename_segments(relative_path).map(drop)
}
@ -243,12 +248,27 @@ impl ArtifactStore {
}
}
/// Artifact filenames end up as paths on someone else's disk — extracted from a
/// ZIP, written by a worker — so they must stay relative and portable. The
/// backslash, NUL, and drive-letter rules are what make the path safe on
/// Windows as well as Unix.
fn validate_filename_segments(filename: &str) -> Result<Vec<&str>> {
if filename.contains('\\') {
return Err(Error::Other(
"artifact filename must not contain backslashes".to_string(),
));
}
if filename.contains('\0') {
return Err(Error::Other(
"artifact filename must not contain NUL bytes".to_string(),
));
}
let bytes = filename.as_bytes();
if bytes.first().is_some_and(u8::is_ascii_alphabetic) && bytes.get(1) == Some(&b':') {
return Err(Error::Other(
"artifact filename must not start with a drive letter".to_string(),
));
}
let segments = filename.split('/').collect::<Vec<_>>();
if segments.is_empty() || segments.iter().any(|segment| segment.is_empty()) {
return Err(Error::Other(
@ -507,15 +527,25 @@ mod tests {
for filename in [
"",
"/escape.txt",
"../escape.txt",
"logs//output.txt",
"logs/./output.txt",
r"logs\output.txt",
"C:/escape.txt",
"c:escape.txt",
"bad\0name.txt",
] {
let key = ArtifactKey::new(node.clone(), 1, filename);
let err = store.put(&run_id, &key, b"boom").await.unwrap_err();
assert!(err.to_string().contains("artifact filename"));
assert!(
err.to_string().contains("artifact filename"),
"accepted unsafe artifact filename: {filename:?}"
);
assert!(ArtifactStore::validate_relative_path(filename).is_err());
}
assert!(ArtifactStore::validate_relative_path("nested/result.txt").is_ok());
}
#[tokio::test]

11
lib/components/fabro-store/src/run_state.rs
View file

@ -1363,22 +1363,13 @@ pub(crate) fn projected_billing(state: &RunProjection) -> BilledTokenCounts {
let mut billing = BilledTokenCounts::default();
for (stage_id, stage) in state.iter_stages() {
if !is_boundary_stage(state, stage_id.node_id()) {
if !state.is_boundary_stage(stage_id.node_id()) {
billing.add_counts(&stage.usage);
}
}
billing
}
fn is_boundary_stage(projection: &RunProjection, node_id: &str) -> bool {
projection
.spec()
.graph()
.nodes
.get(node_id)
.is_some_and(|node| matches!(node.handler_type(), Some("start" | "exit")))
}
fn run_models(state: &RunProjection) -> Vec<RunModel> {
let mut models = state
.iter_stages()

11
lib/components/fabro-workflow/src/billing_rollup.rs
View file

@ -52,7 +52,7 @@ pub fn billing_rollup_from_projection(
let mut billed_visit_count = 0_usize;
for (stage_id, stage) in projection.iter_stages() {
if is_boundary_stage(projection, stage_id.node_id()) {
if projection.is_boundary_stage(stage_id.node_id()) {
continue;
}
let usage = stage.billed_usage(catalog);
@ -124,15 +124,6 @@ pub fn billing_rollup_from_projection(
}
}
fn is_boundary_stage(projection: &RunProjection, node_id: &str) -> bool {
projection
.spec()
.graph()
.nodes
.get(node_id)
.is_some_and(|node| matches!(node.handler_type(), Some("start" | "exit")))
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;

14
lib/foundation/fabro-types/src/run_projection.rs
View file

@ -962,6 +962,20 @@ impl RunProjection {
&self.spec
}
/// Whether a graph node is one of the `start`/`exit` boundaries.
///
/// Boundary nodes run no work, so callers that report what a run *did* —
/// billing, stage listings, artifact downloads — leave them out. The test
/// is the node's handler type, not its name: a node may be named
/// `start` and still do real work.
pub fn is_boundary_stage(&self, node_id: &str) -> bool {
self.spec()
.graph()
.nodes
.get(node_id)
.is_some_and(|node| matches!(node.handler_type(), Some("start" | "exit")))
}
pub fn status(&self) -> RunStatus {
self.status
}

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

@ -147,7 +147,7 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
};
},
/**
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order, retry number, and stage ID determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded.
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order and retry number determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded. The archive streams, so the response status is sent before the first artifact is read. A failure after that point aborts the transfer rather than returning `500`. The ZIP central directory is written last, so a truncated download does not open as a valid archive.
* @summary Download Run Artifacts
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
@ -987,7 +987,7 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order, retry number, and stage ID determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded.
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order and retry number determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded. The archive streams, so the response status is sent before the first artifact is read. A failure after that point aborts the transfer rather than returning `500`. The ZIP central directory is written last, so a truncated download does not open as a valid archive.
* @summary Download Run Artifacts
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
@ -1264,7 +1264,7 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
return localVarFp.attachRunEvents(id, sinceSeq, options).then((request) => request(axios, basePath));
},
/**
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order, retry number, and stage ID determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded.
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order and retry number determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded. The archive streams, so the response status is sent before the first artifact is read. A failure after that point aborts the transfer rather than returning `500`. The ZIP central directory is written last, so a truncated download does not open as a valid archive.
* @summary Download Run Artifacts
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
@ -1490,7 +1490,7 @@ export class RunInternalsApi extends BaseAPI {
}
/**
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order, retry number, and stage ID determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded.
* Streams a ZIP archive with the latest captured version of each artifact path. Stage order and retry number determine the latest version, matching the artifacts page. Captures from the `start` and `exit` control nodes are excluded. The archive streams, so the response status is sent before the first artifact is read. A failure after that point aborts the transfer rather than returning `500`. The ZIP central directory is written last, so a truncated download does not open as a valid archive.
* @summary Download Run Artifacts
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.