//! The storage trait every backend implements, and the capped read every //! caller of it goes through. use super::bucket::S3DeleteAuthority; use super::key::S3Key; use crate::error::{AppError, Result}; /// Aggregate a `ByteStream` into memory, aborting once more than `max_bytes` /// have been read. Backs [`StorageBackend::download_object_buf_capped`]; factored /// out as a free function so the cap logic is unit-testable without a full /// backend. `label` is only used in the error message (the object key). pub(crate) async fn read_bytestream_capped( mut stream: s3_storage::ByteStream, label: &str, max_bytes: u64, ) -> Result { let mut buf = bytes::BytesMut::new(); let mut read: u64 = 0; loop { match stream.try_next().await { Ok(Some(chunk)) => { read += chunk.len() as u64; if read > max_bytes { return Err(AppError::Storage(format!( "object {label} exceeds scan in-memory cap ({read} > {max_bytes} bytes); \ recorded size under-reported the real object" ))); } buf.extend_from_slice(&chunk); } Ok(None) => break, Err(e) => return Err(AppError::Storage(format!("read object from S3: {e}"))), } } Ok(buf.freeze()) } /// Abstract storage backend, implemented by `S3Client` (production) and /// `InMemoryStorage` (tests). Routes access storage through this trait. #[async_trait::async_trait] pub trait StorageBackend: Send + Sync { /// Generate a presigned upload URL. `max_bytes`, when set, is signed into /// the URL as `Content-Length` so S3 itself enforces the size cap at the /// protocol level (prevents oversized PUTs from burning bandwidth before /// hitting the post-PUT delete-and-charge fallback). async fn presign_upload( &self, s3_key: &S3Key, content_type: &str, expiry_secs: Option, cache_control: Option<&str>, max_bytes: Option, ) -> Result; async fn presign_download(&self, s3_key: &S3Key, expiry_secs: Option) -> Result; async fn object_exists(&self, s3_key: &str) -> Result; async fn object_size(&self, s3_key: &str) -> Result>; async fn download_object(&self, s3_key: &str) -> Result>; /// Download as `bytes::Bytes`, no `to_vec` copy of the aggregated body. /// Memory-sensitive callers (the scanner's buffered branch) use this so the /// payload isn't transiently doubled. async fn download_object_buf(&self, s3_key: &str) -> Result; /// Stream the object body without buffering the whole payload. Callers /// drive the stream to disk (scanner spool) or to a layer that consumes /// chunks directly. async fn download_stream(&self, s3_key: &str) -> Result; /// Download into memory like [`download_object_buf`], but abort if the body /// exceeds `max_bytes`. The scanner routes files it *recorded* as small to an /// in-memory branch, but `file_size_bytes` is asserted at upload time and can /// under-report the real object; this bounds the aggregation so a mis-recorded /// or abusive object can't pull an unbounded body into RAM, the independent /// ceiling the spool path already enforces. Streams via /// `download_stream`, so no backend can hand back the whole body up front. async fn download_object_buf_capped( &self, s3_key: &str, max_bytes: u64, ) -> Result { let stream = self.download_stream(s3_key).await?; read_bytestream_capped(stream, s3_key, max_bytes).await } /// Read up to the first `len` bytes of an object. Production overrides this /// with a ranged `GetObject` so a content sniff transfers only the header, /// not the whole object. The default streams and stops early, correct, but /// it still initiates a full GET, which is fine for in-memory test backends. async fn download_head(&self, s3_key: &str, len: usize) -> Result> { let mut stream = self.download_stream(s3_key).await?; let mut head = Vec::with_capacity(len.min(64 * 1024)); while head.len() < len { match stream.try_next().await { Ok(Some(chunk)) => head.extend_from_slice(&chunk), Ok(None) => break, Err(e) => return Err(AppError::Storage(format!("read object head from S3: {e}"))), } } head.truncate(len); Ok(head) } async fn upload_object( &self, s3_key: &S3Key, content_type: &str, data: Vec, cache_control: Option<&str>, ) -> Result<()>; /// Delete an object. Requires an [`S3DeleteAuthority`], route handlers /// cannot mint one, so they must enqueue through `pending_s3_deletions` /// instead of deleting directly. async fn delete_object(&self, auth: &S3DeleteAuthority, s3_key: &S3Key) -> Result<()>; /// Delete a batch of objects in a single S3 `DeleteObjects` request /// (up to 1000 keys/call). Default loops `delete_object` so test backends /// don't have to implement it, but production should override. async fn delete_objects(&self, auth: &S3DeleteAuthority, keys: &[S3Key]) -> Result<()> { let mut failed = 0usize; for k in keys { if let Err(e) = self.delete_object(auth, k).await { failed += 1; tracing::warn!(key = %k, error = ?e, "delete_objects: per-key delete failed"); } } // Don't report success when every key failed, a total failure must // surface so the caller can fall back (Run #2 Storage MINOR). Partial // failures stay logged; callers pre-enqueue to pending_s3_deletions. if !keys.is_empty() && failed == keys.len() { return Err(AppError::Storage(format!( "delete_objects: all {failed} keys failed" ))); } Ok(()) } /// Delete all objects under a key prefix. Default logs a warning (no-op). async fn delete_prefix(&self, _auth: &S3DeleteAuthority, _prefix: &str) -> Result<()> { tracing::warn!("delete_prefix called on a storage backend that does not implement it"); Ok(()) } /// Upload a file via S3 multipart upload. Required (not defaulted): a /// default that `tokio::fs::read`s the whole file into RAM + single PUT /// silently defeats streaming, so a future backend that forgot to override /// it would quietly lose multipart. Every backend must declare its strategy. async fn upload_multipart( &self, s3_key: &S3Key, content_type: &str, file_path: &std::path::Path, ) -> Result<()>; /// Server-side copy `src_key` to `dst_key` within this backend's bucket /// (no bytes transit the process). The scan-then-promote primitive: a Clean /// staging object is copied to the served key the client holds no presign /// for, so served bytes are provably the scanned bytes. Required (not /// defaulted): a silent no-op default would make a /// promote "succeed" while the served key stays empty. async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()>; /// Server-side copy from `src_bucket` into THIS backend's bucket. The /// cross-bucket half of scan-then-promote: a Clean staging object in the /// private bucket is lifted into the public (CDN-served) bucket. Call on the /// public backend with the private bucket name as `src_bucket`. Required /// (not defaulted) for the same reason as `copy_object`. async fn copy_object_from( &self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, ) -> Result<()>; /// Server-side multipart copy (`UploadPartCopy`) for sources over the 5 GiB /// single-part `CopyObject` limit, the >5 GiB half of scan-then-promote. /// Always takes `src_bucket` explicitly, collapsing the /// `copy_object`/`copy_object_from` pair into one method (pass this /// backend's own bucket for a same-bucket promote). `content_type` sets the /// destination's type, since a fresh multipart upload does not inherit the /// source's metadata the way `CopyObject` does. Required (not defaulted) for /// the same reason as `copy_object`. async fn copy_object_multipart( &self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, content_type: &str, src_size: u64, part_size: Option, ) -> Result<()>; // Client-direct multipart sessions // // The counterpart to `upload_multipart`, which drives a whole transfer // server-side from a local file. Here the server only mints the session and // the per-part presigned URLs; the client streams parts straight to S3, so // no object bytes transit the server. This is the path large CLI/desktop // uploads take (a browser stays on the single-PUT `presign_upload`). // // All four are required (not defaulted): a no-op default would mint a // session no client could complete, or silently drop the cleanup that // stops orphaned parts billing forever. /// Begin a client-direct multipart upload, returning the `upload_id` the /// part/complete/abort calls key on. async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result; /// Presign an `UploadPart` request for one part (1-based `part_number`). /// `max_bytes`, when set, is signed as `Content-Length`, the same /// defense-in-depth as [`Self::presign_upload`], the authoritative size /// check still happens at confirm time. /// /// `checksum_sha256` (base64 of the raw digest), when set, is signed as /// `x-amz-checksum-sha256` and IS enforced: S3 rehashes the part and /// rejects a mismatch before the bytes are durable. async fn presign_upload_part( &self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option, max_bytes: Option, checksum_sha256: Option<&str>, ) -> Result; /// Complete a multipart upload from the collected `(part_number, etag)` /// pairs. Parts may be passed in any order; the backend sorts them. async fn complete_multipart_upload( &self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)], ) -> Result<()>; /// Abort a multipart upload, releasing its uploaded parts. The /// pending-upload reaper calls this on sessions that were never confirmed, /// incomplete multipart uploads bill for their parts indefinitely. async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()>; /// Upload ids of the in-progress multipart sessions for exactly `s3_key`. /// The reaper recovers them from S3 rather than the database, so a session /// whose tracking row was lost is still cleaned up. Required (not defaulted): /// an empty default would silently strand billed parts. async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result>; async fn check_connectivity(&self) -> std::result::Result<(), String>; fn bucket(&self) -> &str; }