feat: file upload to object storage and then async file verification#954
feat: file upload to object storage and then async file verification#954heldtogether wants to merge 9 commits into
Conversation
heldtogether
changed the title
| [dependencies] | ||
| axum = { version = "0.8", features = ["multipart"] } | ||
| fusillade = { version = "14.2.1" } | ||
| fusillade = { version = "14.2.1", path = "../../fusillade" } |
There was a problem hiding this comment.
I assume this needs bumping when https://github.com/doublewordai/fusillade/pull/207/changes is merged in.
| // If asynchronous file ingestion is enabled, gate batch creation on ingest state. | ||
| if let Some(status) = crate::api::handlers::files::get_file_ingest_status(state.db.write(), file_id).await? { | ||
| match status.as_str() { | ||
| "pending" | "processing" => { |
There was a problem hiding this comment.
Consider waiting for async processing to complete if we don't want to hard fail at this point
heldtogether
changed the title
There was a problem hiding this comment.
Pull request overview
This PR adds an optional S3-compatible object storage backend for batch file uploads (async ingest/verification) and introduces an object-store-backed cache for completed batch/file result downloads to reduce Postgres load and decouple upload latency from DB throughput.
Changes:
- Add object storage client + config, and a new
file_ingest_jobstable to track async ingestion state. - Split file upload flow: store raw JSONL in object storage, create a placeholder file record, enqueue an ingest job, and gate batch creation on ingest completion.
- Add completed-result pull-through caching to object storage for batch results and output/error file content, with invalidation on deletions.
Reviewed changes
Copilot reviewed 15 out of 16 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| dwctl/src/test/utils.rs | Updates test app state to include new task state fields (db_pool, config). |
| dwctl/src/test/mod.rs | Updates task state construction in tests for new background worker requirements. |
| dwctl/src/tasks.rs | Extends task runner to register and run the new ingest-file worker. |
| dwctl/src/lib.rs | Wires new modules and passes config/db pool into background task state. |
| dwctl/src/config.rs | Adds object store config + backend enum + validation + tests. |
| dwctl/src/blob_storage.rs | Implements S3-compatible blob client (put/get/delete/prefix delete). |
| dwctl/src/batch_result_cache.rs | Adds deterministic cache keys, JSONL serialization, slice/pagination helpers, and cache read/build/invalidate logic. |
| dwctl/src/api/models/files.rs | Extends file API response shape with status and status_details. |
| dwctl/src/api/handlers/files.rs | Adds async ingest pipeline, object-store upload path, caching for completed output/error downloads, and cache invalidation on delete. |
| dwctl/src/api/handlers/batches.rs | Gates batch creation on ingest state; adds caching for completed batch results; invalidates cache on batch delete. |
| dwctl/migrations/090_add_file_ingest_jobs.sql | Creates file_ingest_jobs table and status index. |
| dwctl/Cargo.toml | Adds AWS SDK deps; changes fusillade dependency configuration. |
| dashboard/src/components/features/batches/FilesTable/columns.tsx | Adds UI disable/tooltip behavior based on file status. |
| dashboard/src/api/control-layer/types.ts | Updates dashboard FileObject type with status and status_details. |
| config.yaml | Documents new storage backend and object store config options. |
| Cargo.lock | Locks new AWS SDK dependency graph updates. |
| pub struct IngestFileInput { | ||
| pub file_id: Uuid, | ||
| pub object_key: String, | ||
| pub endpoint: String, | ||
| pub api_key: String, | ||
| pub api_key_id: Uuid, | ||
| pub uploaded_by: String, | ||
| pub filename: String, | ||
| pub size_bytes: i64, | ||
| } |
There was a problem hiding this comment.
IngestFileInput includes the raw api_key, which will be persisted as part of the underway job payload. That stores a sensitive credential at rest in the job queue tables. Prefer passing only api_key_id (and any other identifiers) and fetching/decrypting the API key inside the worker step from the database, or using a short-lived token mechanism, so secrets aren’t serialized into job inputs.
| let bytes = client | ||
| .get_file_bytes(&input.object_key) | ||
| .await | ||
| .map_err(|e| TaskError::Retryable(format!("download blob for ingest: {e}")))?; | ||
|
|
||
| let content = std::str::from_utf8(&bytes).map_err(|e| TaskError::Fatal(format!("Invalid UTF-8 in upload: {e}")))?; | ||
|
|
||
| let mut conn = db_pool |
There was a problem hiding this comment.
The ingest worker downloads the entire uploaded blob into memory (get_file_bytes -> Vec<u8>), then converts to UTF-8 and iterates lines. This can cause large memory spikes/OOM for big files (especially if limits.files.max_file_size=0 allows unlimited uploads). Consider streaming the S3 body (process line-by-line from a reader) or enforcing a non-zero max file size when storage_backend=object_store to keep memory bounded.
| let file_id = Uuid::new_v4(); | ||
| let tmp_path = format!("/tmp/dwctl-upload-{}.jsonl", file_id); | ||
| let mut tmp_file = tokio::fs::File::create(&tmp_path).await.map_err(|e| Error::Internal { | ||
| operation: format!("create temp upload file: {e}"), | ||
| })?; | ||
| let mut saw_file = false; |
There was a problem hiding this comment.
The object-store upload path writes the multipart file to a temp path under /tmp but only removes it after a successful put_file_from_path. Any early return (invalid purpose, payload too large, multipart parse errors, S3 upload error, etc.) will leak the temp file and can fill disk over time. Use a tempfile that cleans up on drop and/or ensure remove_file runs in a drop guard / scopeguard on all exit paths.
| } | ||
|
|
||
| // If asynchronous file ingestion is enabled, gate batch creation on ingest state. | ||
| if let Some(status) = crate::api::handlers::files::get_file_ingest_status(state.db.write(), file_id).await? { |
There was a problem hiding this comment.
This ingest-status gating read uses state.db.write() even though it’s a read-only query. That unnecessarily consumes primary connections (and bypasses replicas if configured). Use state.db.read() here for consistency with the rest of this file (e.g., other read paths use state.db.read().acquire() / state.db.read()).
| if let Some(status) = crate::api::handlers::files::get_file_ingest_status(state.db.write(), file_id).await? { | |
| if let Some(status) = crate::api::handlers::files::get_file_ingest_status(state.db.read(), file_id).await? { |
| } | ||
|
|
||
| let config = state.current_config(); | ||
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), file_id).await?; |
There was a problem hiding this comment.
Cache invalidation is performed before deleting the file, and any object-store error will currently fail the entire delete operation. Since this is a best-effort derived cache, consider making invalidation non-fatal (log and continue) so users can still delete files when the object store is temporarily unavailable.
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), file_id).await?; | |
| if let Err(e) = batch_result_cache::invalidate_cached_file_results(config.as_ref(), file_id).await { | |
| tracing::warn!( | |
| file_id = %file_id, | |
| error = %e, | |
| "failed to invalidate cached batch results before file deletion; continuing with delete" | |
| ); | |
| } |
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), *output_file_id).await?; | ||
| } | ||
| if let Some(error_file_id) = batch.error_file_id { | ||
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), *error_file_id).await?; |
There was a problem hiding this comment.
Same as file deletion: cache invalidation errors will currently abort batch deletion. Since cached results are derivative, it’s usually better to treat invalidation as best-effort (log failures, continue deleting) to avoid making deletion dependent on object-store availability.
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), *output_file_id).await?; | |
| } | |
| if let Some(error_file_id) = batch.error_file_id { | |
| batch_result_cache::invalidate_cached_file_results(config.as_ref(), *error_file_id).await?; | |
| if let Err(err) = batch_result_cache::invalidate_cached_file_results(config.as_ref(), *output_file_id).await { | |
| tracing::warn!( | |
| batch_id = %batch_id, | |
| file_id = %output_file_id, | |
| error = %err, | |
| "Failed to invalidate cached batch output file results during batch deletion" | |
| ); | |
| } | |
| } | |
| if let Some(error_file_id) = batch.error_file_id { | |
| if let Err(err) = batch_result_cache::invalidate_cached_file_results(config.as_ref(), *error_file_id).await { | |
| tracing::warn!( | |
| batch_id = %batch_id, | |
| file_id = %error_file_id, | |
| error = %err, | |
| "Failed to invalidate cached batch error file results during batch deletion" | |
| ); | |
| } |
| [dependencies] | ||
| axum = { version = "0.8", features = ["multipart"] } | ||
| fusillade = { version = "14.2.1" } | ||
| fusillade = { version = "14.2.1", path = "../../fusillade" } |
There was a problem hiding this comment.
fusillade is switched to a path dependency at ../../fusillade, which is outside this repository/workspace. That will break reproducible builds/CI unless the environment happens to have that directory present. Prefer using the crates.io version, adding fusillade as a workspace member within this repo, or using a git dependency pinned to a commit/tag.
| fusillade = { version = "14.2.1", path = "../../fusillade" } | |
| fusillade = "14.2.1" |
| async fn collect_stream_bytes<T, S, F>(mut stream: S, mut serialize: F) -> Result<Vec<u8>> | ||
| where | ||
| S: futures::Stream<Item = fusillade::Result<T>> + Unpin, | ||
| F: FnMut(T) -> Result<Vec<u8>>, | ||
| { | ||
| let mut bytes = Vec::new(); | ||
|
|
||
| while let Some(item) = stream.next().await { | ||
| let item = item.map_err(|e| Error::Internal { | ||
| operation: format!("stream batch result data: {e}"), | ||
| })?; | ||
| bytes.extend(serialize(item)?); | ||
| } | ||
|
|
||
| Ok(bytes) | ||
| } |
There was a problem hiding this comment.
The cache build path collects the entire result stream into a single Vec<u8> before responding and before writing to object storage. For large completed batches/output files this can create significant memory spikes and defeats the previous streaming behavior on the first request. Consider streaming directly into the object-store writer (multipart upload) while also streaming the response, or imposing a maximum cacheable payload size/falling back to streaming for large result sets.
| const viewDisabled = file.status === "error"; | ||
| const disabledReason = file.status_details || "This file cannot be viewed."; |
There was a problem hiding this comment.
The UI only disables View/Download actions when file.status === "error", but in the new async-ingest flow files can be in "uploaded" (pending/processing). In that state the backend won’t have templates/results available yet, so these actions will likely show empty content or confusing errors. Consider also disabling (or showing a distinct “processing” tooltip/state) when status !== "processed".
| const viewDisabled = file.status === "error"; | |
| const disabledReason = file.status_details || "This file cannot be viewed."; | |
| const viewDisabled = file.status !== "processed"; | |
| const disabledReason = | |
| file.status === "error" | |
| ? file.status_details || "This file cannot be viewed." | |
| : "This file is still processing and cannot be viewed yet."; |
Summary
This PR introduces object storage as an optional backend for batch file uploads, decoupling upload latency from Postgres connection throughput.
When batches.files.storage_backend=object_store:
It also adds an S3-backed pull-through cache for immutable batch download results.
Completed batch downloads now build JSONL from Postgres on the first request, write the result to object storage, and serve subsequent identical requests from cache. In-progress batches still read directly from Postgres.
When storage_backend=postgres, existing behavior remains unchanged.
Problem
Previously, both upload and download paths were DB-bound:
What Changed
New object storage client
New ingest tracking table
Upload handler flow split
Download handler
Background task integration
Batch creation gating
Config support
Key Design Decisions
Tradeoffs
Failure Modes and Handling