//! Shared S3-compatible storage client. //! //! Overview: maintainer wiki. //! //! //! Thin wrapper around the AWS SDK providing upload, download, delete, //! presigned URL generation, and bucket management. Used by MNW and //! Multithreaded to avoid duplicating S3 initialization and operations. #![warn(missing_docs)] use aws_config::BehaviorVersion; use aws_sdk_s3::Client; use aws_sdk_s3::config::{Credentials, Region}; use aws_sdk_s3::presigning::PresigningConfig; use aws_sdk_s3::types::{ CompletedMultipartUpload, CompletedPart, CorsConfiguration, CorsRule, Delete, ObjectIdentifier, }; use std::time::Duration; pub use aws_sdk_s3::primitives::ByteStream; /// The HTTPS client every S3 client is built on. /// /// Built here rather than taken from the SDK's `default-https-client` feature, /// which is a hard alias for the aws-lc-rs (C) crypto backend. Ring is the /// pure-Rust provider the rest of the tree uses, and the SDK will accept it only /// through an explicitly constructed client. fn https_client() -> aws_sdk_s3::config::SharedHttpClient { aws_smithy_http_client::Builder::new() .tls_provider(aws_smithy_http_client::tls::Provider::Rustls( aws_smithy_http_client::tls::rustls_provider::CryptoMode::Ring, )) .build_https() } /// S3 connection configuration. #[derive(Debug, Clone)] pub struct S3Config { /// Endpoint URL (e.g., `https://fsn1.your-objectstorage.com`) pub endpoint: String, /// Bucket name pub bucket: String, /// Access key ID pub access_key: String, /// Secret access key pub secret_key: String, /// Region (e.g., `fsn1`) pub region: String, } /// S3 client wrapper. #[derive(Clone)] pub struct S3Client { client: Client, bucket: String, } /// SigV4's hard maximum presign lifetime (7 days). A presign request above this /// is rejected by the signer, so we clamp callers to it rather than surface an /// opaque signing error — and it bounds how long any single minted URL can live. const MAX_PRESIGN_EXPIRY_SECS: u64 = 7 * 24 * 60 * 60; /// S3 multipart-upload limits, straight from the S3 API contract. Every layer /// that plans a client-direct multipart upload (the blob and creator-media /// session endpoints) validates against these, so the client is never handed a /// plan S3 would reject at complete time. /// /// Minimum size of every part *except the last* (5 MiB). The final part may be /// anything down to 1 byte. pub const MULTIPART_MIN_PART_SIZE: usize = 5 * 1024 * 1024; /// Maximum number of parts in a single multipart upload. pub const MULTIPART_MAX_PARTS: u32 = 10_000; /// Maximum size of a single part (5 GiB). pub const MULTIPART_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024; /// Maximum size of an object assembled from a multipart upload (5 TiB). pub const MULTIPART_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024; /// Default/floor part size for an auto-planned multipart upload (16 MiB): big /// enough to keep the per-part round-trip overhead low, small enough that a /// resumable client re-sends little on a retry. Used when the object is small /// enough not to force larger parts to stay within the part-count limit. pub const MULTIPART_DEFAULT_PART_SIZE: usize = 16 * 1024 * 1024; // The part budget at max part size must be able to cover the object ceiling, or // a large-but-legal object would be unplannable at any part size. 10,000 x 5 GiB // ~= 48.8 TiB, comfortably over the 5 TiB object cap. (The 5 MiB *minimum* part // deliberately does NOT cover 5 TiB — a large object simply needs bigger parts, // which `MultipartPlan::new` enforces via the part-count limit.) const _: () = assert!( MULTIPART_MAX_PARTS as u128 * MULTIPART_MAX_PART_SIZE as u128 >= MULTIPART_MAX_OBJECT_SIZE as u128 ); /// A validated multipart-upload plan: the part size to use and how many parts a /// body of `total_size` bytes splits into. Pure arithmetic with no S3 call, so /// the blob and creator-media session endpoints share one source of truth for /// part geometry (and the client computes the same boundaries independently). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct MultipartPlan { /// Total object size in bytes. pub total_size: u64, /// Size of every part except the last. pub part_size: usize, /// Number of parts (1..=[`MULTIPART_MAX_PARTS`]). pub part_count: u32, } impl MultipartPlan { /// Plan a multipart upload of `total_size` bytes into `part_size`-byte parts, /// the last part taking the remainder. /// /// Errors if the object is empty (use a single PUT), exceeds the 5 TiB /// multipart ceiling, the part size is below S3's 5 MiB floor or above its /// 5 GiB ceiling, or the resulting part count would exceed 10,000 — the last /// meaning the part size is too small for this object and the caller should /// pick a larger one. pub fn new(total_size: u64, part_size: usize) -> Result { if total_size == 0 { return Err( "multipart upload needs a non-empty object; use a single PUT for empty objects" .to_string(), ); } if total_size > MULTIPART_MAX_OBJECT_SIZE { return Err(format!( "object is {total_size} bytes, over the {MULTIPART_MAX_OBJECT_SIZE}-byte (5 TiB) multipart ceiling" )); } if part_size < MULTIPART_MIN_PART_SIZE { return Err(format!( "part size {part_size} is below the {MULTIPART_MIN_PART_SIZE}-byte (5 MiB) S3 minimum" )); } if part_size as u64 > MULTIPART_MAX_PART_SIZE { return Err(format!( "part size {part_size} is above the {MULTIPART_MAX_PART_SIZE}-byte (5 GiB) S3 maximum" )); } let part_count = total_size.div_ceil(part_size as u64); if part_count > MULTIPART_MAX_PARTS as u64 { return Err(format!( "object of {total_size} bytes needs {part_count} parts at part size {part_size}, over the {MULTIPART_MAX_PARTS}-part limit; use a larger part size" )); } Ok(Self { total_size, part_size, part_count: part_count as u32, }) } /// Plan a multipart upload of `total_size` bytes, choosing the part size /// automatically: [`MULTIPART_DEFAULT_PART_SIZE`] when that keeps the object /// within [`MULTIPART_MAX_PARTS`] parts, otherwise the smallest whole-MiB /// part size that does. Errors only if the object is empty or over the 5 TiB /// ceiling — a valid non-empty object always yields a plan. pub fn auto(total_size: u64) -> Result { const MIB: u64 = 1024 * 1024; // Smallest part size that fits the object within the part-count cap, // rounded up to a whole MiB, then floored at the default part size. let needed = total_size.div_ceil(MULTIPART_MAX_PARTS as u64); let rounded = needed.div_ceil(MIB) * MIB; let part_size = (rounded as usize).max(MULTIPART_DEFAULT_PART_SIZE); Self::new(total_size, part_size) } /// Byte length of part `part_number` (1-based). The last part is the /// remainder; every earlier part is exactly `part_size`. Returns 0 for a /// part number outside `1..=part_count`. pub fn part_len(&self, part_number: u32) -> u64 { if part_number == 0 || part_number > self.part_count { return 0; } if part_number < self.part_count { return self.part_size as u64; } // Last part: the remainder, or a full part when the size divides evenly. match self.total_size % self.part_size as u64 { 0 => self.part_size as u64, rem => rem, } } /// Inclusive `[start, end]` byte range of part `part_number` (1-based), the /// form S3 `UploadPartCopy` wants as `bytes=start-end`. `None` for a part /// number outside `1..=part_count`. pub fn part_range(&self, part_number: u32) -> Option<(u64, u64)> { if part_number == 0 || part_number > self.part_count { return None; } let start = (part_number as u64 - 1) * self.part_size as u64; Some((start, start + self.part_len(part_number) - 1)) } } impl S3Client { /// Create a new S3 client from configuration. // Public async constructor: kept async for API stability across callers. #[allow(clippy::unused_async)] pub async fn new(config: &S3Config) -> Result { let credentials = Credentials::new( &config.access_key, &config.secret_key, None, None, "s3-storage", ); // Bound every S3 operation so a hung endpoint (e.g. a stalled // HeadObject on a blob-confirm path) can't wedge a caller forever. These // apply to establishing the connection and to a single attempt's // round-trip to first byte of the response — they do NOT cap the time // spent streaming a large object body, so big uploads/downloads are // unaffected. The SDK's default retry policy still applies per attempt. let timeout_config = aws_sdk_s3::config::timeout::TimeoutConfig::builder() .connect_timeout(Duration::from_secs(10)) .operation_attempt_timeout(Duration::from_mins(1)) .build(); let s3_config = aws_sdk_s3::Config::builder() .behavior_version(BehaviorVersion::latest()) .http_client(https_client()) .region(Region::new(config.region.clone())) .endpoint_url(&config.endpoint) .credentials_provider(credentials) .timeout_config(timeout_config) .force_path_style(true) .build(); let client = Client::from_conf(s3_config); Ok(Self { client, bucket: config.bucket.clone(), }) } /// Bucket name accessor. pub fn bucket(&self) -> &str { &self.bucket } /// Upload bytes to S3. pub async fn upload( &self, key: &str, content_type: &str, data: Vec, cache_control: Option<&str>, ) -> Result<(), String> { let mut req = self .client .put_object() .bucket(&self.bucket) .key(key) .content_type(content_type) .body(data.into()); if let Some(cc) = cache_control { req = req.cache_control(cc); } req.send() .await .map_err(|e| format!("S3 upload failed: {e}"))?; Ok(()) } /// Download bytes from S3. Returns `(data, content_type)`. /// /// Convenience `Vec` form; for the zero-extra-copy path use /// [`download_buf`](Self::download_buf), which returns the aggregated /// `Bytes` directly. pub async fn download(&self, key: &str) -> Result<(Vec, String), String> { let (bytes, content_type) = self.download_buf(key).await?; Ok((bytes.to_vec(), content_type)) } /// Download an object as `bytes::Bytes`, returning `(data, content_type)`. /// /// Unlike [`download`](Self::download) this does not copy the aggregated /// body into a fresh `Vec` — the caller gets the SDK's buffer directly. Use /// it on memory-sensitive paths (e.g. the scanner's buffered branch) where /// the extra `to_vec` would transiently double the footprint. pub async fn download_buf(&self, key: &str) -> Result<(bytes::Bytes, String), String> { let resp = self .client .get_object() .bucket(&self.bucket) .key(key) .send() .await .map_err(|e| format!("S3 download failed: {e}"))?; let content_type = resp .content_type() .unwrap_or("application/octet-stream") .to_string(); let bytes = resp .body .collect() .await .map_err(|e| format!("S3 read body failed: {e}"))?; Ok((bytes.into_bytes(), content_type)) } /// Stream an object's body from S3 without buffering. Caller drives the /// `ByteStream` to disk or hands it to a layer that wants chunks. pub async fn download_stream( &self, key: &str, ) -> Result { let resp = self .client .get_object() .bucket(&self.bucket) .key(key) .send() .await .map_err(|e| format!("S3 download failed: {e}"))?; Ok(resp.body) } /// Download only the first `len` bytes of an object via a ranged /// `GetObject` (`Range: bytes=0-{len-1}`). Used for content sniffing so a /// 4 KB header read doesn't initiate a transfer of the whole object. Returns /// fewer bytes if the object is smaller than `len`. pub async fn download_head(&self, key: &str, len: usize) -> Result, String> { if len == 0 { return Ok(Vec::new()); } let resp = self .client .get_object() .bucket(&self.bucket) .key(key) .range(format!("bytes=0-{}", len - 1)) .send() .await .map_err(|e| format!("S3 ranged download failed: {e}"))?; let data = resp .body .collect() .await .map_err(|e| format!("S3 ranged body read failed: {e}"))?; Ok(data.to_vec()) } /// Delete an object from S3. pub async fn delete(&self, key: &str) -> Result<(), String> { self.client .delete_object() .bucket(&self.bucket) .key(key) .send() .await .map_err(|e| format!("S3 delete failed: {e}"))?; Ok(()) } /// Server-side copy an object from `src_key` to `dst_key` within this /// bucket. No bytes transit the caller — S3 performs the copy internally. /// /// This is the primitive behind scan-then-promote: an upload is scanned at a /// staging key the client can overwrite, and only a Clean object is copied to /// the served key (which the client holds no presign for), so the served /// bytes are provably the scanned bytes (ultra-fuzz Run #24 Storage HIGH). /// /// `CopySource` is `{bucket}/{key}`. Keys in this system are sanitized to /// `[A-Za-z0-9._/-]` (see the server's `sanitize_filename`), none of which /// require percent-encoding, so the source is formed directly. pub async fn copy_object(&self, src_key: &str, dst_key: &str) -> Result<(), String> { self.copy_object_from(&self.bucket, src_key, dst_key).await } /// Server-side copy from an arbitrary source bucket into THIS client's /// bucket. Used by the scan-then-promote path to lift a Clean object from /// the private staging bucket into the public (CDN-served) bucket in one /// server-side operation (no bytes transit the process). The credentials /// this client holds must have read on `src_bucket`; in this deployment all /// buckets share one Hetzner project/key, so cross-bucket copy is permitted. /// /// `CopySource` is `{src_bucket}/{src_key}`. Keys are sanitized to /// `[A-Za-z0-9._/-]` (server `sanitize_filename`), so no percent-encoding is /// needed and the source is formed directly. When `src_bucket` equals this /// client's bucket the copy is the ordinary same-bucket promote. pub async fn copy_object_from( &self, src_bucket: &str, src_key: &str, dst_key: &str, ) -> Result<(), String> { self.client .copy_object() .bucket(&self.bucket) .copy_source(format!("{src_bucket}/{src_key}")) .key(dst_key) .send() .await .map_err(|e| { format!( "S3 copy_object {src_bucket}/{src_key} -> {}/{dst_key} failed: {e}", self.bucket ) })?; Ok(()) } /// Delete a batch of objects in a single S3 `DeleteObjects` request. /// /// S3 accepts up to 1000 keys per call; the caller is responsible for /// chunking. Returns the keys that failed (if any) along with their /// per-object error message. A successful response with `Errors` is /// not a hard error — partial success is normal for batched deletes. pub async fn delete_objects(&self, keys: &[String]) -> Result, String> { if keys.is_empty() { return Ok(Vec::new()); } // A key that won't build an ObjectIdentifier is reported as a failure // rather than silently dropped — otherwise the caller believes it was // deleted and the object leaks forever (ultra-fuzz Run 11 Storage MED). let mut failures: Vec<(String, String)> = Vec::new(); let mut objects: Vec = Vec::with_capacity(keys.len()); for k in keys { match ObjectIdentifier::builder().key(k).build() { Ok(o) => objects.push(o), Err(e) => failures.push((k.clone(), format!("malformed key: {e}"))), } } if objects.is_empty() { return Ok(failures); } let delete = Delete::builder() .set_objects(Some(objects)) .quiet(true) .build() .map_err(|e| format!("S3 delete_objects build failed: {e}"))?; let resp = self .client .delete_objects() .bucket(&self.bucket) .delete(delete) .send() .await .map_err(|e| format!("S3 delete_objects failed: {e}"))?; failures.extend( resp.errors .unwrap_or_default() .into_iter() .filter_map(|err| { let key = err.key?; let msg = err.message.unwrap_or_else(|| "".into()); Some((key, msg)) }), ); Ok(failures) } /// Delete all objects under a given key prefix. /// /// Lists in pages of 1000 and deletes each page in a single batched /// `DeleteObjects` call (S3's max). For a 50k-key prefix this is 50 /// round-trips instead of 50,000. pub async fn delete_prefix(&self, prefix: &str) -> Result<(), String> { let mut continuation_token: Option = None; loop { let mut req = self .client .list_objects_v2() .bucket(&self.bucket) .prefix(prefix) .max_keys(1000); if let Some(ref token) = continuation_token { req = req.continuation_token(token); } let resp = req .send() .await .map_err(|e| format!("S3 list objects failed: {e}"))?; let keys: Vec = resp .contents .unwrap_or_default() .into_iter() .filter_map(|obj| obj.key) .collect(); if !keys.is_empty() { let failures = self.delete_objects(&keys).await?; if !failures.is_empty() { // Surface per-key failures to the caller (and abort further // pages): the durable deletion queue is the convergence // backstop, but a silent partial wipe would hide a bucket // policy / permissions problem. Preview the first few keys. let preview: Vec = failures .iter() .take(5) .map(|(k, e)| format!("{k}: {e}")) .collect(); return Err(format!( "S3 delete_prefix partial failure: {} keys failed (first 5: {})", failures.len(), preview.join(", ") )); } } if resp.is_truncated.unwrap_or(false) { continuation_token = resp.next_continuation_token; } else { break; } } Ok(()) } /// Check if an object exists in S3. pub async fn object_exists(&self, key: &str) -> Result { match self .client .head_object() .bucket(&self.bucket) .key(key) .send() .await { Ok(_) => Ok(true), Err(e) => { let service_error = e.into_service_error(); if service_error.is_not_found() { Ok(false) } else { Err(format!("S3 head_object failed: {service_error}")) } } } } /// Get the size of an object in bytes, or `None` if not found. pub async fn object_size(&self, key: &str) -> Result, String> { match self .client .head_object() .bucket(&self.bucket) .key(key) .send() .await { Ok(resp) => Ok(resp.content_length()), Err(e) => { let service_error = e.into_service_error(); if service_error.is_not_found() { Ok(None) } else { Err(format!("S3 head_object failed: {service_error}")) } } } } /// Generate a presigned URL for uploading. /// /// When `max_bytes` is set, the value is signed into the request as /// `Content-Length` (a SignedHeader). A client that sends a *different* /// `Content-Length` than the one signed produces a signature mismatch, so /// the common "lie about the size" case fails. This is NOT a hard, /// server-enforced ceiling, though: it relies on the client sending an /// honest `Content-Length`, and S3-compatible backends (Ceph/MinIO/Garage) /// vary in how strictly they reconcile the declared length with the actual /// body. Treat it as defense-in-depth, not the boundary — the authoritative /// per-file cap is enforced at confirm time, where the handler reads the /// object's real size via [`Self::object_size`] before crediting storage. /// (A presigned POST with a `content-length-range` policy would be a true /// server-side range check; it is not used here because every upload path is /// a single PUT.) pub async fn presign_upload( &self, key: &str, content_type: &str, expiry_secs: u64, cache_control: Option<&str>, max_bytes: Option, ) -> Result { let presigning_config = PresigningConfig::builder() .expires_in(Duration::from_secs( expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS), )) .build() .map_err(|e| format!("Presigning config error: {e}"))?; let mut req = self .client .put_object() .bucket(&self.bucket) .key(key) .content_type(content_type); if let Some(cc) = cache_control { req = req.cache_control(cc); } if let Some(n) = max_bytes { req = req.content_length(n); } let presigned = req .presigned(presigning_config) .await .map_err(|e| format!("Failed to generate upload URL: {e}"))?; Ok(presigned.uri().to_string()) } /// Generate a presigned URL for downloading. pub async fn presign_download(&self, key: &str, expiry_secs: u64) -> Result { let presigning_config = PresigningConfig::builder() .expires_in(Duration::from_secs( expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS), )) .build() .map_err(|e| format!("Presigning config error: {e}"))?; let presigned = self .client .get_object() .bucket(&self.bucket) .key(key) .presigned(presigning_config) .await .map_err(|e| format!("Failed to generate download URL: {e}"))?; Ok(presigned.uri().to_string()) } /// Upload a file to S3 using multipart upload. /// /// Reads the file in `part_size` chunks (default 10 MB, minimum 5 MB) and /// uploads each as a part. Aborts the multipart upload on any failure. /// /// The abort lives in exactly ONE place: once the multipart upload has been /// created, all the fallible part/complete work runs in /// the private `run_multipart_upload`, and this wrapper aborts on any `Err` it /// returns. So a future failure path added inside the inner method aborts by /// construction — it cannot forget to (the Sto-S1 fix, made structural rather /// than per-branch). pub async fn upload_multipart( &self, key: &str, content_type: &str, file_path: &std::path::Path, part_size: Option, ) -> Result<(), String> { let part_size = part_size.unwrap_or(10 * 1024 * 1024); // 10 MB default if part_size < 5 * 1024 * 1024 { // Pre-flight: nothing created yet, nothing to abort. return Err("Multipart part size must be at least 5 MB".to_string()); } let upload_id = self.create_multipart_upload(key, content_type).await?; // Single abort site: any failure past this point aborts exactly once. match self .run_multipart_upload(key, file_path, part_size, &upload_id) .await { Ok(()) => Ok(()), Err(e) => { // Best-effort abort; the create succeeded, so on failure surface // the original error and just log if the cleanup also fails. if let Err(abort_err) = self.abort_multipart_upload(key, &upload_id).await { tracing::warn!("Failed to abort multipart upload for {key}: {abort_err}"); } Err(e) } } } /// Begin a client-direct multipart upload, returning the `upload_id` that /// [`Self::presign_upload_part`], [`Self::complete_multipart_upload`], and /// [`Self::abort_multipart_upload`] key on. /// /// The counterpart to [`Self::upload_multipart`], which drives the whole /// transfer server-side from a local file. Here the server only mints the id /// and (via `presign_upload_part`) the per-part URLs; the client streams the /// parts straight to S3, so no object bytes transit the server. pub async fn create_multipart_upload( &self, key: &str, content_type: &str, ) -> Result { let create = self .client .create_multipart_upload() .bucket(&self.bucket) .key(key) .content_type(content_type) .send() .await .map_err(|e| format!("S3 create multipart upload failed: {e}"))?; create .upload_id() .map(str::to_string) .ok_or_else(|| "S3 create multipart upload returned no upload_id".to_string()) } /// Presign an `UploadPart` request for one part of an in-progress multipart /// upload. `part_number` is 1-based (`1..=`[`MULTIPART_MAX_PARTS`]). /// /// When `max_bytes` is set it is signed as `Content-Length`, the same /// defense-in-depth (not a hard, server-enforced ceiling) as /// [`Self::presign_upload`] — see its docs for why the authoritative size /// check still lives at confirm time. /// /// `checksum_sha256` (base64 of the raw 32-byte digest) is signed as /// `x-amz-checksum-sha256`, and unlike the length this one S3 *does* /// enforce: it hashes the received part and rejects a mismatch with /// `BadDigest` before the bytes are durable. The caller must therefore send /// the header — it is in `SignedHeaders`, so omitting it fails the /// signature. This is transport integrity (the bytes S3 wrote are the bytes /// the client hashed), not a statement about what those bytes mean. pub async fn presign_upload_part( &self, key: &str, upload_id: &str, part_number: i32, expiry_secs: u64, max_bytes: Option, checksum_sha256: Option<&str>, ) -> Result { if !(1..=MULTIPART_MAX_PARTS as i32).contains(&part_number) { return Err(format!( "part number {part_number} out of range 1..={MULTIPART_MAX_PARTS}" )); } let presigning_config = PresigningConfig::builder() .expires_in(Duration::from_secs( expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS), )) .build() .map_err(|e| format!("Presigning config error: {e}"))?; let mut req = self .client .upload_part() .bucket(&self.bucket) .key(key) .upload_id(upload_id) .part_number(part_number); if let Some(n) = max_bytes { req = req.content_length(n); } if let Some(c) = checksum_sha256 { req = req.checksum_sha256(c); } let presigned = req .presigned(presigning_config) .await .map_err(|e| format!("Failed to generate upload part URL: {e}"))?; Ok(presigned.uri().to_string()) } /// Complete a multipart upload from the collected `(part_number, etag)` /// pairs. Parts are sorted by number before assembly (S3 requires ascending /// order), so the caller may pass them in completion order. /// /// Retries the completion on transient failure: by this point every part is /// uploaded and paid for, so losing the completion call would orphan the /// whole upload and force a restart from byte 0 (ultra-fuzz Run 11 Storage /// HIGH). The shared completion path for both the server-driven /// [`Self::upload_multipart`] and the client-direct session flow. pub async fn complete_multipart_upload( &self, key: &str, upload_id: &str, parts: &[(i32, String)], ) -> Result<(), String> { if parts.is_empty() { return Err("cannot complete a multipart upload with no parts".to_string()); } let mut parts = parts.to_vec(); parts.sort_by_key(|(n, _)| *n); let completed_parts: Vec = parts .into_iter() .map(|(n, etag)| CompletedPart::builder().e_tag(etag).part_number(n).build()) .collect(); let completed = CompletedMultipartUpload::builder() .set_parts(Some(completed_parts)) .build(); let mut attempt: u32 = 0; loop { attempt += 1; match self .client .complete_multipart_upload() .bucket(&self.bucket) .key(key) .upload_id(upload_id) .multipart_upload(completed.clone()) .send() .await { Ok(_) => return Ok(()), Err(e) if attempt < 3 => { let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2)); tracing::warn!( attempt, delay_ms, error = ?e, "S3 complete_multipart_upload transient failure, retrying" ); tokio::time::sleep(Duration::from_millis(delay_ms)).await; } Err(e) => { return Err(format!( "S3 complete multipart upload failed after retries: {e}" )); } } } } /// Read the file and upload every part, then complete the multipart upload. /// /// Returns `Err` (without aborting) on any failure; the caller /// [`Self::upload_multipart`] owns the single abort. Keep all abort handling /// out of here so the "abort on failure" contract can't be partially applied. async fn run_multipart_upload( &self, key: &str, file_path: &std::path::Path, part_size: usize, upload_id: &str, ) -> Result<(), String> { use tokio::io::AsyncReadExt; let mut file = tokio::fs::File::open(file_path) .await .map_err(|e| format!("Failed to open file for multipart upload: {e}"))?; let mut part_number: i32 = 1; let mut completed_parts: Vec<(i32, String)> = Vec::new(); loop { // One owned buffer per part, frozen into `Bytes` so each retry attempt // clones a refcount instead of re-copying the part. Previously the // body was `buf[..n].to_vec()` *inside* the retry loop, so every part // (and every retry) paid a full part_size heap copy on top of the // resident staging buffer — ~2x part_size live (ultra-fuzz Run 11 // Storage HIGH). `Bytes::from(Vec)` takes ownership without copying. let mut buf = vec![0u8; part_size]; let mut bytes_read = 0; // Fill the buffer completely (or until EOF) while bytes_read < part_size { match file.read(&mut buf[bytes_read..]).await { Ok(0) => break, Ok(n) => bytes_read += n, Err(e) => return Err(format!("Failed to read file: {e}")), } } if bytes_read == 0 { break; } buf.truncate(bytes_read); let part: bytes::Bytes = buf.into(); // Retry the part upload up to 3 times on transient failures. // S3 part uploads can flake on network blips; aborting the // whole multipart upload because of one timeout means the // caller has to restart from byte 0. Three attempts with // exponential backoff covers the common transient cases // without making a permanent failure (auth, oversize, etc.) // wait forever. let mut attempt: u32 = 0; let resp = loop { attempt += 1; let body = aws_sdk_s3::primitives::ByteStream::from(part.clone()); match self .client .upload_part() .bucket(&self.bucket) .key(key) .upload_id(upload_id) .part_number(part_number) .body(body) .send() .await { Ok(resp) => break Ok(resp), Err(e) if attempt < 3 => { // Backoff: 200ms, 800ms. Cheap enough not to // mask a permanent failure; long enough that // a brief network glitch resolves. let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2)); tracing::warn!( part_number, attempt, delay_ms, error = ?e, "S3 upload_part transient failure, retrying" ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; } Err(e) => break Err(e), } }; let resp = resp .map_err(|e| format!("S3 upload part {part_number} failed after retries: {e}"))?; let etag = resp.e_tag().unwrap_or_default().to_string(); completed_parts.push((part_number, etag)); part_number += 1; } if completed_parts.is_empty() { return Err("No parts uploaded (empty file)".to_string()); } // Shared completion path (with its own transient retry). The caller owns // the single abort-on-Err, so a failure here still aborts exactly once. self.complete_multipart_upload(key, upload_id, &completed_parts) .await } /// Abort a multipart upload, releasing any uploaded parts. /// /// Public and `Result`-returning so the pending-upload reaper can abort an /// orphaned session and retry on failure. Internal callers that only want /// best-effort cleanup (e.g. [`Self::upload_multipart`] unwinding a failed /// transfer) log the error and move on. pub async fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> Result<(), String> { self.client .abort_multipart_upload() .bucket(&self.bucket) .key(key) .upload_id(upload_id) .send() .await .map(|_| ()) .map_err(|e| format!("S3 abort multipart upload for {key} failed: {e}")) } /// List the in-progress multipart uploads for exactly `key`, returning their /// upload ids. /// /// The orphan reaper needs this: a session that was started but never /// completed has **no object to delete**, only uploaded parts that S3 bills /// for until they are aborted, so a plain `delete` is a no-op against it. The /// upload id is not recorded durably anywhere, so it is recovered from S3, /// which also catches sessions whose tracking row was lost entirely. /// /// `ListMultipartUploads` matches a *prefix*, so results are filtered to an /// exact key match — otherwise reaping `staging/abc` would also abort a live /// session for `staging/abcdef`. pub async fn list_multipart_uploads_for_key(&self, key: &str) -> Result, String> { let mut ids = Vec::new(); let mut key_marker: Option = None; let mut upload_id_marker: Option = None; loop { let mut req = self .client .list_multipart_uploads() .bucket(&self.bucket) .prefix(key); if let Some(ref k) = key_marker { req = req.key_marker(k); } if let Some(ref u) = upload_id_marker { req = req.upload_id_marker(u); } let resp = req .send() .await .map_err(|e| format!("S3 list_multipart_uploads for {key} failed: {e}"))?; for upload in resp.uploads() { // Exact-key filter: the request matched on prefix. if upload.key() == Some(key) && let Some(id) = upload.upload_id() { ids.push(id.to_string()); } } if resp.is_truncated().unwrap_or(false) { key_marker = resp.next_key_marker().map(str::to_string); upload_id_marker = resp.next_upload_id_marker().map(str::to_string); // Defensive: a truncated response with no markers would loop forever. if key_marker.is_none() && upload_id_marker.is_none() { break; } } else { break; } } Ok(ids) } /// Server-side copy an object using multipart `UploadPartCopy`, for sources /// larger than the 5 GiB single-part [`Self::copy_object`]/`CopyObject` /// limit. No bytes transit the caller — S3 copies each byte range internally. /// /// This is the >5 GiB promote path: scan-then-promote lifts a Clean staging /// object to the served content key, and a single `CopyObject` caps at 5 GiB, /// so a large video would otherwise succeed at upload and fail at promote — /// the worst failure position. `src_size` is the source object's size (the /// caller already reads it at confirm); `part_size` defaults to an /// auto-chosen size via [`MultipartPlan::auto`]. `content_type` sets the /// destination's type, since a fresh multipart upload does not inherit the /// source's metadata the way `CopyObject` does. /// /// Abort lives in one place, as in [`Self::upload_multipart`]: any failure /// after the destination upload is created aborts it exactly once. pub async fn copy_object_multipart( &self, src_bucket: &str, src_key: &str, dst_key: &str, content_type: &str, src_size: u64, part_size: Option, ) -> Result<(), String> { // Pre-flight: plan before anything is created, so a bad size strands // nothing and makes no network call. let plan = match part_size { Some(ps) => MultipartPlan::new(src_size, ps)?, None => MultipartPlan::auto(src_size)?, }; let upload_id = self.create_multipart_upload(dst_key, content_type).await?; match self .run_multipart_copy(src_bucket, src_key, dst_key, &plan, &upload_id) .await { Ok(()) => Ok(()), Err(e) => { if let Err(abort_err) = self.abort_multipart_upload(dst_key, &upload_id).await { tracing::warn!("Failed to abort multipart copy for {dst_key}: {abort_err}"); } Err(e) } } } /// Issue every `UploadPartCopy` for a ranged multipart copy, then complete /// it. Returns `Err` (without aborting) on any failure; the caller /// [`Self::copy_object_multipart`] owns the single abort. async fn run_multipart_copy( &self, src_bucket: &str, src_key: &str, dst_key: &str, plan: &MultipartPlan, upload_id: &str, ) -> Result<(), String> { // `CopySource` is `{bucket}/{key}`; keys are sanitized to // `[A-Za-z0-9._/-]` upstream, so no percent-encoding is needed (same as // `copy_object_from`). let copy_source = format!("{src_bucket}/{src_key}"); let mut completed_parts: Vec<(i32, String)> = Vec::with_capacity(plan.part_count as usize); for part_number in 1..=plan.part_count { let (start, end) = plan .part_range(part_number) .ok_or_else(|| format!("internal: part {part_number} outside plan range"))?; // Retry transient copy failures, mirroring the part-upload path. let mut attempt: u32 = 0; let resp = loop { attempt += 1; match self .client .upload_part_copy() .bucket(&self.bucket) .key(dst_key) .upload_id(upload_id) .part_number(part_number as i32) .copy_source(©_source) .copy_source_range(format!("bytes={start}-{end}")) .send() .await { Ok(r) => break Ok(r), Err(e) if attempt < 3 => { let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2)); tracing::warn!( part_number, attempt, delay_ms, error = ?e, "S3 upload_part_copy transient failure, retrying" ); tokio::time::sleep(Duration::from_millis(delay_ms)).await; } Err(e) => break Err(e), } }; let resp = resp.map_err(|e| { format!("S3 upload_part_copy part {part_number} failed after retries: {e}") })?; let etag = resp .copy_part_result() .and_then(|r| r.e_tag()) .unwrap_or_default() .to_string(); completed_parts.push((part_number as i32, etag)); } self.complete_multipart_upload(dst_key, upload_id, &completed_parts) .await } /// Configure CORS on the bucket for browser uploads. pub async fn configure_cors(&self, allowed_origin: &str) { let origin = allowed_origin.trim_end_matches('/').to_string(); let rule = match CorsRule::builder() .allowed_origins(&origin) .allowed_methods("PUT") .allowed_methods("GET") .allowed_methods("HEAD") .allowed_headers("Content-Type") .allowed_headers("Cache-Control") .allowed_headers("Content-Disposition") .expose_headers("ETag") .max_age_seconds(3600) .build() { Ok(r) => r, Err(e) => { tracing::warn!("Failed to build CORS rule: {}", e); return; } }; let cors_config = match CorsConfiguration::builder().cors_rules(rule).build() { Ok(c) => c, Err(e) => { tracing::warn!("Failed to build CORS config: {}", e); return; } }; match self .client .put_bucket_cors() .bucket(&self.bucket) .cors_configuration(cors_config) .send() .await { Ok(_) => tracing::info!("S3 bucket CORS configured for {}", origin), Err(e) => tracing::warn!("Failed to configure S3 CORS: {}", e), } } /// Lightweight connectivity check — `list_objects_v2` with `max_keys(0)`. pub async fn check_connectivity(&self) -> Result<(), String> { self.client .list_objects_v2() .bucket(&self.bucket) .max_keys(0) .send() .await .map(|_| ()) .map_err(|e| format!("{e}")) } } #[cfg(test)] mod tests { use super::*; fn test_client() -> S3Client { // `from_conf` is local — no network until a request is sent — so this // builds a usable client without reaching any endpoint. let s3_config = aws_sdk_s3::Config::builder() .behavior_version(BehaviorVersion::latest()) .http_client(https_client()) .region(Region::new("test")) .endpoint_url("http://127.0.0.1:1") .credentials_provider(Credentials::new("ak", "sk", None, None, "test")) .force_path_style(true) .build(); S3Client { client: Client::from_conf(s3_config), bucket: "test-bucket".to_string(), } } const MIB: u64 = 1024 * 1024; #[test] fn multipart_plan_divides_with_remainder() { // 25 MiB in 10 MiB parts -> 10 + 10 + 5. let plan = MultipartPlan::new(25 * MIB, 10 * MIB as usize).unwrap(); assert_eq!(plan.part_count, 3); assert_eq!(plan.part_len(1), 10 * MIB); assert_eq!(plan.part_len(2), 10 * MIB); assert_eq!(plan.part_len(3), 5 * MIB); assert_eq!(plan.part_len(4), 0, "out-of-range part"); assert_eq!(plan.part_range(1), Some((0, 10 * MIB - 1))); assert_eq!(plan.part_range(3), Some((20 * MIB, 25 * MIB - 1))); assert_eq!(plan.part_range(4), None); } #[test] fn multipart_plan_divides_evenly() { // 20 MiB in 5 MiB parts -> four full parts, last is a full part. let plan = MultipartPlan::new(20 * MIB, MULTIPART_MIN_PART_SIZE).unwrap(); assert_eq!(plan.part_count, 4); assert_eq!(plan.part_len(4), 5 * MIB); assert_eq!(plan.part_range(4), Some((15 * MIB, 20 * MIB - 1))); } #[test] fn multipart_plan_rejects_empty_object() { let err = MultipartPlan::new(0, MULTIPART_MIN_PART_SIZE).unwrap_err(); assert!(err.contains("non-empty"), "unexpected error: {err}"); } #[test] fn multipart_plan_rejects_undersized_part() { let err = MultipartPlan::new(100 * MIB, MULTIPART_MIN_PART_SIZE - 1).unwrap_err(); assert!(err.contains("5 MiB"), "unexpected error: {err}"); } #[test] fn multipart_plan_rejects_oversized_part() { let err = MultipartPlan::new(10 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1).unwrap_err(); assert!(err.contains("5 GiB"), "unexpected error: {err}"); } #[test] fn multipart_plan_rejects_too_many_parts() { // One 5 MiB part past the 10k limit at the minimum part size. let total = MULTIPART_MIN_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 1); let err = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap_err(); assert!(err.contains("10000-part"), "unexpected error: {err}"); } #[test] fn multipart_plan_accepts_exactly_max_parts() { let total = MULTIPART_MIN_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64; let plan = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap(); assert_eq!(plan.part_count, MULTIPART_MAX_PARTS); } #[test] fn multipart_plan_rejects_over_object_ceiling() { let err = MultipartPlan::new( MULTIPART_MAX_OBJECT_SIZE + 1, MULTIPART_MAX_PART_SIZE as usize, ) .unwrap_err(); assert!(err.contains("5 TiB"), "unexpected error: {err}"); } #[test] fn multipart_plan_auto_uses_default_for_small_objects() { let plan = MultipartPlan::auto(100 * MIB).unwrap(); assert_eq!(plan.part_size, MULTIPART_DEFAULT_PART_SIZE); // 100 MiB / 16 MiB -> 7 parts (ceil). assert_eq!(plan.part_count, 7); } #[test] fn multipart_plan_auto_scales_part_size_to_stay_within_part_cap() { // An object too big for the default part size within 10k parts must get // a larger part size, and the resulting plan must be valid. let big = MULTIPART_DEFAULT_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 500); let plan = MultipartPlan::auto(big).unwrap(); assert!(plan.part_size > MULTIPART_DEFAULT_PART_SIZE); assert!(plan.part_count <= MULTIPART_MAX_PARTS); // Whole-MiB part size. assert_eq!(plan.part_size as u64 % MIB, 0); } #[test] fn multipart_plan_auto_rejects_empty() { assert!(MultipartPlan::auto(0).is_err()); } #[tokio::test] async fn copy_object_multipart_rejects_empty_source_before_any_request() { // Plan is pre-flight: an empty source fails before the destination // multipart upload is created, so the unreachable endpoint is untouched. let client = test_client(); let err = client .copy_object_multipart("bkt", "src", "dst", "application/octet-stream", 0, None) .await .expect_err("empty source must be rejected"); assert!(err.contains("non-empty"), "unexpected error: {err}"); } /// The `X-Amz-SignedHeaders` list from a presigned URL. fn signed_headers(url: &str) -> String { url.split('&') .find_map(|p| p.strip_prefix("X-Amz-SignedHeaders=")) .map(|v| v.replace("%3B", ";")) .expect("presigned URL must carry X-Amz-SignedHeaders") } #[tokio::test] async fn presign_upload_signs_content_length_when_bound() { // Callers rely on `max_bytes` being enforced, and it is enforced only // because it lands in SignedHeaders: a client sending a different // Content-Length then fails the signature. That also makes the declared // size a hard contract — a caller that declares anything other than the // exact body length breaks every upload — so pin it here rather than // discovering it against production S3. let client = test_client(); let bound = client .presign_upload("k", "application/octet-stream", 900, None, Some(12_345)) .await .unwrap(); let headers = signed_headers(&bound); assert!( headers.contains("content-length"), "max_bytes must be signed, got: {headers}" ); let unbound = client .presign_upload("k", "application/octet-stream", 900, None, None) .await .unwrap(); assert!( !signed_headers(&unbound).contains("content-length"), "without max_bytes the client is free to send any length" ); } #[tokio::test] async fn presign_upload_part_rejects_out_of_range_part_number() { // Pre-flight range check: fires before any network call, so the // unreachable dummy endpoint is never touched. let client = test_client(); for bad in [0, MULTIPART_MAX_PARTS as i32 + 1] { let err = client .presign_upload_part("k", "uid", bad, 3600, None, None) .await .expect_err("out-of-range part number must be rejected"); assert!(err.contains("out of range"), "unexpected error: {err}"); } } #[tokio::test] async fn presign_upload_part_signs_the_checksum_when_bound() { // S3 enforces a bound checksum by rehashing the part, but only if the // client sends the header — which it must, because signing it makes it // mandatory. Both halves of that live in SignedHeaders. let client = test_client(); let bound = client .presign_upload_part("k", "uid", 1, 900, Some(64), Some("Zm9vYmFyYmF6")) .await .unwrap(); let headers = signed_headers(&bound); assert!( headers.contains("x-amz-checksum-sha256"), "a bound checksum must be signed, got: {headers}" ); let unbound = client .presign_upload_part("k", "uid", 1, 900, Some(64), None) .await .unwrap(); assert!( !signed_headers(&unbound).contains("checksum"), "no checksum bound means no checksum header is required" ); } #[tokio::test] async fn complete_multipart_rejects_empty_parts() { let client = test_client(); let err = client .complete_multipart_upload("k", "uid", &[]) .await .expect_err("empty parts must be rejected"); assert!(err.contains("no parts"), "unexpected error: {err}"); } #[tokio::test] async fn upload_multipart_rejects_undersized_part_before_any_request() { // The minimum-part-size guard is pre-flight: it must fire before the // multipart upload is created, so there is nothing to strand and no // network call (the dummy endpoint is unreachable — reaching it would // hang/error instead of returning this exact message). let client = test_client(); let path = std::path::Path::new("/nonexistent"); let err = client .upload_multipart("k", "application/octet-stream", path, Some(1024)) .await .expect_err("undersized part size must be rejected"); assert!(err.contains("at least 5 MB"), "unexpected error: {err}"); } }