From 494a7fe1cce98c34b505ae4289e90efa62415d2b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 16:56:37 -0400 Subject: [PATCH] feat(artifacts): finish object-backed artifact uploads Add scoped worker upload tokens and HTTP artifact upload clients. Support manifest-first multipart stage artifact uploads with validation and checksums. Gate artifact reads by run capability while preserving legacy scratch fallback. --- Cargo.lock | 20 + Cargo.toml | 4 +- docs/api-reference/fabro-api.yaml | 82 +- lib/crates/fabro-api/build.rs | 54 + lib/crates/fabro-cli/Cargo.toml | 1 + lib/crates/fabro-cli/src/args.rs | 4 + lib/crates/fabro-cli/src/commands/run/mod.rs | 3 +- .../fabro-cli/src/commands/run/runner.rs | 77 +- .../fabro-cli/src/commands/store/dump.rs | 2 + lib/crates/fabro-cli/src/main.rs | 4 + lib/crates/fabro-cli/src/server_client.rs | 226 +++- lib/crates/fabro-config/src/config.rs | 4 + .../fabro-config/src/effective_settings.rs | 2 + lib/crates/fabro-config/src/server.rs | 6 +- lib/crates/fabro-config/src/settings.rs | 1 + lib/crates/fabro-config/src/storage.rs | 9 + lib/crates/fabro-server/Cargo.toml | 1 + lib/crates/fabro-server/src/jwt_auth.rs | 71 +- lib/crates/fabro-server/src/run_manifest.rs | 1 + lib/crates/fabro-server/src/serve.rs | 46 +- lib/crates/fabro-server/src/server.rs | 1204 ++++++++++++++--- lib/crates/fabro-store/src/artifact_store.rs | 82 +- lib/crates/fabro-store/src/run_state.rs | 1 + lib/crates/fabro-store/src/slate/mod.rs | 1 + lib/crates/fabro-types/src/lib.rs | 6 +- lib/crates/fabro-types/src/run.rs | 18 + lib/crates/fabro-types/src/run_event/run.rs | 4 +- lib/crates/fabro-types/src/settings/mod.rs | 8 +- lib/crates/fabro-types/src/settings/server.rs | 41 + .../fabro-workflow/src/artifact_upload.rs | 17 + lib/crates/fabro-workflow/src/event.rs | 4 + .../src/handler/manager_loop.rs | 1 + lib/crates/fabro-workflow/src/lib.rs | 1 + .../fabro-workflow/src/lifecycle/artifact.rs | 54 + .../fabro-workflow/src/lifecycle/mod.rs | 3 + .../fabro-workflow/src/operations/create.rs | 9 +- .../src/operations/rebuild_meta.rs | 2 + .../fabro-workflow/src/operations/start.rs | 6 + .../fabro-workflow/src/pipeline/execute.rs | 2 + .../src/pipeline/execute/tests.rs | 3 + .../fabro-workflow/src/pipeline/initialize.rs | 3 + .../fabro-workflow/src/pipeline/persist.rs | 2 + .../src/pipeline/pull_request.rs | 6 + .../fabro-workflow/src/pipeline/retro.rs | 2 + .../fabro-workflow/src/pipeline/types.rs | 3 + lib/crates/fabro-workflow/src/run_lookup.rs | 2 + .../fabro-workflow/src/runtime_store.rs | 2 + lib/crates/fabro-workflow/src/test_support.rs | 2 + .../src/.openapi-generator/FILES | 2 + .../src/api/run-internals-api.ts | 35 +- .../src/models/artifact-batch-upload-entry.ts | 41 + .../models/artifact-batch-upload-manifest.ts | 25 + .../fabro-api-client/src/models/index.ts | 2 + 53 files changed, 1953 insertions(+), 259 deletions(-) create mode 100644 lib/crates/fabro-workflow/src/artifact_upload.rs create mode 100644 lib/packages/fabro-api-client/src/models/artifact-batch-upload-entry.ts create mode 100644 lib/packages/fabro-api-client/src/models/artifact-batch-upload-manifest.ts diff --git a/Cargo.lock b/Cargo.lock index 4d1c2510d..174a0230e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1587,6 +1587,7 @@ dependencies = [ "shlex", "tempfile", "tokio", + "tokio-util", "toml 0.8.23", "tracing", "tracing-appender", @@ -1885,6 +1886,7 @@ dependencies = [ "hyper-util", "jsonwebtoken", "mime_guess", + "multer", "object_store", "openapiv3", "rand 0.8.5", @@ -3664,6 +3666,23 @@ dependencies = [ "parking_lot", ] +[[package]] +name = "multer" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http", + "httparse", + "memchr", + "mime", + "spin", + "version_check", +] + [[package]] name = "naive-timer" version = "0.2.0" @@ -4966,6 +4985,7 @@ dependencies = [ "hyper-util", "js-sys", "log", + "mime_guess", "percent-encoding", "pin-project-lite", "quinn", diff --git a/Cargo.toml b/Cargo.toml index 66c779105..90331a6f7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ thiserror = "2" serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } tokio = { version = "1", features = ["full"] } -reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "rustls", "query", "form"] } +reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "rustls", "query", "form", "multipart"] } ulid = "1" uuid = { version = "1", features = ["v4", "v7", "v8"] } rand = "0.8" @@ -69,7 +69,7 @@ sentry = { version = "0.35", default-features = false, features = ["backtrace", fork = "0.2" exec = "0.3" slatedb = "0.11.2" -object_store = "0.12.5" +object_store = { version = "0.12.5", features = ["aws"] } rust-embed = "8" percent-encoding = "2" diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index 83bf6de25..805c5ff55 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -562,6 +562,24 @@ paths: schema: type: string format: binary + multipart/form-data: + schema: + type: object + required: + - manifest + properties: + manifest: + $ref: "#/components/schemas/ArtifactBatchUploadManifest" + additionalProperties: + type: string + format: binary + description: | + Strict multipart upload format. The `manifest` part must arrive first with JSON + matching `ArtifactBatchUploadManifest`. Each subsequent file part name must match + a manifest entry `part` value. + encoding: + manifest: + contentType: application/json responses: "200": description: Blob written @@ -758,11 +776,24 @@ paths: operationId: putStageArtifact tags: [Run Internals] summary: Put Stage Artifact - description: Uploads an artifact for a stage. Intended for trusted internal callers. + description: | + Uploads one or more artifacts for a stage. Intended for trusted internal callers. + + The server accepts both: + - `application/octet-stream` for single-file uploads with the `filename` query parameter + - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` + + The generated Rust client currently exposes the octet-stream variant because the OpenAPI + code generator in this repo does not support multiple request media types on one operation. parameters: - $ref: "#/components/parameters/RunId" - $ref: "#/components/parameters/StageId" - - $ref: "#/components/parameters/ArtifactFilename" + - name: filename + in: query + required: false + description: Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. + schema: + type: string requestBody: required: true content: @@ -774,7 +805,7 @@ paths: "204": description: Artifact written "400": - description: Missing filename + description: Invalid filename, multipart manifest, checksum, or upload body content: application/json: schema: @@ -2684,6 +2715,51 @@ components: items: $ref: "#/components/schemas/ArtifactEntry" + ArtifactBatchUploadEntry: + description: One file entry in a strict multipart artifact upload manifest. + type: object + required: + - part + - path + properties: + part: + type: string + description: Multipart field name for the file part. + example: file1 + path: + type: string + description: Relative artifact path to store. + example: src/lib.rs + sha256: + type: string + nullable: true + description: Optional lowercase hex SHA-256 checksum for the file contents. + example: 3f785df4c5b7d3f1f4c1f0ecb0f55f1d9f6f6a3d9f0a8a98f7a74f29d1f81a2c + expected_bytes: + type: integer + format: int64 + nullable: true + minimum: 0 + description: Optional exact byte length expected for the file part. + example: 1234 + content_type: + type: string + nullable: true + description: Optional client-supplied content type for the file part. + example: text/plain + + ArtifactBatchUploadManifest: + description: Manifest for strict multipart artifact uploads. + type: object + required: + - entries + properties: + entries: + type: array + minItems: 1 + items: + $ref: "#/components/schemas/ArtifactBatchUploadEntry" + RunArtifactEntry: description: A captured artifact file for a run. type: object diff --git a/lib/crates/fabro-api/build.rs b/lib/crates/fabro-api/build.rs index 9a7495bc2..7f5688698 100644 --- a/lib/crates/fabro-api/build.rs +++ b/lib/crates/fabro-api/build.rs @@ -64,6 +64,59 @@ fn patch_nullable(value: &mut serde_json::Value) { } } +/// Progenitor currently panics when an operation advertises more than one request-body media type. +/// +/// Keep the source OpenAPI spec accurate for docs, but collapse the generated-client view down to +/// a single preferred media type so code generation can proceed. +fn patch_codegen_request_body_media_types(value: &mut serde_json::Value) { + let Some(paths) = value + .get_mut("paths") + .and_then(serde_json::Value::as_object_mut) + else { + return; + }; + + for path_item in paths.values_mut() { + let Some(item) = path_item.as_object_mut() else { + continue; + }; + + for method in ["get", "put", "post", "delete", "patch"] { + let Some(operation) = item + .get_mut(method) + .and_then(serde_json::Value::as_object_mut) + else { + continue; + }; + let Some(content) = operation + .get_mut("requestBody") + .and_then(|request_body| request_body.get_mut("content")) + .and_then(serde_json::Value::as_object_mut) + else { + continue; + }; + if content.len() <= 1 { + continue; + } + + let preferred = content + .get("application/octet-stream") + .cloned() + .map(|value| ("application/octet-stream".to_string(), value)) + .or_else(|| { + content + .iter() + .next() + .map(|(key, value)| (key.clone(), value.clone())) + }); + if let Some((key, value)) = preferred { + content.clear(); + content.insert(key, value); + } + } + } +} + fn main() { let spec_path = Path::new(env!("CARGO_MANIFEST_DIR")) .parent() @@ -86,6 +139,7 @@ fn main() { // rely on any 3.1-only features that affect codegen. spec_value["openapi"] = serde_json::Value::String("3.0.3".to_string()); patch_nullable(&mut spec_value); + patch_codegen_request_body_media_types(&mut spec_value); let spec: openapiv3::OpenAPI = serde_json::from_value(spec_value).expect("failed to deserialize OpenAPI spec"); diff --git a/lib/crates/fabro-cli/Cargo.toml b/lib/crates/fabro-cli/Cargo.toml index 0fa126743..e294575d7 100644 --- a/lib/crates/fabro-cli/Cargo.toml +++ b/lib/crates/fabro-cli/Cargo.toml @@ -83,6 +83,7 @@ shlex = "1" walkdir.workspace = true object_store.workspace = true bytes.workspace = true +tokio-util.workspace = true [target.'cfg(target_os = "macos")'.dependencies] core-foundation = { version = "0.9", optional = true } diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index 32fa21c89..2ef6d0f76 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -730,6 +730,10 @@ pub(crate) struct RunWorkerArgs { #[arg(long)] pub(crate) server: String, + /// Short-lived bearer token for artifact uploads + #[arg(long, hide = true)] + pub(crate) artifact_upload_token: Option, + /// Run scratch directory #[arg(long)] pub(crate) run_dir: PathBuf, diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 7a10c3b58..8652ce4ef 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -78,10 +78,11 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } RunCommands::RunWorker(RunWorkerArgs { server, + artifact_upload_token, run_dir, run_id, mode, - }) => runner::execute(run_id, server, run_dir, mode).await, + }) => runner::execute(run_id, server, artifact_upload_token, run_dir, mode).await, RunCommands::Diff(args) => diff::run(args, globals).await, RunCommands::Logs(args) => { let styles = Styles::detect_stdout(); diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 998bd2cbd..2a9c7dbb6 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -1,4 +1,4 @@ -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; @@ -9,6 +9,8 @@ use fabro_config::RunScratch; use fabro_interview::FileInterviewer; use fabro_store::{EventEnvelope, EventPayload, RunProjection}; use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason}; +use fabro_workflow::artifact_snapshot::CapturedArtifactInfo; +use fabro_workflow::artifact_upload::StageArtifactUploader; use fabro_workflow::event::{Emitter, RunEventSink}; use fabro_workflow::run_control::RunControlState; use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle}; @@ -43,6 +45,7 @@ enum WorkerTitlePhase { pub(crate) async fn execute( run_id: RunId, server: String, + artifact_upload_token: Option, run_dir: PathBuf, mode: RunWorkerMode, ) -> Result<()> { @@ -59,6 +62,12 @@ pub(crate) async fn execute( .run .as_ref() .ok_or_else(|| anyhow!("Run {run_id} has no run record in store"))?; + let artifact_uploader = build_artifact_uploader( + run_id, + run_record, + client.clone_for_reuse(), + artifact_upload_token, + )?; let scratch = RunScratch::new(&run_dir); let interviewer = Arc::new(FileInterviewer::new( scratch.interview_request_path(), @@ -82,6 +91,7 @@ pub(crate) async fn execute( async move { Ok(()) } }), ]), + artifact_uploader, run_control: Some(run_control), github_app, on_node: None, @@ -100,6 +110,70 @@ pub(crate) async fn execute( Ok(()) } +fn build_artifact_uploader( + run_id: RunId, + run_record: &fabro_types::RunRecord, + client: server_client::ServerStoreClient, + artifact_upload_token: Option, +) -> Result>> { + if !run_record.uses_object_backed_artifacts() { + return Ok(None); + } + + let token = artifact_upload_token + .ok_or_else(|| anyhow!("run {run_id} is configured for object-backed artifacts but the worker did not receive an artifact upload token"))?; + + Ok(Some(Arc::new(HttpArtifactUploader { + run_id, + client, + bearer_token: token, + }))) +} + +struct HttpArtifactUploader { + run_id: RunId, + client: server_client::ServerStoreClient, + bearer_token: String, +} + +#[async_trait] +impl StageArtifactUploader for HttpArtifactUploader { + async fn upload_stage_artifacts( + &self, + stage_id: &fabro_types::StageId, + artifact_capture_dir: &Path, + artifacts: &[CapturedArtifactInfo], + ) -> Result<()> { + if artifacts.is_empty() { + return Ok(()); + } + + if artifacts.len() == 1 { + let artifact = &artifacts[0]; + return self + .client + .upload_stage_artifact_file( + &self.run_id, + stage_id, + &artifact.path, + &artifact_capture_dir.join(&artifact.path), + &self.bearer_token, + ) + .await; + } + + self.client + .upload_stage_artifact_batch( + &self.run_id, + stage_id, + artifact_capture_dir, + artifacts, + &self.bearer_token, + ) + .await + } +} + #[derive(Clone)] struct HttpRunStore { run_id: RunId, @@ -495,6 +569,7 @@ mod tests { let error = execute( run_id, format!("{}/api/v1", server.base_url()), + None, run_dir.path().to_path_buf(), RunWorkerMode::Start, ) diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 7f001272e..0a1764414 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -193,6 +193,7 @@ mod tests { repo_origin_url: Some("https://github.com/fabro-sh/fabro".to_string()), base_branch: Some("main".to_string()), labels: HashMap::from([("team".to_string(), "infra".to_string())]), + artifact_storage: None, provenance: None, } } @@ -328,6 +329,7 @@ mod tests { base_branch: run_record.base_branch.clone(), workflow_slug: run_record.workflow_slug.clone(), db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 95f404985..5fb95fbd9 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -498,6 +498,8 @@ mod tests { "__run-worker", "--server", "/tmp/fabro.sock", + "--artifact-upload-token", + "token-123", "--run-dir", "/tmp/run", "--run-id", @@ -509,6 +511,7 @@ mod tests { match *cli.command { Commands::RunCmd(RunCommands::RunWorker(args)) => { assert_eq!(args.server, "/tmp/fabro.sock"); + assert_eq!(args.artifact_upload_token.as_deref(), Some("token-123")); assert_eq!(args.run_dir, std::path::PathBuf::from("/tmp/run")); assert_eq!(args.run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); assert!(matches!(args.mode, args::RunWorkerMode::Start)); @@ -535,6 +538,7 @@ mod tests { match *cli.command { Commands::RunCmd(RunCommands::RunWorker(args)) => { assert_eq!(args.server, "http://127.0.0.1:3000"); + assert!(args.artifact_upload_token.is_none()); assert_eq!(args.run_dir, std::path::PathBuf::from("/tmp/run")); assert_eq!(args.run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); assert!(matches!(args.mode, args::RunWorkerMode::Resume)); diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index f1933df7b..6bcac2133 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -9,9 +9,12 @@ use fabro_api::types; use fabro_server::bind::Bind; use fabro_store::{EventEnvelope, RunSummary, StageId}; use fabro_types::{RunBlobId, RunEvent, RunId, Settings}; +use fabro_workflow::artifact_snapshot::CapturedArtifactInfo; use futures::StreamExt; +use serde::Serialize; use serde::de::DeserializeOwned; use tokio::time::sleep; +use tokio_util::io::ReaderStream; use crate::args::ServerTargetArgs; use crate::commands::server::start; @@ -21,6 +24,8 @@ use crate::user_config; #[derive(Clone)] pub(crate) struct ServerStoreClient { client: fabro_api::Client, + http_client: reqwest::Client, + base_url: String, } #[derive(Debug, Clone)] @@ -76,22 +81,19 @@ impl RunAttachEventStream { pub(crate) use fabro_store::RunProjection; pub(crate) async fn connect_server(storage_dir: &Path) -> Result { - Ok(ServerStoreClient { - client: connect_api_client(storage_dir).await?, - }) + connect_api_client_bundle(storage_dir).await } pub(crate) async fn connect_server_target_direct(target: &str) -> Result { - let client = if target.starts_with("http://") || target.starts_with("https://") { - connect_remote_api_client(target, None)? + if target.starts_with("http://") || target.starts_with("https://") { + connect_remote_api_client_bundle(target, None) } else { let path = Path::new(target); if !path.is_absolute() { bail!("server target must be an http(s) URL or absolute Unix socket path"); } - connect_unix_socket_api_client(path).await? - }; - Ok(ServerStoreClient { client }) + connect_unix_socket_api_client_bundle(path).await + } } pub(crate) async fn connect_server_only(args: &ServerTargetArgs) -> Result { @@ -101,33 +103,46 @@ pub(crate) async fn connect_server_only(args: &ServerTargetArgs) -> Result Result { +async fn connect_api_client_bundle(storage_dir: &Path) -> Result { let config_path = user_config::active_settings_path(None); let bind = start::ensure_server_running_for_storage(storage_dir, &config_path) .with_context(|| format!("Failed to start fabro server for {}", storage_dir.display()))?; match bind { - Bind::Unix(path) => connect_unix_socket_api_client(&path).await, + Bind::Unix(path) => connect_unix_socket_api_client_bundle(&path).await, Bind::Tcp(addr) => Err(anyhow!( "Unsupported server bind for store client auto-connect: {addr}" )), } } +pub(crate) async fn connect_api_client(storage_dir: &Path) -> Result { + connect_api_client_bundle(storage_dir) + .await + .map(|client| client.client) +} + async fn connect_target_api_client( target: &user_config::ServerTarget, runtime: &LocalServerRuntime, ) -> Result { + connect_target_api_client_bundle(target, runtime) + .await + .map(|client| client.client) +} + +async fn connect_target_api_client_bundle( + target: &user_config::ServerTarget, + runtime: &LocalServerRuntime, +) -> Result { match target { user_config::ServerTarget::HttpUrl { api_url, tls } => { - Ok(connect_remote_api_client(api_url, tls.as_ref())?) + connect_remote_api_client_bundle(api_url, tls.as_ref()) } user_config::ServerTarget::UnixSocket(path) => { - if let Ok(client) = connect_unix_socket_api_client(path).await { + if let Ok(client) = connect_unix_socket_api_client_bundle(path).await { Ok(client) } else { start::ensure_server_running_on_socket( @@ -136,7 +151,7 @@ async fn connect_target_api_client( &runtime.storage_dir, ) .with_context(|| format!("Failed to start fabro server for {}", path.display()))?; - connect_unix_socket_api_client(path).await + connect_unix_socket_api_client_bundle(path).await } } } @@ -161,13 +176,18 @@ pub(crate) async fn connect_server_backed_api_client_with_storage_dir( connect_target_api_client(&target, &runtime).await } -pub(crate) fn connect_remote_api_client( +fn connect_remote_api_client_bundle( api_url: &str, tls: Option<&user_config::ClientTlsSettings>, -) -> Result { +) -> Result { let http_client = user_config::build_server_client(tls)?; let normalized = normalize_remote_server_target(api_url); - Ok(fabro_api::Client::new_with_client(&normalized, http_client)) + let client = fabro_api::Client::new_with_client(&normalized, http_client.clone()); + Ok(ServerStoreClient { + client, + http_client, + base_url: normalized, + }) } fn normalize_remote_server_target(api_url: &str) -> String { @@ -178,7 +198,7 @@ fn normalize_remote_server_target(api_url: &str) -> String { .to_string() } -pub(crate) async fn connect_unix_socket_api_client(path: &Path) -> Result { +async fn connect_unix_socket_api_client_bundle(path: &Path) -> Result { let http_client = crate::user_config::cli_http_client_builder() .unix_socket(path) .no_proxy() @@ -186,10 +206,13 @@ pub(crate) async fn connect_unix_socket_api_client(path: &Path) -> Result Result<()> { @@ -213,6 +236,23 @@ async fn wait_for_server_ready(http_client: &reqwest::Client) -> Result<()> { Err(last_error.unwrap_or_else(|| anyhow!("server did not become ready in time"))) } +#[derive(Debug, Serialize)] +struct ArtifactBatchUploadManifest { + entries: Vec, +} + +#[derive(Debug, Serialize)] +struct ArtifactBatchUploadEntry { + part: String, + path: String, + #[serde(skip_serializing_if = "Option::is_none")] + sha256: Option, + #[serde(skip_serializing_if = "Option::is_none")] + expected_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + content_type: Option, +} + impl ServerStoreClient { pub(crate) fn clone_for_reuse(&self) -> Self { self.clone() @@ -508,6 +548,120 @@ impl ServerStoreClient { Ok(bytes) } + fn stage_artifacts_url(&self, run_id: &RunId, stage_id: &StageId) -> Result { + let mut url = reqwest::Url::parse(&self.base_url) + .with_context(|| format!("invalid server base URL {}", self.base_url))?; + url.path_segments_mut() + .map_err(|_| anyhow!("server base URL cannot accept path segments"))? + .extend([ + "api", + "v1", + "runs", + &run_id.to_string(), + "stages", + &stage_id.to_string(), + "artifacts", + ]); + Ok(url) + } + + pub(crate) async fn upload_stage_artifact_file( + &self, + run_id: &RunId, + stage_id: &StageId, + filename: &str, + path: &Path, + bearer_token: &str, + ) -> Result<()> { + let mut url = self.stage_artifacts_url(run_id, stage_id)?; + url.query_pairs_mut().append_pair("filename", filename); + + let file = tokio::fs::File::open(path) + .await + .with_context(|| format!("failed to open artifact {}", path.display()))?; + let content_length = file + .metadata() + .await + .with_context(|| format!("failed to stat artifact {}", path.display()))? + .len(); + let body = reqwest::Body::wrap_stream(ReaderStream::new(file)); + + let response = self + .http_client + .post(url) + .bearer_auth(bearer_token) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .header(reqwest::header::CONTENT_LENGTH, content_length.to_string()) + .body(body) + .send() + .await + .with_context(|| format!("failed to upload artifact {}", path.display()))?; + ensure_raw_response_success(response).await + } + + pub(crate) async fn upload_stage_artifact_batch( + &self, + run_id: &RunId, + stage_id: &StageId, + artifact_capture_dir: &Path, + artifacts: &[CapturedArtifactInfo], + bearer_token: &str, + ) -> Result<()> { + let url = self.stage_artifacts_url(run_id, stage_id)?; + let mut manifest_entries = Vec::with_capacity(artifacts.len()); + let mut file_parts = Vec::with_capacity(artifacts.len()); + + for (index, artifact) in artifacts.iter().enumerate() { + let part_name = format!("file{}", index + 1); + let path = artifact_capture_dir.join(&artifact.path); + let file = tokio::fs::File::open(&path) + .await + .with_context(|| format!("failed to open artifact {}", path.display()))?; + let content_length = file + .metadata() + .await + .with_context(|| format!("failed to stat artifact {}", path.display()))? + .len(); + + manifest_entries.push(ArtifactBatchUploadEntry { + part: part_name.clone(), + path: artifact.path.clone(), + sha256: Some(artifact.content_sha256.clone()), + expected_bytes: Some(artifact.bytes), + content_type: Some(artifact.mime.clone()), + }); + + file_parts.push(( + part_name, + reqwest::multipart::Part::stream_with_length( + reqwest::Body::wrap_stream(ReaderStream::new(file)), + content_length, + ) + .file_name(artifact.path.clone()), + )); + } + + let manifest = ArtifactBatchUploadManifest { + entries: manifest_entries, + }; + let manifest_part = reqwest::multipart::Part::text(serde_json::to_string(&manifest)?) + .mime_str("application/json")?; + let mut form = reqwest::multipart::Form::new().part("manifest", manifest_part); + for (part_name, part) in file_parts { + form = form.part(part_name, part); + } + + let response = self + .http_client + .post(url) + .bearer_auth(bearer_token) + .multipart(form) + .send() + .await + .context("failed to upload artifact batch")?; + ensure_raw_response_success(response).await + } + pub(crate) async fn generate_preview_url( &self, run_id: &RunId, @@ -629,6 +783,32 @@ where } } +async fn ensure_raw_response_success(response: reqwest::Response) -> Result<()> { + if response.status().is_success() { + return Ok(()); + } + + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + if let Ok(value) = serde_json::from_str::(&body) { + if let Some(detail) = value + .get("errors") + .and_then(serde_json::Value::as_array) + .and_then(|errors| errors.first()) + .and_then(|entry| entry.get("detail")) + .and_then(serde_json::Value::as_str) + { + bail!("{detail}"); + } + } + + if body.is_empty() { + bail!("request failed with status {status}"); + } + + bail!("request failed with status {status}: {body}"); +} + fn is_not_found_error(err: &progenitor_client::Error) -> bool where E: serde::Serialize + std::fmt::Debug, diff --git a/lib/crates/fabro-config/src/config.rs b/lib/crates/fabro-config/src/config.rs index e731ae0ef..fd3a44c34 100644 --- a/lib/crates/fabro-config/src/config.rs +++ b/lib/crates/fabro-config/src/config.rs @@ -108,6 +108,9 @@ pub struct ConfigLayer { #[serde(default, skip_serializing_if = "Option::is_none")] pub max_concurrent_runs: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub artifact_storage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub web: Option, @@ -168,6 +171,7 @@ impl Combine for ConfigLayer { no_retro: self.no_retro.combine(other.no_retro), storage_dir: self.storage_dir.combine(other.storage_dir), max_concurrent_runs: self.max_concurrent_runs.combine(other.max_concurrent_runs), + artifact_storage: self.artifact_storage.combine(other.artifact_storage), web: self.web.combine(other.web), api: self.api.combine(other.api), features: self.features.combine(other.features), diff --git a/lib/crates/fabro-config/src/effective_settings.rs b/lib/crates/fabro-config/src/effective_settings.rs index 04ffa275c..f1f499320 100644 --- a/lib/crates/fabro-config/src/effective_settings.rs +++ b/lib/crates/fabro-config/src/effective_settings.rs @@ -97,6 +97,7 @@ fn local_daemon_server_overrides_layer(settings: &Settings) -> Result for Settings { no_retro: value.no_retro, storage_dir: value.storage_dir, max_concurrent_runs: value.max_concurrent_runs, + artifact_storage: value.artifact_storage, web: value.web.map(Into::into), api: value.api.map(TryInto::try_into).transpose()?, features: value.features.map(Into::into), diff --git a/lib/crates/fabro-config/src/storage.rs b/lib/crates/fabro-config/src/storage.rs index f12aa4ff0..4fa0c9ce3 100644 --- a/lib/crates/fabro-config/src/storage.rs +++ b/lib/crates/fabro-config/src/storage.rs @@ -58,6 +58,11 @@ impl Storage { pub fn store_dir(&self) -> PathBuf { self.root.join("store") } + + #[must_use] + pub fn artifact_store_dir(&self) -> PathBuf { + self.root.join("artifacts") + } } impl ServerState { @@ -193,6 +198,10 @@ mod tests { storage.store_dir(), std::path::Path::new("/tmp/fabro-data/store") ); + assert_eq!( + storage.artifact_store_dir(), + std::path::Path::new("/tmp/fabro-data/artifacts") + ); assert_eq!( storage.server_state().record_path(), std::path::Path::new("/tmp/fabro-data/server.json") diff --git a/lib/crates/fabro-server/Cargo.toml b/lib/crates/fabro-server/Cargo.toml index 950b22439..db4d1149e 100644 --- a/lib/crates/fabro-server/Cargo.toml +++ b/lib/crates/fabro-server/Cargo.toml @@ -71,6 +71,7 @@ rust-embed.workspace = true regex.workspace = true semver.workspace = true walkdir.workspace = true +multer = "3" [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/lib/crates/fabro-server/src/jwt_auth.rs b/lib/crates/fabro-server/src/jwt_auth.rs index 3a7101609..79adaa30e 100644 --- a/lib/crates/fabro-server/src/jwt_auth.rs +++ b/lib/crates/fabro-server/src/jwt_auth.rs @@ -253,43 +253,48 @@ fn try_cookie(parts: &Parts) -> Result<(), ApiError> { /// When auth is disabled, the extractor accepts all requests. pub struct AuthenticatedService; +pub fn authenticate_service_parts(parts: &Parts) -> Result<(), ApiError> { + let auth_mode = parts + .extensions + .get::() + .expect("AuthMode extension must be added to the router"); + + let strategies = match auth_mode { + AuthMode::Disabled => return Ok(()), + AuthMode::Strategies(strategies) => strategies, + }; + + if strategies.is_empty() { + return Err(ApiError::unauthorized()); + } + + let mut last_err = ApiError::unauthorized(); + + for strategy in strategies { + let result = match strategy { + AuthStrategy::Mtls => try_mtls(parts), + AuthStrategy::Cookie => try_cookie(parts), + AuthStrategy::Jwt { + key, + validation, + allowed_usernames, + } => try_jwt(parts, key, validation, allowed_usernames), + }; + match result { + Ok(()) => return Ok(()), + Err(err) => last_err = err, + } + } + + Err(last_err) +} + impl FromRequestParts for AuthenticatedService { type Rejection = ApiError; async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { - let auth_mode = parts - .extensions - .get::() - .expect("AuthMode extension must be added to the router"); - - let strategies = match auth_mode { - AuthMode::Disabled => return Ok(Self), - AuthMode::Strategies(strategies) => strategies, - }; - - if strategies.is_empty() { - return Err(ApiError::unauthorized()); - } - - let mut last_err = ApiError::unauthorized(); - - for strategy in strategies { - let result = match strategy { - AuthStrategy::Mtls => try_mtls(parts), - AuthStrategy::Cookie => try_cookie(parts), - AuthStrategy::Jwt { - key, - validation, - allowed_usernames, - } => try_jwt(parts, key, validation, allowed_usernames), - }; - match result { - Ok(()) => return Ok(Self), - Err(e) => last_err = e, - } - } - - Err(last_err) + authenticate_service_parts(parts)?; + Ok(Self) } } diff --git a/lib/crates/fabro-server/src/run_manifest.rs b/lib/crates/fabro-server/src/run_manifest.rs index a471d9db3..9f8bfce7a 100644 --- a/lib/crates/fabro-server/src/run_manifest.rs +++ b/lib/crates/fabro-server/src/run_manifest.rs @@ -131,6 +131,7 @@ pub(crate) fn create_run_input(prepared: PreparedManifest) -> CreateRunInput { .as_ref() .map(|git| fabro_github::normalize_repo_origin_url(&git.origin_url)), base_branch: prepared.git.as_ref().map(|git| git.branch.clone()), + artifact_storage: None, provenance: None, } } diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index 55cfbc2f7..202d82012 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -3,10 +3,11 @@ use std::sync::{Arc, RwLock}; use std::time::Duration; use fabro_config::Storage; -use fabro_config::server::resolve_storage_dir; +use fabro_config::server::{ArtifactStorageBackend, resolve_storage_dir}; use fabro_config::user::{active_settings_path, load_settings_config}; use fabro_util::terminal::Styles; use object_store::ObjectStore; +use object_store::aws::AmazonS3Builder; use object_store::local::LocalFileSystem; use object_store::memory::InMemory; use tokio::net::{TcpListener, UnixListener}; @@ -129,6 +130,43 @@ fn build_object_store(store_path: &Path) -> anyhow::Result> build_object_store_with_preference(store_path, use_in_memory_store()) } +fn build_artifact_object_store( + settings: &Settings, + storage: &Storage, +) -> anyhow::Result<(Arc, String)> { + let artifact_settings = settings.artifact_storage.clone().unwrap_or_default(); + + if use_in_memory_store() { + return Ok((Arc::new(InMemory::new()), artifact_settings.prefix)); + } + + match artifact_settings.backend { + ArtifactStorageBackend::Local => { + std::fs::create_dir_all(storage.artifact_store_dir())?; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage.root())?); + Ok((object_store, artifact_settings.prefix)) + } + ArtifactStorageBackend::S3 => { + let bucket = artifact_settings + .bucket + .ok_or_else(|| anyhow::anyhow!("artifact_storage.bucket is required for s3"))?; + let region = artifact_settings + .region + .ok_or_else(|| anyhow::anyhow!("artifact_storage.region is required for s3"))?; + + let mut builder = AmazonS3Builder::from_env() + .with_bucket_name(bucket) + .with_region(region) + .with_virtual_hosted_style_request(!artifact_settings.path_style.unwrap_or(false)); + if let Some(endpoint) = artifact_settings.endpoint { + builder = builder.with_endpoint(endpoint); + } + let object_store = Arc::new(builder.build()?); + Ok((object_store, artifact_settings.prefix)) + } + } +} + /// Start the HTTP API server. /// /// # Errors @@ -215,7 +253,11 @@ pub async fn serve_command( "", Duration::from_millis(1), )); - let artifact_store = fabro_store::ArtifactStore::new(object_store, "artifacts"); + let (artifact_object_store, artifact_prefix) = build_artifact_object_store( + &shared_settings.read().expect("config lock poisoned"), + &storage, + )?; + let artifact_store = fabro_store::ArtifactStore::new(artifact_object_store, artifact_prefix); let state = build_app_state_with_path( Arc::clone(&shared_settings), None, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 8f69e84b9..4c0633d1a 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -1,5 +1,5 @@ use std::collections::{HashMap, HashSet}; -use std::path::{Component, PathBuf}; +use std::path::PathBuf; use std::process::Stdio; use std::str::FromStr; use std::sync::atomic::{AtomicBool, Ordering}; @@ -9,8 +9,8 @@ use std::time::{Duration, Instant}; use crate::bind::Bind; #[cfg(test)] use axum::body::to_bytes; -use axum::extract::{self as axum_extract, Path, Query, State}; -use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header}; +use axum::extract::{self as axum_extract, DefaultBodyLimit, Path, Query, State}; +use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header, request::Parts}; use axum::middleware::{self, Next}; use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::{IntoResponse, Response}; @@ -31,8 +31,8 @@ use fabro_llm::types::{ use fabro_model::{BilledModelUsage, BilledTokenCounts}; use fabro_store::{ArtifactStore, Database, EventEnvelope, EventPayload, StageId}; use fabro_types::{ - EventBody, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, - RunServerProvenance, RunSubjectProvenance, Settings, + EventBody, RunArtifactStorage, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, + RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, Settings, }; use fabro_util::redact::redact_jsonl_line; use fabro_util::version::FABRO_VERSION; @@ -40,7 +40,10 @@ use fabro_workflow::artifacts as workflow_artifacts; use fabro_workflow::error::FabroError; use fabro_workflow::handler::HandlerRegistry; use futures_util::stream; +use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation}; use object_store::memory::InMemory as MemoryObjectStore; +use rand::RngCore; +use sha2::{Digest, Sha256}; use tempfile::NamedTempFile; use tokio::fs; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -63,7 +66,9 @@ use tracing::{error, info}; use crate::demo; use crate::diagnostics; use crate::error::ApiError; -use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedSubject}; +use crate::jwt_auth::{ + AuthMode, AuthenticatedService, AuthenticatedSubject, authenticate_service_parts, +}; use crate::run_manifest; use crate::secret_store::{SecretStore, SecretStoreError}; use crate::static_files; @@ -233,6 +238,46 @@ enum ExecutionResult { const FILE_INTERVIEW_QUESTION_ID: &str = "q-file"; const WORKER_STDERR_LOG: &str = "worker.stderr.log"; const WORKER_CANCEL_GRACE: Duration = Duration::from_secs(5); +const ARTIFACT_UPLOAD_TOKEN_ISSUER: &str = "fabro-server-artifact-upload"; +const ARTIFACT_UPLOAD_TOKEN_SCOPE: &str = "stage_artifacts:upload"; +const ARTIFACT_UPLOAD_TOKEN_TTL_SECS: u64 = 24 * 60 * 60; +const MAX_SINGLE_ARTIFACT_BYTES: u64 = 10 * 1024 * 1024; +const MAX_MULTIPART_ARTIFACTS: usize = 100; +const MAX_MULTIPART_REQUEST_BYTES: u64 = 50 * 1024 * 1024; +const MAX_MULTIPART_MANIFEST_BYTES: usize = 256 * 1024; + +#[derive(Clone)] +struct ArtifactUploadTokenKeys { + encoding: Arc, + decoding: Arc, + validation: Arc, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +struct ArtifactUploadClaims { + iss: String, + iat: u64, + exp: u64, + run_id: String, + scope: String, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +struct ArtifactBatchUploadManifest { + entries: Vec, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +struct ArtifactBatchUploadEntry { + part: String, + path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + sha256: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + expected_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + content_type: Option, +} /// Per-model billing totals. #[derive(Default)] @@ -257,6 +302,7 @@ pub struct AppState { aggregate_billing: Mutex, store: Arc, artifact_store: ArtifactStore, + artifact_upload_tokens: ArtifactUploadTokenKeys, started_at: Instant, max_concurrent_runs: usize, scheduler_notify: Notify, @@ -374,6 +420,31 @@ impl AppState { })) } + fn issue_artifact_upload_token(&self, run_id: &RunId) -> Result { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or(0); + let claims = ArtifactUploadClaims { + iss: ARTIFACT_UPLOAD_TOKEN_ISSUER.to_string(), + iat: now, + exp: now + ARTIFACT_UPLOAD_TOKEN_TTL_SECS, + run_id: run_id.to_string(), + scope: ARTIFACT_UPLOAD_TOKEN_SCOPE.to_string(), + }; + jsonwebtoken::encode( + &Header::new(Algorithm::HS256), + &claims, + &self.artifact_upload_tokens.encoding, + ) + .map_err(|err| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("failed to sign artifact upload token: {err}"), + ) + }) + } + fn begin_shutdown(&self) { self.shutting_down.store(true, Ordering::Relaxed); self.scheduler_notify.notify_waiters(); @@ -384,6 +455,67 @@ impl AppState { } } +fn artifact_upload_token_keys() -> ArtifactUploadTokenKeys { + let mut secret = [0_u8; 32]; + rand::rngs::OsRng.fill_bytes(&mut secret); + + let mut validation = Validation::new(Algorithm::HS256); + validation.set_required_spec_claims(&["iss", "iat", "exp"]); + validation.set_issuer(&[ARTIFACT_UPLOAD_TOKEN_ISSUER]); + + ArtifactUploadTokenKeys { + encoding: Arc::new(EncodingKey::from_secret(&secret)), + decoding: Arc::new(DecodingKey::from_secret(&secret)), + validation: Arc::new(validation), + } +} + +fn maybe_authorize_artifact_upload_token( + parts: &Parts, + run_id: &RunId, + keys: &ArtifactUploadTokenKeys, +) -> Result { + let header = match parts + .headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + { + Some(header) => header, + None => return Ok(false), + }; + let token = match header.strip_prefix("Bearer ") { + Some(token) => token, + None => return Ok(false), + }; + + let claims = + match jsonwebtoken::decode::(token, &keys.decoding, &keys.validation) + { + Ok(token_data) => token_data.claims, + Err(_) => return Ok(false), + }; + + if claims.scope != ARTIFACT_UPLOAD_TOKEN_SCOPE { + return Err(ApiError::forbidden()); + } + if claims.run_id != run_id.to_string() { + return Err(ApiError::forbidden()); + } + + Ok(true) +} + +fn authorize_artifact_upload( + parts: &Parts, + state: &AppState, + run_id: &RunId, +) -> Result<(), ApiError> { + if maybe_authorize_artifact_upload_token(parts, run_id, &state.artifact_upload_tokens)? { + return Ok(()); + } + authenticate_service_parts(parts) +} + fn decode_secret_pem(name: &str, raw: &str) -> Result { if raw.starts_with("-----") { return Ok(raw.to_string()); @@ -553,7 +685,9 @@ fn real_routes() -> Router> { .route("/runs/{id}/stages/{stageId}/turns", get(not_implemented)) .route( "/runs/{id}/stages/{stageId}/artifacts", - get(list_stage_artifacts).post(put_stage_artifact), + get(list_stage_artifacts) + .post(put_stage_artifact) + .layer(DefaultBodyLimit::disable()), ) .route( "/runs/{id}/stages/{stageId}/artifacts/download", @@ -1575,6 +1709,7 @@ pub(crate) fn build_app_state_with_path( aggregate_billing: Mutex::new(BillingAccumulator::default()), store, artifact_store, + artifact_upload_tokens: artifact_upload_token_keys(), started_at: Instant::now(), max_concurrent_runs, scheduler_notify: Notify::new(), @@ -1829,24 +1964,43 @@ fn required_filename(params: ArtifactFilenameParams) -> Result } #[allow(clippy::result_large_err)] -fn validate_relative_artifact_path(kind: &str, value: &str) -> Result { - let mut normalized = PathBuf::new(); - for component in PathBuf::from(value).components() { - match component { - Component::Normal(part) => normalized.push(part), - Component::CurDir => {} - Component::ParentDir | Component::RootDir | Component::Prefix(_) => { - return Err(ApiError::bad_request(format!( - "{kind} must be a relative path without '..'" - )) - .into_response()); - } - } - } - if normalized.as_os_str().is_empty() { +fn validate_relative_artifact_path(kind: &str, value: &str) -> Result { + if value.is_empty() { return Err(ApiError::bad_request(format!("{kind} must not be empty")).into_response()); } - Ok(normalized) + + if value.contains('\\') { + return Err( + ApiError::bad_request(format!("{kind} must not contain backslashes")).into_response(), + ); + } + + let segments = value.split('/').collect::>(); + if segments.iter().any(|segment| segment.is_empty()) { + return Err( + ApiError::bad_request(format!("{kind} must not contain empty path segments")) + .into_response(), + ); + } + if segments + .iter() + .any(|segment| matches!(*segment, "." | "..")) + { + return Err(ApiError::bad_request(format!( + "{kind} must be a relative path without '.' or '..' segments" + )) + .into_response()); + } + + Ok(segments.join("/")) +} + +fn bad_request_response(detail: impl Into) -> Response { + ApiError::bad_request(detail.into()).into_response() +} + +fn payload_too_large_response(detail: impl Into) -> Response { + ApiError::new(StatusCode::PAYLOAD_TOO_LARGE, detail.into()).into_response() } #[allow(clippy::result_large_err)] @@ -2357,10 +2511,15 @@ fn worker_command( .expect("settings lock poisoned") .storage_dir(); let server_target = current_server_target(&storage_dir)?; + let artifact_upload_token = state + .issue_artifact_upload_token(&run_id) + .map_err(|_| anyhow::anyhow!("failed to sign artifact upload token"))?; let mut cmd = Command::new(exe); cmd.arg("__run-worker") .arg("--server") .arg(server_target) + .arg("--artifact-upload-token") + .arg(artifact_upload_token) .arg("--run-dir") .arg(run_dir) .arg("--run-id") @@ -2469,6 +2628,7 @@ async fn create_run( let mut create_input = run_manifest::create_run_input(prepared.clone()); create_input.run_id = Some(run_id); + create_input.artifact_storage = Some(RunArtifactStorage::ObjectStoreV1); create_input.provenance = Some(run_provenance(&headers, &subject)); let created = match Box::pin(operations::create(state.store.as_ref(), create_input)).await { @@ -2874,6 +3034,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { interviewer: Arc::clone(&interviewer) as Arc, run_store: run_store.clone().into(), event_sink: workflow_event::RunEventSink::store(run_store.clone()), + artifact_uploader: None, run_control: None, github_app, on_node: None, @@ -3696,6 +3857,27 @@ async fn read_run_blob( } } +async fn load_run_record( + state: &AppState, + run_id: &RunId, +) -> Result { + let run_store = state + .store + .open_run_reader(run_id) + .await + .map_err(|_| ApiError::not_found("Run not found.").into_response())?; + let run_state = run_store.state().await.map_err(|err| { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + })?; + run_state.run.ok_or_else(|| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "run record missing from store", + ) + .into_response() + }) +} + async fn list_run_artifacts( _auth: AuthenticatedService, State(state): State>, @@ -3705,39 +3887,47 @@ async fn list_run_artifacts( Ok(id) => id, Err(response) => return response, }; - match state.store.open_run_reader(&id).await { - Ok(run_store) => match run_store.state().await { - Ok(run_state) => { - let Some(run) = run_state.run.as_ref() else { - return ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "run record missing from store", - ) - .into_response(); - }; - match scan_run_artifacts(run, &id, None, None) { - Ok(entries) => Json(RunArtifactListResponse { - data: entries - .into_iter() - .map(|entry| RunArtifactEntry { - stage_id: StageId::new(entry.node_slug.clone(), entry.retry) - .to_string(), - node_slug: entry.node_slug, - retry: entry.retry.cast_signed(), - relative_path: entry.relative_path, - size: entry.size.cast_signed(), - }) - .collect(), + let run = match load_run_record(state.as_ref(), &id).await { + Ok(run) => run, + Err(response) => return response, + }; + + if run.uses_object_backed_artifacts() { + return match state.artifact_store.list_for_run(&id).await { + Ok(entries) => Json(RunArtifactListResponse { + data: entries + .into_iter() + .map(|entry| RunArtifactEntry { + stage_id: entry.node.to_string(), + node_slug: entry.node.node_id().to_string(), + retry: entry.node.visit().cast_signed(), + relative_path: entry.filename, + size: entry.size.cast_signed(), }) - .into_response(), - Err(response) => response, - } - } + .collect(), + }) + .into_response(), Err(err) => { ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() } - }, - Err(_) => ApiError::not_found("Run not found.").into_response(), + }; + } + + match scan_run_artifacts(&run, &id, None, None) { + Ok(entries) => Json(RunArtifactListResponse { + data: entries + .into_iter() + .map(|entry| RunArtifactEntry { + stage_id: StageId::new(entry.node_slug.clone(), entry.retry).to_string(), + node_slug: entry.node_slug, + retry: entry.retry.cast_signed(), + relative_path: entry.relative_path, + size: entry.size.cast_signed(), + }) + .collect(), + }) + .into_response(), + Err(response) => response, } } @@ -3754,59 +3944,395 @@ async fn list_stage_artifacts( Ok(stage_id) => stage_id, Err(response) => return response, }; - match state.store.open_run_reader(&id).await { - Ok(run_store) => match run_store.state().await { - Ok(run_state) => { - let Some(run) = run_state.run.as_ref() else { - return ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "run record missing from store", - ) - .into_response(); - }; - match state.artifact_store.list_for_node(&id, &stage_id).await { - Ok(filenames) if !filenames.is_empty() => Json(ArtifactListResponse { - data: filenames - .into_iter() - .map(|filename| ArtifactEntry { filename }) - .collect(), - }) - .into_response(), - Ok(_) => match scan_run_artifacts( - run, - &id, - Some(stage_id.node_id()), - Some(stage_id.visit()), - ) { - Ok(entries) => Json(ArtifactListResponse { - data: entries - .into_iter() - .map(|entry| ArtifactEntry { - filename: entry.relative_path, - }) - .collect(), + let run = match load_run_record(state.as_ref(), &id).await { + Ok(run) => run, + Err(response) => return response, + }; + + match state.artifact_store.list_for_node(&id, &stage_id).await { + Ok(filenames) if run.uses_object_backed_artifacts() || !filenames.is_empty() => { + Json(ArtifactListResponse { + data: filenames + .into_iter() + .map(|filename| ArtifactEntry { filename }) + .collect(), + }) + .into_response() + } + Ok(_) => { + match scan_run_artifacts(&run, &id, Some(stage_id.node_id()), Some(stage_id.visit())) { + Ok(entries) => Json(ArtifactListResponse { + data: entries + .into_iter() + .map(|entry| ArtifactEntry { + filename: entry.relative_path, }) - .into_response(), - Err(response) => response, - }, - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), - } + .collect(), + }) + .into_response(), + Err(response) => response, } - Err(err) => { - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() - } - }, - Err(_) => ApiError::not_found("Run not found.").into_response(), + } + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } } +enum ArtifactUploadContentType { + OctetStream, + Multipart { boundary: String }, +} + +struct ValidatedArtifactBatchEntry { + path: String, + sha256: Option, + expected_bytes: Option, +} + +#[allow(clippy::result_large_err)] +fn artifact_upload_content_type( + headers: &HeaderMap, +) -> Result { + let value = headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| { + ApiError::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "artifact uploads require a supported Content-Type", + ) + .into_response() + })?; + + let mime = value.split(';').next().unwrap_or(value).trim(); + match mime { + "application/octet-stream" => Ok(ArtifactUploadContentType::OctetStream), + "multipart/form-data" => multer::parse_boundary(value) + .map(|boundary| ArtifactUploadContentType::Multipart { boundary }) + .map_err(|err| bad_request_response(format!("invalid multipart boundary: {err}"))), + _ => Err(ApiError::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "artifact uploads only support application/octet-stream or multipart/form-data", + ) + .into_response()), + } +} + +#[allow(clippy::result_large_err)] +fn content_length_from_headers(headers: &HeaderMap) -> Result, Response> { + headers + .get(header::CONTENT_LENGTH) + .map(|value| { + value + .to_str() + .map_err(|err| { + bad_request_response(format!("invalid content-length header: {err}")) + }) + .and_then(|value| { + value.parse::().map_err(|err| { + bad_request_response(format!("invalid content-length header: {err}")) + }) + }) + }) + .transpose() +} + +#[allow(clippy::result_large_err)] +async fn read_multipart_manifest( + field: &mut multer::Field<'_>, +) -> Result { + let mut manifest_bytes = Vec::new(); + while let Some(chunk) = field + .chunk() + .await + .map_err(|err| bad_request_response(format!("invalid multipart body: {err}")))? + { + manifest_bytes.extend_from_slice(&chunk); + if manifest_bytes.len() > MAX_MULTIPART_MANIFEST_BYTES { + return Err(payload_too_large_response( + "multipart manifest exceeds the server limit", + )); + } + } + + serde_json::from_slice(&manifest_bytes) + .map_err(|err| bad_request_response(format!("invalid multipart manifest: {err}"))) +} + +#[allow(clippy::result_large_err)] +fn validate_artifact_batch_manifest( + manifest: ArtifactBatchUploadManifest, +) -> Result, Response> { + if manifest.entries.is_empty() { + return Err(bad_request_response( + "multipart manifest must include at least one artifact entry", + )); + } + if manifest.entries.len() > MAX_MULTIPART_ARTIFACTS { + return Err(payload_too_large_response(format!( + "multipart upload exceeds the {} artifact limit", + MAX_MULTIPART_ARTIFACTS + ))); + } + + let mut entries = HashMap::with_capacity(manifest.entries.len()); + let mut seen_paths = HashSet::new(); + let mut expected_total_bytes = 0_u64; + + for entry in manifest.entries { + if entry.part.is_empty() { + return Err(bad_request_response( + "multipart manifest part names must not be empty", + )); + } + if entry.part == "manifest" { + return Err(bad_request_response( + "multipart manifest part name 'manifest' is reserved", + )); + } + let path = validate_relative_artifact_path("manifest path", &entry.path)?; + if !seen_paths.insert(path.clone()) { + return Err(bad_request_response(format!( + "duplicate artifact path in multipart manifest: {path}" + ))); + } + if let Some(sha256) = entry.sha256.as_ref() { + if sha256.len() != 64 || !sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(bad_request_response(format!( + "invalid sha256 for multipart part {}", + entry.part + ))); + } + } + if let Some(expected_bytes) = entry.expected_bytes { + if expected_bytes > MAX_SINGLE_ARTIFACT_BYTES { + return Err(payload_too_large_response(format!( + "artifact {} exceeds the {} byte limit", + path, MAX_SINGLE_ARTIFACT_BYTES + ))); + } + expected_total_bytes = expected_total_bytes.saturating_add(expected_bytes); + if expected_total_bytes > MAX_MULTIPART_REQUEST_BYTES { + return Err(payload_too_large_response(format!( + "multipart upload exceeds the {} byte limit", + MAX_MULTIPART_REQUEST_BYTES + ))); + } + } + if entries + .insert( + entry.part.clone(), + ValidatedArtifactBatchEntry { + path, + sha256: entry.sha256.map(|value| value.to_ascii_lowercase()), + expected_bytes: entry.expected_bytes, + }, + ) + .is_some() + { + return Err(bad_request_response(format!( + "duplicate multipart part name in manifest: {}", + entry.part + ))); + } + } + + Ok(entries) +} + +async fn upload_stage_artifact_octet_stream( + state: &AppState, + run_id: &RunId, + stage_id: &StageId, + filename: String, + body: axum::body::Body, + content_length: Option, +) -> Response { + let relative_path = match validate_relative_artifact_path("filename", &filename) { + Ok(path) => path, + Err(response) => return response, + }; + + if content_length.is_some_and(|length| length > MAX_SINGLE_ARTIFACT_BYTES) { + return payload_too_large_response(format!( + "artifact exceeds the {} byte limit", + MAX_SINGLE_ARTIFACT_BYTES + )); + } + + let mut writer = match state + .artifact_store + .writer(run_id, stage_id, &relative_path) + { + Ok(writer) => writer, + Err(err) => { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + }; + + let mut bytes_written = 0_u64; + let mut data_stream = body.into_data_stream(); + while let Some(chunk) = data_stream.next().await { + let chunk = match chunk + .map_err(|err| bad_request_response(format!("invalid request body: {err}"))) + { + Ok(chunk) => chunk, + Err(response) => return response, + }; + bytes_written = + bytes_written.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX)); + if bytes_written > MAX_SINGLE_ARTIFACT_BYTES { + return payload_too_large_response(format!( + "artifact exceeds the {} byte limit", + MAX_SINGLE_ARTIFACT_BYTES + )); + } + if let Err(err) = writer.write_all(&chunk).await { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + } + + match writer.shutdown().await { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } + } +} + +async fn upload_stage_artifact_multipart( + state: &AppState, + run_id: &RunId, + stage_id: &StageId, + boundary: String, + body: axum::body::Body, +) -> Response { + let mut multipart = multer::Multipart::new(body.into_data_stream(), boundary); + let Some(mut manifest_field) = (match multipart + .next_field() + .await + .map_err(|err| bad_request_response(format!("invalid multipart body: {err}"))) + { + Ok(field) => field, + Err(response) => return response, + }) else { + return bad_request_response("multipart upload must begin with a manifest part"); + }; + + if manifest_field.name() != Some("manifest") { + return bad_request_response("multipart upload must begin with a manifest part"); + } + + let manifest = match read_multipart_manifest(&mut manifest_field).await { + Ok(manifest) => manifest, + Err(response) => return response, + }; + drop(manifest_field); + let mut expected_parts = match validate_artifact_batch_manifest(manifest) { + Ok(entries) => entries, + Err(response) => return response, + }; + let mut total_bytes = 0_u64; + + while let Some(mut field) = match multipart + .next_field() + .await + .map_err(|err| bad_request_response(format!("invalid multipart body: {err}"))) + { + Ok(field) => field, + Err(response) => return response, + } { + let Some(part_name) = field.name().map(ToOwned::to_owned) else { + return bad_request_response("multipart file parts must be named"); + }; + let Some(entry) = expected_parts.remove(&part_name) else { + return bad_request_response(format!("unexpected multipart part: {part_name}")); + }; + + let mut writer = match state.artifact_store.writer(run_id, stage_id, &entry.path) { + Ok(writer) => writer, + Err(err) => { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + }; + let mut bytes_written = 0_u64; + let mut sha256 = Sha256::new(); + + while let Some(chunk) = match field + .chunk() + .await + .map_err(|err| bad_request_response(format!("invalid multipart body: {err}"))) + { + Ok(chunk) => chunk, + Err(response) => return response, + } { + let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX); + bytes_written = bytes_written.saturating_add(chunk_len); + total_bytes = total_bytes.saturating_add(chunk_len); + + if bytes_written > MAX_SINGLE_ARTIFACT_BYTES { + return payload_too_large_response(format!( + "artifact {} exceeds the {} byte limit", + entry.path, MAX_SINGLE_ARTIFACT_BYTES + )); + } + if total_bytes > MAX_MULTIPART_REQUEST_BYTES { + return payload_too_large_response(format!( + "multipart upload exceeds the {} byte limit", + MAX_MULTIPART_REQUEST_BYTES + )); + } + + sha256.update(&chunk); + if let Err(err) = writer.write_all(&chunk).await { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + } + + if let Some(expected_bytes) = entry.expected_bytes { + if bytes_written != expected_bytes { + return bad_request_response(format!( + "multipart part {part_name} expected {expected_bytes} bytes but received {bytes_written}" + )); + } + } + if let Some(expected_sha256) = entry.sha256.as_ref() { + let actual_sha256 = hex::encode(sha256.finalize()); + if actual_sha256 != *expected_sha256 { + return bad_request_response(format!( + "multipart part {part_name} sha256 did not match manifest" + )); + } + } + + if let Err(err) = writer.shutdown().await { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + } + + if !expected_parts.is_empty() { + let mut missing = expected_parts.into_keys().collect::>(); + missing.sort(); + return bad_request_response(format!( + "multipart upload is missing part(s): {}", + missing.join(", ") + )); + } + + StatusCode::NO_CONTENT.into_response() +} + async fn put_stage_artifact( - _auth: AuthenticatedService, State(state): State>, Path((id, stage_id)): Path<(String, String)>, Query(params): Query, - body: Bytes, + request: axum_extract::Request, ) -> Response { let id = match parse_run_id_path(&id) { Ok(id) => id, @@ -3816,22 +4342,45 @@ async fn put_stage_artifact( Ok(stage_id) => stage_id, Err(response) => return response, }; - let filename = match required_filename(params) { - Ok(filename) => filename, + let (parts, body) = request.into_parts(); + + if let Err(err) = authorize_artifact_upload(&parts, state.as_ref(), &id) { + return err.into_response(); + } + if let Err(response) = load_run_record(state.as_ref(), &id).await.map(|_| ()) { + return response; + } + + let content_length = match content_length_from_headers(&parts.headers) { + Ok(length) => length, Err(response) => return response, }; - match state.store.open_run_reader(&id).await { - Ok(_) => match state - .artifact_store - .put(&id, &stage_id, &filename, &body) + match artifact_upload_content_type(&parts.headers) { + Ok(ArtifactUploadContentType::OctetStream) => { + let filename = match required_filename(params) { + Ok(filename) => filename, + Err(response) => return response, + }; + upload_stage_artifact_octet_stream( + state.as_ref(), + &id, + &stage_id, + filename, + body, + content_length, + ) .await - { - Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => { - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } + Ok(ArtifactUploadContentType::Multipart { boundary }) => { + if content_length.is_some_and(|length| length > MAX_MULTIPART_REQUEST_BYTES) { + return payload_too_large_response(format!( + "multipart upload exceeds the {} byte limit", + MAX_MULTIPART_REQUEST_BYTES + )); } - }, - Err(_) => ApiError::not_found("Run not found.").into_response(), + upload_stage_artifact_multipart(state.as_ref(), &id, &stage_id, boundary, body).await + } + Err(response) => response, } } @@ -3853,48 +4402,41 @@ async fn get_stage_artifact( Ok(filename) => filename, Err(response) => return response, }; - match state.store.open_run_reader(&id).await { - Ok(run_store) => match run_store.state().await { - Ok(run_state) => { - let Some(run) = run_state.run.as_ref() else { - return ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "run record missing from store", - ) - .into_response(); - }; - match state.artifact_store.get(&id, &stage_id, &filename).await { - Ok(Some(bytes)) => octet_stream_response(bytes), - Ok(None) => { - let relative_path = - match validate_relative_artifact_path("filename", &filename) { - Ok(path) => path, - Err(response) => return response, - }; - let artifact_path = run_artifacts_dir(run, &id) - .join(stage_id.node_id()) - .join(format!("retry_{}", stage_id.visit())) - .join(relative_path); - match std::fs::read(&artifact_path) { - Ok(bytes) => octet_stream_response(Bytes::from(bytes)), - Err(err) if err.kind() == std::io::ErrorKind::NotFound => { - ApiError::not_found("Artifact not found.").into_response() - } - Err(err) => { - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response() - } - } - } - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + let relative_path = match validate_relative_artifact_path("filename", &filename) { + Ok(path) => path, + Err(response) => return response, + }; + let run = match load_run_record(state.as_ref(), &id).await { + Ok(run) => run, + Err(response) => return response, + }; + + match state + .artifact_store + .get(&id, &stage_id, &relative_path) + .await + { + Ok(Some(bytes)) => octet_stream_response(bytes), + Ok(None) if run.uses_object_backed_artifacts() => { + ApiError::not_found("Artifact not found.").into_response() + } + Ok(None) => { + let artifact_path = run_artifacts_dir(&run, &id) + .join(stage_id.node_id()) + .join(format!("retry_{}", stage_id.visit())) + .join(&relative_path); + match std::fs::read(&artifact_path) { + Ok(bytes) => octet_stream_response(Bytes::from(bytes)), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + ApiError::not_found("Artifact not found.").into_response() } + Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(), } - Err(err) => { - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() - } - }, - Err(_) => ApiError::not_found("Run not found.").into_response(), + } + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } } @@ -4878,9 +5420,7 @@ mod tests { Body::from(serde_json::to_string(&minimal_manifest_json(dot_source)).unwrap()) } - /// Create a run via POST /runs, then start it via POST /runs/{id}/start. - /// Returns the run_id string. - async fn create_and_start_run(app: &Router, dot_source: &str) -> String { + async fn create_run(app: &Router, dot_source: &str) -> String { let req = Request::builder() .method("POST") .uri(api("/runs")) @@ -4889,7 +5429,42 @@ mod tests { .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap().to_string(); + body["id"].as_str().unwrap().to_string() + } + + fn multipart_body( + boundary: &str, + manifest: &serde_json::Value, + files: &[(&str, &str, &[u8])], + ) -> Body { + let mut body = Vec::new(); + body.extend_from_slice(format!("--{boundary}\r\n").as_bytes()); + body.extend_from_slice(b"Content-Disposition: form-data; name=\"manifest\"\r\n"); + body.extend_from_slice(b"Content-Type: application/json\r\n\r\n"); + body.extend_from_slice(serde_json::to_string(manifest).unwrap().as_bytes()); + body.extend_from_slice(b"\r\n"); + + for (part, filename, bytes) in files { + body.extend_from_slice(format!("--{boundary}\r\n").as_bytes()); + body.extend_from_slice( + format!( + "Content-Disposition: form-data; name=\"{part}\"; filename=\"{filename}\"\r\n" + ) + .as_bytes(), + ); + body.extend_from_slice(b"Content-Type: application/octet-stream\r\n\r\n"); + body.extend_from_slice(bytes); + body.extend_from_slice(b"\r\n"); + } + + body.extend_from_slice(format!("--{boundary}--\r\n").as_bytes()); + Body::from(body) + } + + /// Create a run via POST /runs, then start it via POST /runs/{id}/start. + /// Returns the run_id string. + async fn create_and_start_run(app: &Router, dot_source: &str) -> String { + let run_id = create_run(app, dot_source).await; let req = Request::builder() .method("POST") @@ -4901,6 +5476,32 @@ mod tests { run_id } + async fn create_legacy_run(state: &Arc, settings: &Settings) -> RunId { + operations::create( + state.store.as_ref(), + operations::CreateRunInput { + workflow: operations::WorkflowInput::DotSource { + source: MINIMAL_DOT.to_string(), + base_dir: None, + }, + settings: settings.clone(), + cwd: PathBuf::from("/tmp"), + workflow_slug: None, + workflow_path: None, + workflow_bundle: None, + run_id: None, + host_repo_path: None, + repo_origin_url: None, + base_branch: None, + artifact_storage: None, + provenance: None, + }, + ) + .await + .unwrap() + .run_id + } + async fn create_durable_run_with_events( state: &Arc, run_id: RunId, @@ -5525,16 +6126,7 @@ mod tests { let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(manifest_body(MINIMAL_DOT)) - .unwrap(); - - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap(); + let run_id = create_run(&app, MINIMAL_DOT).await; let stage_id = "code@2"; let req = Request::builder() @@ -5546,7 +6138,11 @@ mod tests { .body(Body::from("fn main() {}")) .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::NO_CONTENT); + if response.status() != StatusCode::NO_CONTENT { + let status = response.status(); + let body = body_json(response.into_body()).await; + panic!("expected 204, got {status}: {body}"); + } let req = Request::builder() .method("GET") @@ -5571,6 +6167,288 @@ mod tests { assert_eq!(&bytes[..], b"fn main() {}"); } + #[tokio::test] + async fn create_run_marks_object_backed_artifacts() { + let state = create_app_state(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT) + .await + .parse::() + .unwrap(); + let run_state = state + .store + .open_run_reader(&run_id) + .await + .unwrap() + .state() + .await + .unwrap(); + + assert!( + run_state + .run + .as_ref() + .unwrap() + .uses_object_backed_artifacts() + ); + } + + #[tokio::test] + async fn stage_artifact_upload_rejects_invalid_filename() { + let state = create_app_state(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT).await; + + let req = Request::builder() + .method("POST") + .uri(api(&format!( + "/runs/{run_id}/stages/code@2/artifacts?filename=../escape.txt" + ))) + .header("content-type", "application/octet-stream") + .body(Body::from("nope")) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn stage_artifacts_multipart_round_trip() { + let state = create_app_state(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT).await; + let stage_id = "code@2"; + let source_bytes = b"fn main() {}\n"; + let log_bytes = b"build ok\n"; + let manifest = serde_json::json!({ + "entries": [ + { + "part": "file1", + "path": "src/lib.rs", + "sha256": hex::encode(Sha256::digest(source_bytes)), + "expected_bytes": source_bytes.len(), + "content_type": "text/plain" + }, + { + "part": "file2", + "path": "logs/output.txt", + "sha256": hex::encode(Sha256::digest(log_bytes)), + "expected_bytes": log_bytes.len(), + "content_type": "text/plain" + } + ] + }); + let boundary = "fabro-test-boundary"; + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/stages/{stage_id}/artifacts"))) + .header( + "content-type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(multipart_body( + boundary, + &manifest, + &[ + ("file1", "src/lib.rs", source_bytes), + ("file2", "logs/output.txt", log_bytes), + ], + )) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + if response.status() != StatusCode::NO_CONTENT { + let status = response.status(); + let body = body_json(response.into_body()).await; + panic!("expected 204, got {status}: {body}"); + } + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/stages/{stage_id}/artifacts"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = body_json(response.into_body()).await; + assert_eq!(body["data"][0]["filename"], "logs/output.txt"); + assert_eq!(body["data"][1]["filename"], "src/lib.rs"); + + let req = Request::builder() + .method("GET") + .uri(api(&format!( + "/runs/{run_id}/stages/{stage_id}/artifacts/download?filename=logs/output.txt" + ))) + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(&bytes[..], log_bytes); + } + + #[tokio::test] + async fn stage_artifacts_multipart_requires_manifest_first() { + let state = create_app_state(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT).await; + let boundary = "fabro-test-boundary"; + let body = format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"file1\"; filename=\"src/lib.rs\"\r\n\r\nfn main() {{}}\r\n--{boundary}\r\nContent-Disposition: form-data; name=\"manifest\"\r\nContent-Type: application/json\r\n\r\n{{\"entries\":[{{\"part\":\"file1\",\"path\":\"src/lib.rs\"}}]}}\r\n--{boundary}--\r\n" + ); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/stages/code@2/artifacts"))) + .header( + "content-type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(body)) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn object_backed_runs_do_not_fallback_to_scratch_artifacts() { + let temp = tempfile::tempdir().unwrap(); + let mut settings = dry_run_settings(); + settings.storage_dir = Some(temp.path().join("storage")); + let state = create_app_state_with_options(settings.clone(), 5); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT) + .await + .parse::() + .unwrap(); + let artifact_path = Storage::new(settings.storage_dir()) + .run_scratch(&run_id) + .artifact_files_dir() + .join("code") + .join("retry_2") + .join("src/lib.rs"); + std::fs::create_dir_all(artifact_path.parent().unwrap()).unwrap(); + std::fs::write(&artifact_path, "legacy scratch only").unwrap(); + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/stages/code@2/artifacts"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = body_json(response.into_body()).await; + assert_eq!(body["data"].as_array().unwrap().len(), 0); + + let req = Request::builder() + .method("GET") + .uri(api(&format!( + "/runs/{run_id}/stages/code@2/artifacts/download?filename=src/lib.rs" + ))) + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn legacy_runs_fallback_to_scratch_artifacts() { + let temp = tempfile::tempdir().unwrap(); + let mut settings = dry_run_settings(); + settings.storage_dir = Some(temp.path().join("storage")); + let state = create_app_state_with_options(settings.clone(), 5); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_legacy_run(&state, &settings).await; + let artifact_path = Storage::new(settings.storage_dir()) + .run_scratch(&run_id) + .artifact_files_dir() + .join("code") + .join("retry_2") + .join("src/lib.rs"); + let retry_dir = artifact_path + .parent() + .unwrap() + .parent() + .unwrap() + .to_path_buf(); + std::fs::create_dir_all(artifact_path.parent().unwrap()).unwrap(); + std::fs::write(&artifact_path, "legacy scratch only").unwrap(); + std::fs::write( + retry_dir.join("manifest.json"), + serde_json::to_string( + &fabro_workflow::artifact_snapshot::ArtifactCollectionSummary { + files_copied: 1, + total_bytes: u64::try_from(b"legacy scratch only".len()).unwrap(), + files_skipped: 0, + download_errors: 0, + hash_errors: 0, + captured_assets: vec![ + fabro_workflow::artifact_snapshot::CapturedArtifactInfo { + path: "src/lib.rs".to_string(), + mime: "text/plain".to_string(), + content_md5: "0".repeat(32), + content_sha256: "0".repeat(64), + bytes: u64::try_from(b"legacy scratch only".len()).unwrap(), + }, + ], + }, + ) + .unwrap(), + ) + .unwrap(); + let run_state = state + .store + .open_run_reader(&run_id) + .await + .unwrap() + .state() + .await + .unwrap(); + assert!( + !run_state + .run + .as_ref() + .unwrap() + .uses_object_backed_artifacts() + ); + let scanned = workflow_artifacts::scan_artifacts( + &Storage::new(settings.storage_dir()) + .run_scratch(&run_id) + .artifact_files_dir(), + Some("code"), + Some(2), + ) + .unwrap(); + assert_eq!(scanned.len(), 1); + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/stages/code@2/artifacts"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = body_json(response.into_body()).await; + assert_eq!(body["data"][0]["filename"], "src/lib.rs"); + + let req = Request::builder() + .method("GET") + .uri(api(&format!( + "/runs/{run_id}/stages/code@2/artifacts/download?filename=src/lib.rs" + ))) + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(&bytes[..], b"legacy scratch only"); + } + #[tokio::test] async fn create_run_returns_submitted() { let state = create_app_state(); diff --git a/lib/crates/fabro-store/src/artifact_store.rs b/lib/crates/fabro-store/src/artifact_store.rs index a9e8f086a..f277a1c3d 100644 --- a/lib/crates/fabro-store/src/artifact_store.rs +++ b/lib/crates/fabro-store/src/artifact_store.rs @@ -4,17 +4,20 @@ use bytes::Bytes; use futures::StreamExt; use object_store::{ObjectStore, path::Path as ObjectPath}; use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, percent_decode_str, utf8_percent_encode}; +use tokio::io::AsyncWriteExt; use crate::{Result, StageId, StoreError}; use fabro_types::RunId; const ARTIFACT_SEGMENT_ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC.remove(b'.').remove(b'_').remove(b'-'); +const STREAM_BUFFER_BYTES: usize = 1024 * 1024; #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct NodeArtifact { pub node: StageId, pub filename: String, + pub size: u64, } #[derive(Clone)] @@ -54,6 +57,45 @@ impl ArtifactStore { Ok(()) } + pub fn writer( + &self, + run_id: &RunId, + node: &StageId, + filename: &str, + ) -> Result { + let path = self.artifact_path(run_id, node, filename)?; + Ok(object_store::buffered::BufWriter::with_capacity( + Arc::clone(&self.object_store), + path, + STREAM_BUFFER_BYTES, + )) + } + + pub async fn put_stream( + &self, + run_id: &RunId, + node: &StageId, + filename: &str, + mut stream: S, + ) -> Result<()> + where + S: futures::Stream> + Unpin, + { + let mut writer = self.writer(run_id, node, filename)?; + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + writer + .write_all(&chunk) + .await + .map_err(|err| StoreError::Other(format!("artifact write failed: {err}")))?; + } + writer + .shutdown() + .await + .map_err(|err| StoreError::Other(format!("artifact finalize failed: {err}")))?; + Ok(()) + } + pub async fn get( &self, run_id: &RunId, @@ -73,7 +115,11 @@ impl ArtifactStore { let mut stream = self.object_store.list(Some(&prefix)); let mut artifacts = Vec::new(); while let Some(meta) = stream.next().await.transpose()? { - artifacts.push(decode_artifact_location(&prefix, &meta.location)?); + artifacts.push(decode_artifact_location( + &prefix, + &meta.location, + meta.size, + )?); } artifacts.sort(); Ok(artifacts) @@ -166,7 +212,11 @@ fn decode_path_segment(kind: &str, value: &str) -> Result { .map_err(|err| StoreError::Other(format!("invalid {kind}: {err}"))) } -fn decode_artifact_location(prefix: &ObjectPath, location: &ObjectPath) -> Result { +fn decode_artifact_location( + prefix: &ObjectPath, + location: &ObjectPath, + size: u64, +) -> Result { let mut parts = location.prefix_match(prefix).ok_or_else(|| { StoreError::Other(format!( "artifact location {location} does not match expected prefix {prefix}" @@ -199,6 +249,7 @@ fn decode_artifact_location(prefix: &ObjectPath, location: &ObjectPath) -> Resul Ok(NodeArtifact { node: StageId::new(node_id, visit), filename: filename_segments.join("/"), + size, }) } @@ -260,10 +311,37 @@ mod tests { vec![NodeArtifact { node, filename: filename.to_string(), + size: 5, }] ); } + #[tokio::test] + async fn put_stream_round_trips_chunked_writes() { + let store = test_store(); + let run_id = fixtures::RUN_1; + let node = StageId::new("build", 2); + let filename = "logs/output.txt"; + + store + .put_stream( + &run_id, + &node, + filename, + futures::stream::iter(vec![ + Ok(Bytes::from_static(b"hello ")), + Ok(Bytes::from_static(b"world")), + ]), + ) + .await + .unwrap(); + + assert_eq!( + store.get(&run_id, &node, filename).await.unwrap(), + Some(Bytes::from_static(b"hello world")) + ); + } + #[tokio::test] async fn rejects_invalid_relative_filenames() { let store = test_store(); diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index bfed74152..0402db3af 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -85,6 +85,7 @@ impl RunProjection { repo_origin_url: props.repo_origin_url.clone(), base_branch: props.base_branch.clone(), labels, + artifact_storage: props.artifact_storage, provenance: props.provenance.clone(), }); self.graph_source.clone_from(&props.workflow_source); diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 3ae03d978..cecafad7c 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -288,6 +288,7 @@ mod tests { repo_origin_url: Some("https://github.com/fabro-sh/fabro".to_string()), base_branch: Some("main".to_string()), labels: std::collections::HashMap::from([("team".to_string(), "infra".to_string())]), + artifact_storage: None, provenance: None, } } diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 52d0d8dda..8de31a10b 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -39,15 +39,15 @@ pub use retro::{ OpenItemKind, Retro, RetroNarrative, SmoothnessRating, StageRetro, }; pub use run::{ - RunAuthMethod, RunClientProvenance, RunProvenance, RunRecord, RunServerProvenance, - RunSubjectProvenance, + RunArtifactStorage, RunAuthMethod, RunClientProvenance, RunProvenance, RunRecord, + RunServerProvenance, RunSubjectProvenance, }; pub use run_blob_id::RunBlobId; pub use run_event::{EventBody, RunEvent, RunNoticeLevel}; pub use run_id::RunId; pub use run_id::fixtures; pub use sandbox_record::SandboxRecord; -pub use settings::Settings; +pub use settings::{ArtifactStorageBackend, ArtifactStorageSettings, Settings}; pub use stage_id::StageId; pub use start::StartRecord; pub use status::{ diff --git a/lib/crates/fabro-types/src/run.rs b/lib/crates/fabro-types/src/run.rs index cafef8d28..e25e31f92 100644 --- a/lib/crates/fabro-types/src/run.rs +++ b/lib/crates/fabro-types/src/run.rs @@ -48,6 +48,12 @@ pub struct RunProvenance { pub subject: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunArtifactStorage { + ObjectStoreV1, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RunRecord { pub run_id: RunId, @@ -65,5 +71,17 @@ pub struct RunRecord { #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub labels: HashMap, #[serde(default, skip_serializing_if = "Option::is_none")] + pub artifact_storage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub provenance: Option, } + +impl RunRecord { + #[must_use] + pub fn uses_object_backed_artifacts(&self) -> bool { + matches!( + self.artifact_storage, + Some(RunArtifactStorage::ObjectStoreV1) + ) + } +} diff --git a/lib/crates/fabro-types/src/run_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs index bae55712a..bf5fcd2a0 100644 --- a/lib/crates/fabro-types/src/run_event/run.rs +++ b/lib/crates/fabro-types/src/run_event/run.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; -use crate::{Graph, RunControlAction, RunProvenance, Settings, StatusReason}; +use crate::{Graph, RunArtifactStorage, RunControlAction, RunProvenance, Settings, StatusReason}; use super::{BilledTokenCounts, RunNoticeLevel}; @@ -29,6 +29,8 @@ pub struct RunCreatedProps { #[serde(default, skip_serializing_if = "Option::is_none")] pub db_prefix: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + pub artifact_storage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub provenance: Option, } diff --git a/lib/crates/fabro-types/src/settings/mod.rs b/lib/crates/fabro-types/src/settings/mod.rs index f0beae569..83e6473c1 100644 --- a/lib/crates/fabro-types/src/settings/mod.rs +++ b/lib/crates/fabro-types/src/settings/mod.rs @@ -26,9 +26,9 @@ pub use sandbox::{ LocalSandboxSettings, SandboxSettings, WorktreeMode, }; pub use server::{ - ApiAuthStrategy, ApiSettings, AuthProvider, AuthSettings, FeaturesSettings, GitAuthorSettings, - GitProvider, GitSettings, LogSettings, TlsSettings, WebSettings, WebhookSettings, - WebhookStrategy, + ApiAuthStrategy, ApiSettings, ArtifactStorageBackend, ArtifactStorageSettings, AuthProvider, + AuthSettings, FeaturesSettings, GitAuthorSettings, GitProvider, GitSettings, LogSettings, + TlsSettings, WebSettings, WebhookSettings, WebhookStrategy, }; pub use user::{ClientTlsSettings, ExecSettings, OutputFormat, PermissionLevel, ServerSettings}; @@ -91,6 +91,8 @@ pub struct Settings { #[serde(default, skip_serializing_if = "Option::is_none")] pub max_concurrent_runs: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + pub artifact_storage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub web: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub api: Option, diff --git a/lib/crates/fabro-types/src/settings/server.rs b/lib/crates/fabro-types/src/settings/server.rs index 48cc4bf98..2ad6437a1 100644 --- a/lib/crates/fabro-types/src/settings/server.rs +++ b/lib/crates/fabro-types/src/settings/server.rs @@ -2,6 +2,10 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; +fn default_artifact_storage_prefix() -> String { + "artifacts".to_string() +} + #[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)] #[serde(rename_all = "snake_case")] pub enum AuthProvider { @@ -124,3 +128,40 @@ pub struct FeaturesSettings { pub struct LogSettings { pub level: Option, } + +#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactStorageBackend { + #[default] + Local, + S3, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize, crate::Combine)] +pub struct ArtifactStorageSettings { + #[serde(default)] + pub backend: ArtifactStorageBackend, + #[serde(default = "default_artifact_storage_prefix")] + pub prefix: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bucket: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub region: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub endpoint: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub path_style: Option, +} + +impl Default for ArtifactStorageSettings { + fn default() -> Self { + Self { + backend: ArtifactStorageBackend::Local, + prefix: default_artifact_storage_prefix(), + bucket: None, + region: None, + endpoint: None, + path_style: None, + } + } +} diff --git a/lib/crates/fabro-workflow/src/artifact_upload.rs b/lib/crates/fabro-workflow/src/artifact_upload.rs new file mode 100644 index 000000000..7fffec8b0 --- /dev/null +++ b/lib/crates/fabro-workflow/src/artifact_upload.rs @@ -0,0 +1,17 @@ +use std::path::Path; + +use anyhow::Result; +use async_trait::async_trait; +use fabro_types::StageId; + +use crate::artifact_snapshot::CapturedArtifactInfo; + +#[async_trait] +pub trait StageArtifactUploader: Send + Sync { + async fn upload_stage_artifacts( + &self, + stage_id: &StageId, + artifact_capture_dir: &Path, + artifacts: &[CapturedArtifactInfo], + ) -> Result<()>; +} diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index e40f2c549..437869fed 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -54,6 +54,8 @@ pub enum Event { #[serde(default, skip_serializing_if = "Option::is_none")] db_prefix: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + artifact_storage: Option<::fabro_types::RunArtifactStorage>, + #[serde(default, skip_serializing_if = "Option::is_none")] provenance: Option<::fabro_types::RunProvenance>, }, WorkflowRunStarted { @@ -1360,6 +1362,7 @@ fn event_body_from_event(event: &Event) -> EventBody { base_branch, workflow_slug, db_prefix, + artifact_storage, provenance, .. } => EventBody::RunCreated(fabro_types::RunCreatedProps { @@ -1375,6 +1378,7 @@ fn event_body_from_event(event: &Event) -> EventBody { base_branch: base_branch.clone(), workflow_slug: workflow_slug.clone(), db_prefix: db_prefix.clone(), + artifact_storage: *artifact_storage, provenance: provenance.clone(), }), Event::WorkflowRunStarted { diff --git a/lib/crates/fabro-workflow/src/handler/manager_loop.rs b/lib/crates/fabro-workflow/src/handler/manager_loop.rs index 9aca0778e..6e2eb879b 100644 --- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs @@ -258,6 +258,7 @@ impl Handler for SubWorkflowHandler { sandbox, registry, on_node: None, + artifact_uploader: None, run_control: None, hook_runner, env, diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index 578e7ce6c..ca73b5bde 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -114,6 +114,7 @@ pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap< #[doc(hidden)] pub mod artifact; pub mod artifact_snapshot; +pub mod artifact_upload; pub mod artifacts; pub(crate) mod condition; pub mod context; diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 8d03313d8..945f705bb 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -1,8 +1,10 @@ use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; use async_trait::async_trait; +use fabro_types::StageId; use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle}; @@ -11,6 +13,7 @@ use fabro_core::state::ExecutionState; use crate::artifact::{offload_large_values, sync_artifacts_to_env}; use crate::artifact_snapshot::collect_artifacts; +use crate::artifact_upload::StageArtifactUploader; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; @@ -23,6 +26,12 @@ type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; type WfNodeDecision = NodeDecision>; +const ARTIFACT_UPLOAD_RETRY_DELAYS: [Duration; 3] = [ + Duration::from_millis(100), + Duration::from_millis(250), + Duration::from_millis(500), +]; + /// Sub-lifecycle responsible for artifact collection, offloading, and syncing. pub(crate) struct ArtifactLifecycle { pub sandbox: Arc, @@ -31,6 +40,7 @@ pub(crate) struct ArtifactLifecycle { pub emitter: Arc, pub artifacts_dir: PathBuf, pub artifact_globs: Vec, + pub artifact_uploader: Option>, pub captured_artifact_count: Arc, /// Per-attempt state: epoch seconds when the attempt started. attempt_start_epoch: std::sync::Mutex>, @@ -45,6 +55,7 @@ impl ArtifactLifecycle { emitter: Arc, artifacts_dir: PathBuf, artifact_globs: Vec, + artifact_uploader: Option>, captured_artifact_count: Arc, ) -> Self { Self { @@ -54,6 +65,7 @@ impl ArtifactLifecycle { emitter, artifacts_dir, artifact_globs, + artifact_uploader, captured_artifact_count, attempt_start_epoch: std::sync::Mutex::new(None), } @@ -113,6 +125,18 @@ impl RunLifecycle for ArtifactLifecycle { .await { Ok(summary) if summary.files_copied > 0 => { + let stage_id = StageId::new(node_id.to_string(), ctx.attempt); + if let Err(err) = self + .upload_artifacts(&stage_id, &artifact_capture_dir, &summary.captured_assets) + .await + { + self.emitter.emit(&Event::RunNotice { + level: RunNoticeLevel::Warn, + code: "artifact_upload_failed".to_string(), + message: format!("[node: {node_id}] artifact upload failed: {err}"), + }); + return Ok(()); + } for asset in &summary.captured_assets { self.captured_artifact_count.fetch_add(1, Ordering::Relaxed); self.emitter.emit(&Event::ArtifactCaptured { @@ -177,3 +201,33 @@ impl RunLifecycle for ArtifactLifecycle { Ok(()) } } + +impl ArtifactLifecycle { + async fn upload_artifacts( + &self, + stage_id: &StageId, + artifact_capture_dir: &std::path::Path, + artifacts: &[crate::artifact_snapshot::CapturedArtifactInfo], + ) -> Result<(), String> { + let Some(uploader) = self.artifact_uploader.as_ref() else { + return Ok(()); + }; + + let mut last_error = None; + for attempt in 0..=ARTIFACT_UPLOAD_RETRY_DELAYS.len() { + match uploader + .upload_stage_artifacts(stage_id, artifact_capture_dir, artifacts) + .await + { + Ok(()) => return Ok(()), + Err(err) => last_error = Some(err.to_string()), + } + + if let Some(delay) = ARTIFACT_UPLOAD_RETRY_DELAYS.get(attempt) { + tokio::time::sleep(*delay).await; + } + } + + Err(last_error.unwrap_or_else(|| "artifact upload failed".to_string())) + } +} diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index a74cee0b5..01d47d07c 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -24,6 +24,7 @@ use fabro_core::lifecycle::{ use fabro_core::outcome::NodeResult; use fabro_core::state::ExecutionState; +use crate::artifact_upload::StageArtifactUploader; use crate::context; use crate::error::{FailureSignature, FailureSignatureExt}; use crate::event::Emitter; @@ -85,6 +86,7 @@ impl WorkflowLifecycle { graph: Arc, run_dir: &PathBuf, run_store: &RunStoreHandle, + artifact_uploader: Option>, run_options: &Arc, is_resume: bool, on_node: crate::OnNodeCallback, @@ -163,6 +165,7 @@ impl WorkflowLifecycle { Arc::clone(emitter), run_scratch.artifact_files_dir(), run_options.artifact_globs().to_vec(), + artifact_uploader, captured_artifact_count, ); diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 6d8f0241b..0edbf44fe 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -3,7 +3,7 @@ use fabro_graphviz::graph::{AttrValue, Graph}; use fabro_model::{Catalog, Provider}; use fabro_sandbox::SandboxProvider; use fabro_store::Database; -use fabro_types::{RunId, RunProvenance, Settings}; +use fabro_types::{RunArtifactStorage, RunId, RunProvenance, Settings}; use std::collections::BTreeMap; use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -35,6 +35,7 @@ pub struct CreateRunInput { pub host_repo_path: Option, pub repo_origin_url: Option, pub base_branch: Option, + pub artifact_storage: Option, pub provenance: Option, } @@ -56,6 +57,7 @@ struct PersistCreateOptions { working_directory: PathBuf, host_repo_path: Option, repo_origin_url: Option, + artifact_storage: Option, provenance: Option, } @@ -83,6 +85,7 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result Result, run_store: RunStoreHandle, event_sink: RunEventSink, + artifact_uploader: Option>, git: Option, github_app: Option, worktree_mode: Option, @@ -72,6 +74,7 @@ pub struct StartServices { pub interviewer: Arc, pub run_store: RunStoreHandle, pub event_sink: RunEventSink, + pub artifact_uploader: Option>, pub run_control: Option>, pub github_app: Option, pub on_node: crate::OnNodeCallback, @@ -393,6 +396,7 @@ impl RunSession { devcontainer, seed_context: None, run_store: services.run_store, + artifact_uploader: services.artifact_uploader, git, github_app: services.github_app.clone(), worktree_mode: Some(resolve_worktree_mode(&settings)), @@ -523,6 +527,7 @@ impl RunSession { git: self.git, worktree_mode: self.worktree_mode, registry_override: self.registry_override, + artifact_uploader: self.artifact_uploader, run_control: self.run_control, checkpoint, seed_context: self.seed_context, @@ -869,6 +874,7 @@ mod tests { interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), run_store: store.open_run(&fixtures::RUN_1).await.unwrap().into(), event_sink: RunEventSink::store(store.open_run(&fixtures::RUN_1).await.unwrap()), + artifact_uploader: None, run_control: None, github_app: None, on_node: None, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index 1c785c22f..e49a526e8 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -47,6 +47,7 @@ pub async fn execute(init: Initialized) -> Executed { sandbox, registry, on_node, + artifact_uploader, run_control, hook_runner, env, @@ -100,6 +101,7 @@ pub async fn execute(init: Initialized) -> Executed { graph_arc, &run_options.run_dir, &run_store, + artifact_uploader, &settings_arc, checkpoint.is_some(), on_node, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index fd2cf9e72..16925eef5 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -219,6 +219,7 @@ async fn execute_test_run_with_options( worktree_mode: None, run_control: None, registry_override, + artifact_uploader: None, checkpoint: None, seed_context: None, }, @@ -275,6 +276,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { worktree_mode: None, run_control: None, registry_override: None, + artifact_uploader: None, checkpoint: None, seed_context: None, }, @@ -341,6 +343,7 @@ async fn run_with_lifecycle( worktree_mode: None, run_control: None, registry_override: Some(Arc::new(registry)), + artifact_uploader: None, checkpoint: None, seed_context: None, }, diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index b3c999188..4af604ffb 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -651,6 +651,7 @@ pub async fn initialize( sandbox, registry, on_node: None, + artifact_uploader: options.artifact_uploader, run_control: options.run_control, hook_runner, env, @@ -804,6 +805,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, + artifact_uploader: None, checkpoint: None, seed_context: None, }, @@ -880,6 +882,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, + artifact_uploader: None, checkpoint: None, seed_context: None, }, diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 9a7c19a30..76ee2872c 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -133,6 +133,7 @@ mod tests { ("env".to_string(), "test".to_string()), ("team".to_string(), "workflow".to_string()), ]), + artifact_storage: None, provenance: None, } } @@ -157,6 +158,7 @@ mod tests { base_branch: record.base_branch.clone(), workflow_slug: record.workflow_slug.clone(), db_prefix: None, + artifact_storage: record.artifact_storage, provenance: record.provenance.clone(), }, ) diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 88b0d0834..eac9a82d5 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1090,6 +1090,7 @@ mod tests { repo_origin_url: None, base_branch: Some("main".to_string()), labels: HashMap::new(), + artifact_storage: None, provenance: None, }; append_event( @@ -1109,6 +1110,7 @@ mod tests { base_branch: run_record.base_branch.clone(), workflow_slug: run_record.workflow_slug.clone(), db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) @@ -1160,6 +1162,7 @@ mod tests { repo_origin_url: None, base_branch: Some("main".to_string()), labels: HashMap::new(), + artifact_storage: None, provenance: None, }; append_event( @@ -1179,6 +1182,7 @@ mod tests { base_branch: run_record.base_branch.clone(), workflow_slug: run_record.workflow_slug.clone(), db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) @@ -1383,6 +1387,7 @@ mod tests { repo_origin_url: None, base_branch: None, labels: std::collections::HashMap::new(), + artifact_storage: None, provenance: None, }; append_event( @@ -1402,6 +1407,7 @@ mod tests { base_branch: None, workflow_slug: None, db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 83886226b..bf1648260 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -241,6 +241,7 @@ mod tests { repo_origin_url: None, base_branch: None, labels: std::collections::HashMap::new(), + artifact_storage: None, provenance: None, }; append_event( @@ -260,6 +261,7 @@ mod tests { base_branch: None, workflow_slug: None, db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index 00e58cd97..b943e711d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -14,6 +14,7 @@ use fabro_sandbox::SandboxSpec; use fabro_types::RunId; use fabro_validate::Diagnostic; +use crate::artifact_upload::StageArtifactUploader; use crate::context::Context; use crate::error::FabroError; use crate::event::Emitter; @@ -246,6 +247,7 @@ pub struct InitOptions { pub git: Option, pub worktree_mode: Option, pub registry_override: Option>, + pub artifact_uploader: Option>, pub run_control: Option>, pub checkpoint: Option, pub seed_context: Option, @@ -266,6 +268,7 @@ pub struct Initialized { pub sandbox: Arc, pub registry: Arc, pub on_node: crate::OnNodeCallback, + pub artifact_uploader: Option>, pub run_control: Option>, pub hook_runner: Option>, pub env: HashMap, diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index db087938b..8f8e473fb 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -427,6 +427,7 @@ mod tests { repo_origin_url: None, base_branch: Some("main".to_string()), labels: HashMap::new(), + artifact_storage: None, provenance: None, } } @@ -458,6 +459,7 @@ mod tests { base_branch: run_record.base_branch.clone(), workflow_slug: run_record.workflow_slug.clone(), db_prefix: None, + artifact_storage: run_record.artifact_storage, provenance: run_record.provenance.clone(), }, ) diff --git a/lib/crates/fabro-workflow/src/runtime_store.rs b/lib/crates/fabro-workflow/src/runtime_store.rs index 3eec4143a..5ce71b450 100644 --- a/lib/crates/fabro-workflow/src/runtime_store.rs +++ b/lib/crates/fabro-workflow/src/runtime_store.rs @@ -140,6 +140,7 @@ mod tests { repo_origin_url: None, base_branch: None, labels: HashMap::new(), + artifact_storage: None, provenance: None, } } @@ -165,6 +166,7 @@ mod tests { base_branch: None, workflow_slug: Some("test".to_string()), db_prefix: None, + artifact_storage: None, provenance: None, }, ) diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index c572d9af4..d69e6664b 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -91,6 +91,7 @@ async fn initialized( base_branch: run_options.base_branch.clone(), workflow_slug: run_options.workflow_slug.clone(), db_prefix: None, + artifact_storage: None, provenance: None, }, ) @@ -113,6 +114,7 @@ async fn initialized( sandbox, registry: Arc::new(registry), on_node: None, + artifact_uploader: None, run_control: None, hook_runner: options.hook_runner, env: options.env, diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 3faa59fbb..f894e9b11 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -22,6 +22,8 @@ models/api-question-option.ts models/api-question.ts models/api-settings.ts models/append-event-response.ts +models/artifact-batch-upload-entry.ts +models/artifact-batch-upload-manifest.ts models/artifact-entry.ts models/artifact-list-response.ts models/artifacts-settings.ts 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 c73ff530f..12509d35c 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 @@ -45,6 +45,8 @@ import type { RunProjection } from '../models'; import type { RunSettings } from '../models'; // @ts-ignore import type { WriteBlobResponse } from '../models'; +// @ts-ignore +import type { WriteRunBlobRequest } from '../models'; /** * RunInternalsApi - axios parameter creator */ @@ -479,22 +481,20 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config }; }, /** - * Uploads an artifact for a stage. Intended for trusted internal callers. + * Uploads one or more artifacts for a stage. Intended for trusted internal callers. The server accepts both: - `application/octet-stream` for single-file uploads with the `filename` query parameter - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` The generated Rust client currently exposes the octet-stream variant because the OpenAPI code generator in this repo does not support multiple request media types on one operation. * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. * @param {File} body + * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - putStageArtifact: async (id: string, stageId: string, filename: string, body: File, options: RawAxiosRequestConfig = {}): Promise => { + putStageArtifact: async (id: string, stageId: string, body: File, filename?: string, options: RawAxiosRequestConfig = {}): Promise => { // verify required parameter 'id' is not null or undefined assertParamExists('putStageArtifact', 'id', id) // verify required parameter 'stageId' is not null or undefined assertParamExists('putStageArtifact', 'stageId', stageId) - // verify required parameter 'filename' is not null or undefined - assertParamExists('putStageArtifact', 'filename', filename) // verify required parameter 'body' is not null or undefined assertParamExists('putStageArtifact', 'body', body) const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/artifacts` @@ -847,17 +847,17 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Uploads an artifact for a stage. Intended for trusted internal callers. + * Uploads one or more artifacts for a stage. Intended for trusted internal callers. The server accepts both: - `application/octet-stream` for single-file uploads with the `filename` query parameter - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` The generated Rust client currently exposes the octet-stream variant because the OpenAPI code generator in this repo does not support multiple request media types on one operation. * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. * @param {File} body + * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async putStageArtifact(id: string, stageId: string, filename: string, body: File, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { - const localVarAxiosArgs = await localVarAxiosParamCreator.putStageArtifact(id, stageId, filename, body, options); + async putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.putStageArtifact(id, stageId, body, filename, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.putStageArtifact']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); @@ -1028,17 +1028,17 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b return localVarFp.listStageTurns(id, stageId, pageLimit, pageOffset, options).then((request) => request(axios, basePath)); }, /** - * Uploads an artifact for a stage. Intended for trusted internal callers. + * Uploads one or more artifacts for a stage. Intended for trusted internal callers. The server accepts both: - `application/octet-stream` for single-file uploads with the `filename` query parameter - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` The generated Rust client currently exposes the octet-stream variant because the OpenAPI code generator in this repo does not support multiple request media types on one operation. * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. * @param {File} body + * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - putStageArtifact(id: string, stageId: string, filename: string, body: File, options?: RawAxiosRequestConfig): AxiosPromise { - return localVarFp.putStageArtifact(id, stageId, filename, body, options).then((request) => request(axios, basePath)); + putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.putStageArtifact(id, stageId, body, filename, options).then((request) => request(axios, basePath)); }, /** * Reads a previously stored blob by identifier. @@ -1201,17 +1201,17 @@ export class RunInternalsApi extends BaseAPI { } /** - * Uploads an artifact for a stage. Intended for trusted internal callers. + * Uploads one or more artifacts for a stage. Intended for trusted internal callers. The server accepts both: - `application/octet-stream` for single-file uploads with the `filename` query parameter - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` The generated Rust client currently exposes the octet-stream variant because the OpenAPI code generator in this repo does not support multiple request media types on one operation. * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. * @param {File} body + * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - public putStageArtifact(id: string, stageId: string, filename: string, body: File, options?: RawAxiosRequestConfig) { - return RunInternalsApiFp(this.configuration).putStageArtifact(id, stageId, filename, body, options).then((request) => request(this.axios, this.basePath)); + public putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig) { + return RunInternalsApiFp(this.configuration).putStageArtifact(id, stageId, body, filename, options).then((request) => request(this.axios, this.basePath)); } /** @@ -1260,4 +1260,3 @@ export class RunInternalsApi extends BaseAPI { return RunInternalsApiFp(this.configuration).writeRunBlob(id, body, options).then((request) => request(this.axios, this.basePath)); } } - diff --git a/lib/packages/fabro-api-client/src/models/artifact-batch-upload-entry.ts b/lib/packages/fabro-api-client/src/models/artifact-batch-upload-entry.ts new file mode 100644 index 000000000..8a42ae05a --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/artifact-batch-upload-entry.ts @@ -0,0 +1,41 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + + +/** + * One file entry in a strict multipart artifact upload manifest. + */ +export interface ArtifactBatchUploadEntry { + /** + * Multipart field name for the file part. + */ + 'part': string; + /** + * Relative artifact path to store. + */ + 'path': string; + /** + * Optional lowercase hex SHA-256 checksum for the file contents. + */ + 'sha256'?: string; + /** + * Optional exact byte length expected for the file part. + */ + 'expected_bytes'?: number; + /** + * Optional client-supplied content type for the file part. + */ + 'content_type'?: string; +} diff --git a/lib/packages/fabro-api-client/src/models/artifact-batch-upload-manifest.ts b/lib/packages/fabro-api-client/src/models/artifact-batch-upload-manifest.ts new file mode 100644 index 000000000..ad483a824 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/artifact-batch-upload-manifest.ts @@ -0,0 +1,25 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + +// May contain unused imports in some cases +// @ts-ignore +import type { ArtifactBatchUploadEntry } from './artifact-batch-upload-entry'; + +/** + * Manifest for strict multipart artifact uploads. + */ +export interface ArtifactBatchUploadManifest { + 'entries': Array; +} diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 30cc31366..aecd845cf 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -4,6 +4,8 @@ export * from './api-question'; export * from './api-question-option'; export * from './api-settings'; export * from './append-event-response'; +export * from './artifact-batch-upload-entry'; +export * from './artifact-batch-upload-manifest'; export * from './artifact-entry'; export * from './artifact-list-response'; export * from './artifacts-settings';