Skip to main content

max / makenotwork

54.4 KB · 1409 lines History Blame Raw
1 //! Shared S3-compatible storage client.
2 //!
3 //! Overview: maintainer wiki.
4 //! <!-- wiki: s3-storage-overview -->
5 //!
6 //! Thin wrapper around the AWS SDK providing upload, download, delete,
7 //! presigned URL generation, and bucket management. Used by MNW and
8 //! Multithreaded to avoid duplicating S3 initialization and operations.
9
10 #![warn(missing_docs)]
11
12 use aws_config::BehaviorVersion;
13 use aws_sdk_s3::Client;
14 use aws_sdk_s3::config::{Credentials, Region};
15 use aws_sdk_s3::presigning::PresigningConfig;
16 use aws_sdk_s3::types::{
17 CompletedMultipartUpload, CompletedPart, CorsConfiguration, CorsRule, Delete, ObjectIdentifier,
18 };
19 use std::time::Duration;
20
21 pub use aws_sdk_s3::primitives::ByteStream;
22
23 /// The HTTPS client every S3 client is built on.
24 ///
25 /// Built here rather than taken from the SDK's `default-https-client` feature,
26 /// which is a hard alias for the aws-lc-rs (C) crypto backend. Ring is the
27 /// pure-Rust provider the rest of the tree uses, and the SDK will accept it only
28 /// through an explicitly constructed client.
29 fn https_client() -> aws_sdk_s3::config::SharedHttpClient {
30 aws_smithy_http_client::Builder::new()
31 .tls_provider(aws_smithy_http_client::tls::Provider::Rustls(
32 aws_smithy_http_client::tls::rustls_provider::CryptoMode::Ring,
33 ))
34 .build_https()
35 }
36
37 /// S3 connection configuration.
38 #[derive(Debug, Clone)]
39 pub struct S3Config {
40 /// Endpoint URL (e.g., `https://fsn1.your-objectstorage.com`)
41 pub endpoint: String,
42 /// Bucket name
43 pub bucket: String,
44 /// Access key ID
45 pub access_key: String,
46 /// Secret access key
47 pub secret_key: String,
48 /// Region (e.g., `fsn1`)
49 pub region: String,
50 }
51
52 /// S3 client wrapper.
53 #[derive(Clone)]
54 pub struct S3Client {
55 client: Client,
56 bucket: String,
57 }
58
59 /// SigV4's hard maximum presign lifetime (7 days). A presign request above this
60 /// is rejected by the signer, so we clamp callers to it rather than surface an
61 /// opaque signing error — and it bounds how long any single minted URL can live.
62 const MAX_PRESIGN_EXPIRY_SECS: u64 = 7 * 24 * 60 * 60;
63
64 /// S3 multipart-upload limits, straight from the S3 API contract. Every layer
65 /// that plans a client-direct multipart upload (the blob and creator-media
66 /// session endpoints) validates against these, so the client is never handed a
67 /// plan S3 would reject at complete time.
68 ///
69 /// Minimum size of every part *except the last* (5 MiB). The final part may be
70 /// anything down to 1 byte.
71 pub const MULTIPART_MIN_PART_SIZE: usize = 5 * 1024 * 1024;
72 /// Maximum number of parts in a single multipart upload.
73 pub const MULTIPART_MAX_PARTS: u32 = 10_000;
74 /// Maximum size of a single part (5 GiB).
75 pub const MULTIPART_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
76 /// Maximum size of an object assembled from a multipart upload (5 TiB).
77 pub const MULTIPART_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024;
78 /// Default/floor part size for an auto-planned multipart upload (16 MiB): big
79 /// enough to keep the per-part round-trip overhead low, small enough that a
80 /// resumable client re-sends little on a retry. Used when the object is small
81 /// enough not to force larger parts to stay within the part-count limit.
82 pub const MULTIPART_DEFAULT_PART_SIZE: usize = 16 * 1024 * 1024;
83
84 // The part budget at max part size must be able to cover the object ceiling, or
85 // a large-but-legal object would be unplannable at any part size. 10,000 x 5 GiB
86 // ~= 48.8 TiB, comfortably over the 5 TiB object cap. (The 5 MiB *minimum* part
87 // deliberately does NOT cover 5 TiB — a large object simply needs bigger parts,
88 // which `MultipartPlan::new` enforces via the part-count limit.)
89 const _: () = assert!(
90 MULTIPART_MAX_PARTS as u128 * MULTIPART_MAX_PART_SIZE as u128
91 >= MULTIPART_MAX_OBJECT_SIZE as u128
92 );
93
94 /// A validated multipart-upload plan: the part size to use and how many parts a
95 /// body of `total_size` bytes splits into. Pure arithmetic with no S3 call, so
96 /// the blob and creator-media session endpoints share one source of truth for
97 /// part geometry (and the client computes the same boundaries independently).
98 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
99 pub struct MultipartPlan {
100 /// Total object size in bytes.
101 pub total_size: u64,
102 /// Size of every part except the last.
103 pub part_size: usize,
104 /// Number of parts (1..=[`MULTIPART_MAX_PARTS`]).
105 pub part_count: u32,
106 }
107
108 impl MultipartPlan {
109 /// Plan a multipart upload of `total_size` bytes into `part_size`-byte parts,
110 /// the last part taking the remainder.
111 ///
112 /// Errors if the object is empty (use a single PUT), exceeds the 5 TiB
113 /// multipart ceiling, the part size is below S3's 5 MiB floor or above its
114 /// 5 GiB ceiling, or the resulting part count would exceed 10,000 — the last
115 /// meaning the part size is too small for this object and the caller should
116 /// pick a larger one.
117 pub fn new(total_size: u64, part_size: usize) -> Result<Self, String> {
118 if total_size == 0 {
119 return Err(
120 "multipart upload needs a non-empty object; use a single PUT for empty objects"
121 .to_string(),
122 );
123 }
124 if total_size > MULTIPART_MAX_OBJECT_SIZE {
125 return Err(format!(
126 "object is {total_size} bytes, over the {MULTIPART_MAX_OBJECT_SIZE}-byte (5 TiB) multipart ceiling"
127 ));
128 }
129 if part_size < MULTIPART_MIN_PART_SIZE {
130 return Err(format!(
131 "part size {part_size} is below the {MULTIPART_MIN_PART_SIZE}-byte (5 MiB) S3 minimum"
132 ));
133 }
134 if part_size as u64 > MULTIPART_MAX_PART_SIZE {
135 return Err(format!(
136 "part size {part_size} is above the {MULTIPART_MAX_PART_SIZE}-byte (5 GiB) S3 maximum"
137 ));
138 }
139 let part_count = total_size.div_ceil(part_size as u64);
140 if part_count > MULTIPART_MAX_PARTS as u64 {
141 return Err(format!(
142 "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"
143 ));
144 }
145 Ok(Self {
146 total_size,
147 part_size,
148 part_count: part_count as u32,
149 })
150 }
151
152 /// Plan a multipart upload of `total_size` bytes, choosing the part size
153 /// automatically: [`MULTIPART_DEFAULT_PART_SIZE`] when that keeps the object
154 /// within [`MULTIPART_MAX_PARTS`] parts, otherwise the smallest whole-MiB
155 /// part size that does. Errors only if the object is empty or over the 5 TiB
156 /// ceiling — a valid non-empty object always yields a plan.
157 pub fn auto(total_size: u64) -> Result<Self, String> {
158 const MIB: u64 = 1024 * 1024;
159 // Smallest part size that fits the object within the part-count cap,
160 // rounded up to a whole MiB, then floored at the default part size.
161 let needed = total_size.div_ceil(MULTIPART_MAX_PARTS as u64);
162 let rounded = needed.div_ceil(MIB) * MIB;
163 let part_size = (rounded as usize).max(MULTIPART_DEFAULT_PART_SIZE);
164 Self::new(total_size, part_size)
165 }
166
167 /// Byte length of part `part_number` (1-based). The last part is the
168 /// remainder; every earlier part is exactly `part_size`. Returns 0 for a
169 /// part number outside `1..=part_count`.
170 pub fn part_len(&self, part_number: u32) -> u64 {
171 if part_number == 0 || part_number > self.part_count {
172 return 0;
173 }
174 if part_number < self.part_count {
175 return self.part_size as u64;
176 }
177 // Last part: the remainder, or a full part when the size divides evenly.
178 match self.total_size % self.part_size as u64 {
179 0 => self.part_size as u64,
180 rem => rem,
181 }
182 }
183
184 /// Inclusive `[start, end]` byte range of part `part_number` (1-based), the
185 /// form S3 `UploadPartCopy` wants as `bytes=start-end`. `None` for a part
186 /// number outside `1..=part_count`.
187 pub fn part_range(&self, part_number: u32) -> Option<(u64, u64)> {
188 if part_number == 0 || part_number > self.part_count {
189 return None;
190 }
191 let start = (part_number as u64 - 1) * self.part_size as u64;
192 Some((start, start + self.part_len(part_number) - 1))
193 }
194 }
195
196 impl S3Client {
197 /// Create a new S3 client from configuration.
198 // Public async constructor: kept async for API stability across callers.
199 #[allow(clippy::unused_async)]
200 pub async fn new(config: &S3Config) -> Result<Self, String> {
201 let credentials = Credentials::new(
202 &config.access_key,
203 &config.secret_key,
204 None,
205 None,
206 "s3-storage",
207 );
208
209 // Bound every S3 operation so a hung endpoint (e.g. a stalled
210 // HeadObject on a blob-confirm path) can't wedge a caller forever. These
211 // apply to establishing the connection and to a single attempt's
212 // round-trip to first byte of the response — they do NOT cap the time
213 // spent streaming a large object body, so big uploads/downloads are
214 // unaffected. The SDK's default retry policy still applies per attempt.
215 let timeout_config = aws_sdk_s3::config::timeout::TimeoutConfig::builder()
216 .connect_timeout(Duration::from_secs(10))
217 .operation_attempt_timeout(Duration::from_mins(1))
218 .build();
219
220 let s3_config = aws_sdk_s3::Config::builder()
221 .behavior_version(BehaviorVersion::latest())
222 .http_client(https_client())
223 .region(Region::new(config.region.clone()))
224 .endpoint_url(&config.endpoint)
225 .credentials_provider(credentials)
226 .timeout_config(timeout_config)
227 .force_path_style(true)
228 .build();
229
230 let client = Client::from_conf(s3_config);
231
232 Ok(Self {
233 client,
234 bucket: config.bucket.clone(),
235 })
236 }
237
238 /// Bucket name accessor.
239 pub fn bucket(&self) -> &str {
240 &self.bucket
241 }
242
243 /// Upload bytes to S3.
244 pub async fn upload(
245 &self,
246 key: &str,
247 content_type: &str,
248 data: Vec<u8>,
249 cache_control: Option<&str>,
250 ) -> Result<(), String> {
251 let mut req = self
252 .client
253 .put_object()
254 .bucket(&self.bucket)
255 .key(key)
256 .content_type(content_type)
257 .body(data.into());
258
259 if let Some(cc) = cache_control {
260 req = req.cache_control(cc);
261 }
262
263 req.send()
264 .await
265 .map_err(|e| format!("S3 upload failed: {e}"))?;
266
267 Ok(())
268 }
269
270 /// Download bytes from S3. Returns `(data, content_type)`.
271 ///
272 /// Convenience `Vec<u8>` form; for the zero-extra-copy path use
273 /// [`download_buf`](Self::download_buf), which returns the aggregated
274 /// `Bytes` directly.
275 pub async fn download(&self, key: &str) -> Result<(Vec<u8>, String), String> {
276 let (bytes, content_type) = self.download_buf(key).await?;
277 Ok((bytes.to_vec(), content_type))
278 }
279
280 /// Download an object as `bytes::Bytes`, returning `(data, content_type)`.
281 ///
282 /// Unlike [`download`](Self::download) this does not copy the aggregated
283 /// body into a fresh `Vec` — the caller gets the SDK's buffer directly. Use
284 /// it on memory-sensitive paths (e.g. the scanner's buffered branch) where
285 /// the extra `to_vec` would transiently double the footprint.
286 pub async fn download_buf(&self, key: &str) -> Result<(bytes::Bytes, String), String> {
287 let resp = self
288 .client
289 .get_object()
290 .bucket(&self.bucket)
291 .key(key)
292 .send()
293 .await
294 .map_err(|e| format!("S3 download failed: {e}"))?;
295
296 let content_type = resp
297 .content_type()
298 .unwrap_or("application/octet-stream")
299 .to_string();
300
301 let bytes = resp
302 .body
303 .collect()
304 .await
305 .map_err(|e| format!("S3 read body failed: {e}"))?;
306
307 Ok((bytes.into_bytes(), content_type))
308 }
309
310 /// Stream an object's body from S3 without buffering. Caller drives the
311 /// `ByteStream` to disk or hands it to a layer that wants chunks.
312 pub async fn download_stream(
313 &self,
314 key: &str,
315 ) -> Result<aws_sdk_s3::primitives::ByteStream, String> {
316 let resp = self
317 .client
318 .get_object()
319 .bucket(&self.bucket)
320 .key(key)
321 .send()
322 .await
323 .map_err(|e| format!("S3 download failed: {e}"))?;
324
325 Ok(resp.body)
326 }
327
328 /// Download only the first `len` bytes of an object via a ranged
329 /// `GetObject` (`Range: bytes=0-{len-1}`). Used for content sniffing so a
330 /// 4 KB header read doesn't initiate a transfer of the whole object. Returns
331 /// fewer bytes if the object is smaller than `len`.
332 pub async fn download_head(&self, key: &str, len: usize) -> Result<Vec<u8>, String> {
333 if len == 0 {
334 return Ok(Vec::new());
335 }
336 let resp = self
337 .client
338 .get_object()
339 .bucket(&self.bucket)
340 .key(key)
341 .range(format!("bytes=0-{}", len - 1))
342 .send()
343 .await
344 .map_err(|e| format!("S3 ranged download failed: {e}"))?;
345 let data = resp
346 .body
347 .collect()
348 .await
349 .map_err(|e| format!("S3 ranged body read failed: {e}"))?;
350 Ok(data.to_vec())
351 }
352
353 /// Delete an object from S3.
354 pub async fn delete(&self, key: &str) -> Result<(), String> {
355 self.client
356 .delete_object()
357 .bucket(&self.bucket)
358 .key(key)
359 .send()
360 .await
361 .map_err(|e| format!("S3 delete failed: {e}"))?;
362
363 Ok(())
364 }
365
366 /// Server-side copy an object from `src_key` to `dst_key` within this
367 /// bucket. No bytes transit the caller — S3 performs the copy internally.
368 ///
369 /// This is the primitive behind scan-then-promote: an upload is scanned at a
370 /// staging key the client can overwrite, and only a Clean object is copied to
371 /// the served key (which the client holds no presign for), so the served
372 /// bytes are provably the scanned bytes (ultra-fuzz Run #24 Storage HIGH).
373 ///
374 /// `CopySource` is `{bucket}/{key}`. Keys in this system are sanitized to
375 /// `[A-Za-z0-9._/-]` (see the server's `sanitize_filename`), none of which
376 /// require percent-encoding, so the source is formed directly.
377 pub async fn copy_object(&self, src_key: &str, dst_key: &str) -> Result<(), String> {
378 self.copy_object_from(&self.bucket, src_key, dst_key).await
379 }
380
381 /// Server-side copy from an arbitrary source bucket into THIS client's
382 /// bucket. Used by the scan-then-promote path to lift a Clean object from
383 /// the private staging bucket into the public (CDN-served) bucket in one
384 /// server-side operation (no bytes transit the process). The credentials
385 /// this client holds must have read on `src_bucket`; in this deployment all
386 /// buckets share one Hetzner project/key, so cross-bucket copy is permitted.
387 ///
388 /// `CopySource` is `{src_bucket}/{src_key}`. Keys are sanitized to
389 /// `[A-Za-z0-9._/-]` (server `sanitize_filename`), so no percent-encoding is
390 /// needed and the source is formed directly. When `src_bucket` equals this
391 /// client's bucket the copy is the ordinary same-bucket promote.
392 pub async fn copy_object_from(
393 &self,
394 src_bucket: &str,
395 src_key: &str,
396 dst_key: &str,
397 ) -> Result<(), String> {
398 self.client
399 .copy_object()
400 .bucket(&self.bucket)
401 .copy_source(format!("{src_bucket}/{src_key}"))
402 .key(dst_key)
403 .send()
404 .await
405 .map_err(|e| {
406 format!(
407 "S3 copy_object {src_bucket}/{src_key} -> {}/{dst_key} failed: {e}",
408 self.bucket
409 )
410 })?;
411 Ok(())
412 }
413
414 /// Delete a batch of objects in a single S3 `DeleteObjects` request.
415 ///
416 /// S3 accepts up to 1000 keys per call; the caller is responsible for
417 /// chunking. Returns the keys that failed (if any) along with their
418 /// per-object error message. A successful response with `Errors` is
419 /// not a hard error — partial success is normal for batched deletes.
420 pub async fn delete_objects(&self, keys: &[String]) -> Result<Vec<(String, String)>, String> {
421 if keys.is_empty() {
422 return Ok(Vec::new());
423 }
424 // A key that won't build an ObjectIdentifier is reported as a failure
425 // rather than silently dropped — otherwise the caller believes it was
426 // deleted and the object leaks forever (ultra-fuzz Run 11 Storage MED).
427 let mut failures: Vec<(String, String)> = Vec::new();
428 let mut objects: Vec<ObjectIdentifier> = Vec::with_capacity(keys.len());
429 for k in keys {
430 match ObjectIdentifier::builder().key(k).build() {
431 Ok(o) => objects.push(o),
432 Err(e) => failures.push((k.clone(), format!("malformed key: {e}"))),
433 }
434 }
435 if objects.is_empty() {
436 return Ok(failures);
437 }
438 let delete = Delete::builder()
439 .set_objects(Some(objects))
440 .quiet(true)
441 .build()
442 .map_err(|e| format!("S3 delete_objects build failed: {e}"))?;
443 let resp = self
444 .client
445 .delete_objects()
446 .bucket(&self.bucket)
447 .delete(delete)
448 .send()
449 .await
450 .map_err(|e| format!("S3 delete_objects failed: {e}"))?;
451 failures.extend(
452 resp.errors
453 .unwrap_or_default()
454 .into_iter()
455 .filter_map(|err| {
456 let key = err.key?;
457 let msg = err.message.unwrap_or_else(|| "<no message>".into());
458 Some((key, msg))
459 }),
460 );
461 Ok(failures)
462 }
463
464 /// Delete all objects under a given key prefix.
465 ///
466 /// Lists in pages of 1000 and deletes each page in a single batched
467 /// `DeleteObjects` call (S3's max). For a 50k-key prefix this is 50
468 /// round-trips instead of 50,000.
469 pub async fn delete_prefix(&self, prefix: &str) -> Result<(), String> {
470 let mut continuation_token: Option<String> = None;
471 loop {
472 let mut req = self
473 .client
474 .list_objects_v2()
475 .bucket(&self.bucket)
476 .prefix(prefix)
477 .max_keys(1000);
478 if let Some(ref token) = continuation_token {
479 req = req.continuation_token(token);
480 }
481 let resp = req
482 .send()
483 .await
484 .map_err(|e| format!("S3 list objects failed: {e}"))?;
485
486 let keys: Vec<String> = resp
487 .contents
488 .unwrap_or_default()
489 .into_iter()
490 .filter_map(|obj| obj.key)
491 .collect();
492
493 if !keys.is_empty() {
494 let failures = self.delete_objects(&keys).await?;
495 if !failures.is_empty() {
496 // Surface per-key failures to the caller (and abort further
497 // pages): the durable deletion queue is the convergence
498 // backstop, but a silent partial wipe would hide a bucket
499 // policy / permissions problem. Preview the first few keys.
500 let preview: Vec<String> = failures
501 .iter()
502 .take(5)
503 .map(|(k, e)| format!("{k}: {e}"))
504 .collect();
505 return Err(format!(
506 "S3 delete_prefix partial failure: {} keys failed (first 5: {})",
507 failures.len(),
508 preview.join(", ")
509 ));
510 }
511 }
512
513 if resp.is_truncated.unwrap_or(false) {
514 continuation_token = resp.next_continuation_token;
515 } else {
516 break;
517 }
518 }
519 Ok(())
520 }
521
522 /// Check if an object exists in S3.
523 pub async fn object_exists(&self, key: &str) -> Result<bool, String> {
524 match self
525 .client
526 .head_object()
527 .bucket(&self.bucket)
528 .key(key)
529 .send()
530 .await
531 {
532 Ok(_) => Ok(true),
533 Err(e) => {
534 let service_error = e.into_service_error();
535 if service_error.is_not_found() {
536 Ok(false)
537 } else {
538 Err(format!("S3 head_object failed: {service_error}"))
539 }
540 }
541 }
542 }
543
544 /// Get the size of an object in bytes, or `None` if not found.
545 pub async fn object_size(&self, key: &str) -> Result<Option<i64>, String> {
546 match self
547 .client
548 .head_object()
549 .bucket(&self.bucket)
550 .key(key)
551 .send()
552 .await
553 {
554 Ok(resp) => Ok(resp.content_length()),
555 Err(e) => {
556 let service_error = e.into_service_error();
557 if service_error.is_not_found() {
558 Ok(None)
559 } else {
560 Err(format!("S3 head_object failed: {service_error}"))
561 }
562 }
563 }
564 }
565
566 /// Generate a presigned URL for uploading.
567 ///
568 /// When `max_bytes` is set, the value is signed into the request as
569 /// `Content-Length` (a SignedHeader). A client that sends a *different*
570 /// `Content-Length` than the one signed produces a signature mismatch, so
571 /// the common "lie about the size" case fails. This is NOT a hard,
572 /// server-enforced ceiling, though: it relies on the client sending an
573 /// honest `Content-Length`, and S3-compatible backends (Ceph/MinIO/Garage)
574 /// vary in how strictly they reconcile the declared length with the actual
575 /// body. Treat it as defense-in-depth, not the boundary — the authoritative
576 /// per-file cap is enforced at confirm time, where the handler reads the
577 /// object's real size via [`Self::object_size`] before crediting storage.
578 /// (A presigned POST with a `content-length-range` policy would be a true
579 /// server-side range check; it is not used here because every upload path is
580 /// a single PUT.)
581 pub async fn presign_upload(
582 &self,
583 key: &str,
584 content_type: &str,
585 expiry_secs: u64,
586 cache_control: Option<&str>,
587 max_bytes: Option<i64>,
588 ) -> Result<String, String> {
589 let presigning_config = PresigningConfig::builder()
590 .expires_in(Duration::from_secs(
591 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
592 ))
593 .build()
594 .map_err(|e| format!("Presigning config error: {e}"))?;
595
596 let mut req = self
597 .client
598 .put_object()
599 .bucket(&self.bucket)
600 .key(key)
601 .content_type(content_type);
602
603 if let Some(cc) = cache_control {
604 req = req.cache_control(cc);
605 }
606
607 if let Some(n) = max_bytes {
608 req = req.content_length(n);
609 }
610
611 let presigned = req
612 .presigned(presigning_config)
613 .await
614 .map_err(|e| format!("Failed to generate upload URL: {e}"))?;
615
616 Ok(presigned.uri().to_string())
617 }
618
619 /// Generate a presigned URL for downloading.
620 pub async fn presign_download(&self, key: &str, expiry_secs: u64) -> Result<String, String> {
621 let presigning_config = PresigningConfig::builder()
622 .expires_in(Duration::from_secs(
623 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
624 ))
625 .build()
626 .map_err(|e| format!("Presigning config error: {e}"))?;
627
628 let presigned = self
629 .client
630 .get_object()
631 .bucket(&self.bucket)
632 .key(key)
633 .presigned(presigning_config)
634 .await
635 .map_err(|e| format!("Failed to generate download URL: {e}"))?;
636
637 Ok(presigned.uri().to_string())
638 }
639
640 /// Upload a file to S3 using multipart upload.
641 ///
642 /// Reads the file in `part_size` chunks (default 10 MB, minimum 5 MB) and
643 /// uploads each as a part. Aborts the multipart upload on any failure.
644 ///
645 /// The abort lives in exactly ONE place: once the multipart upload has been
646 /// created, all the fallible part/complete work runs in
647 /// the private `run_multipart_upload`, and this wrapper aborts on any `Err` it
648 /// returns. So a future failure path added inside the inner method aborts by
649 /// construction — it cannot forget to (the Sto-S1 fix, made structural rather
650 /// than per-branch).
651 pub async fn upload_multipart(
652 &self,
653 key: &str,
654 content_type: &str,
655 file_path: &std::path::Path,
656 part_size: Option<usize>,
657 ) -> Result<(), String> {
658 let part_size = part_size.unwrap_or(10 * 1024 * 1024); // 10 MB default
659 if part_size < 5 * 1024 * 1024 {
660 // Pre-flight: nothing created yet, nothing to abort.
661 return Err("Multipart part size must be at least 5 MB".to_string());
662 }
663
664 let upload_id = self.create_multipart_upload(key, content_type).await?;
665
666 // Single abort site: any failure past this point aborts exactly once.
667 match self
668 .run_multipart_upload(key, file_path, part_size, &upload_id)
669 .await
670 {
671 Ok(()) => Ok(()),
672 Err(e) => {
673 // Best-effort abort; the create succeeded, so on failure surface
674 // the original error and just log if the cleanup also fails.
675 if let Err(abort_err) = self.abort_multipart_upload(key, &upload_id).await {
676 tracing::warn!("Failed to abort multipart upload for {key}: {abort_err}");
677 }
678 Err(e)
679 }
680 }
681 }
682
683 /// Begin a client-direct multipart upload, returning the `upload_id` that
684 /// [`Self::presign_upload_part`], [`Self::complete_multipart_upload`], and
685 /// [`Self::abort_multipart_upload`] key on.
686 ///
687 /// The counterpart to [`Self::upload_multipart`], which drives the whole
688 /// transfer server-side from a local file. Here the server only mints the id
689 /// and (via `presign_upload_part`) the per-part URLs; the client streams the
690 /// parts straight to S3, so no object bytes transit the server.
691 pub async fn create_multipart_upload(
692 &self,
693 key: &str,
694 content_type: &str,
695 ) -> Result<String, String> {
696 let create = self
697 .client
698 .create_multipart_upload()
699 .bucket(&self.bucket)
700 .key(key)
701 .content_type(content_type)
702 .send()
703 .await
704 .map_err(|e| format!("S3 create multipart upload failed: {e}"))?;
705
706 create
707 .upload_id()
708 .map(str::to_string)
709 .ok_or_else(|| "S3 create multipart upload returned no upload_id".to_string())
710 }
711
712 /// Presign an `UploadPart` request for one part of an in-progress multipart
713 /// upload. `part_number` is 1-based (`1..=`[`MULTIPART_MAX_PARTS`]).
714 ///
715 /// When `max_bytes` is set it is signed as `Content-Length`, the same
716 /// defense-in-depth (not a hard, server-enforced ceiling) as
717 /// [`Self::presign_upload`] — see its docs for why the authoritative size
718 /// check still lives at confirm time.
719 ///
720 /// `checksum_sha256` (base64 of the raw 32-byte digest) is signed as
721 /// `x-amz-checksum-sha256`, and unlike the length this one S3 *does*
722 /// enforce: it hashes the received part and rejects a mismatch with
723 /// `BadDigest` before the bytes are durable. The caller must therefore send
724 /// the header — it is in `SignedHeaders`, so omitting it fails the
725 /// signature. This is transport integrity (the bytes S3 wrote are the bytes
726 /// the client hashed), not a statement about what those bytes mean.
727 pub async fn presign_upload_part(
728 &self,
729 key: &str,
730 upload_id: &str,
731 part_number: i32,
732 expiry_secs: u64,
733 max_bytes: Option<i64>,
734 checksum_sha256: Option<&str>,
735 ) -> Result<String, String> {
736 if !(1..=MULTIPART_MAX_PARTS as i32).contains(&part_number) {
737 return Err(format!(
738 "part number {part_number} out of range 1..={MULTIPART_MAX_PARTS}"
739 ));
740 }
741 let presigning_config = PresigningConfig::builder()
742 .expires_in(Duration::from_secs(
743 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
744 ))
745 .build()
746 .map_err(|e| format!("Presigning config error: {e}"))?;
747
748 let mut req = self
749 .client
750 .upload_part()
751 .bucket(&self.bucket)
752 .key(key)
753 .upload_id(upload_id)
754 .part_number(part_number);
755
756 if let Some(n) = max_bytes {
757 req = req.content_length(n);
758 }
759 if let Some(c) = checksum_sha256 {
760 req = req.checksum_sha256(c);
761 }
762
763 let presigned = req
764 .presigned(presigning_config)
765 .await
766 .map_err(|e| format!("Failed to generate upload part URL: {e}"))?;
767
768 Ok(presigned.uri().to_string())
769 }
770
771 /// Complete a multipart upload from the collected `(part_number, etag)`
772 /// pairs. Parts are sorted by number before assembly (S3 requires ascending
773 /// order), so the caller may pass them in completion order.
774 ///
775 /// Retries the completion on transient failure: by this point every part is
776 /// uploaded and paid for, so losing the completion call would orphan the
777 /// whole upload and force a restart from byte 0 (ultra-fuzz Run 11 Storage
778 /// HIGH). The shared completion path for both the server-driven
779 /// [`Self::upload_multipart`] and the client-direct session flow.
780 pub async fn complete_multipart_upload(
781 &self,
782 key: &str,
783 upload_id: &str,
784 parts: &[(i32, String)],
785 ) -> Result<(), String> {
786 if parts.is_empty() {
787 return Err("cannot complete a multipart upload with no parts".to_string());
788 }
789 let mut parts = parts.to_vec();
790 parts.sort_by_key(|(n, _)| *n);
791 let completed_parts: Vec<CompletedPart> = parts
792 .into_iter()
793 .map(|(n, etag)| CompletedPart::builder().e_tag(etag).part_number(n).build())
794 .collect();
795
796 let completed = CompletedMultipartUpload::builder()
797 .set_parts(Some(completed_parts))
798 .build();
799
800 let mut attempt: u32 = 0;
801 loop {
802 attempt += 1;
803 match self
804 .client
805 .complete_multipart_upload()
806 .bucket(&self.bucket)
807 .key(key)
808 .upload_id(upload_id)
809 .multipart_upload(completed.clone())
810 .send()
811 .await
812 {
813 Ok(_) => return Ok(()),
814 Err(e) if attempt < 3 => {
815 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
816 tracing::warn!(
817 attempt, delay_ms, error = ?e,
818 "S3 complete_multipart_upload transient failure, retrying"
819 );
820 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
821 }
822 Err(e) => {
823 return Err(format!(
824 "S3 complete multipart upload failed after retries: {e}"
825 ));
826 }
827 }
828 }
829 }
830
831 /// Read the file and upload every part, then complete the multipart upload.
832 ///
833 /// Returns `Err` (without aborting) on any failure; the caller
834 /// [`Self::upload_multipart`] owns the single abort. Keep all abort handling
835 /// out of here so the "abort on failure" contract can't be partially applied.
836 async fn run_multipart_upload(
837 &self,
838 key: &str,
839 file_path: &std::path::Path,
840 part_size: usize,
841 upload_id: &str,
842 ) -> Result<(), String> {
843 use tokio::io::AsyncReadExt;
844
845 let mut file = tokio::fs::File::open(file_path)
846 .await
847 .map_err(|e| format!("Failed to open file for multipart upload: {e}"))?;
848
849 let mut part_number: i32 = 1;
850 let mut completed_parts: Vec<(i32, String)> = Vec::new();
851
852 loop {
853 // One owned buffer per part, frozen into `Bytes` so each retry attempt
854 // clones a refcount instead of re-copying the part. Previously the
855 // body was `buf[..n].to_vec()` *inside* the retry loop, so every part
856 // (and every retry) paid a full part_size heap copy on top of the
857 // resident staging buffer — ~2x part_size live (ultra-fuzz Run 11
858 // Storage HIGH). `Bytes::from(Vec)` takes ownership without copying.
859 let mut buf = vec![0u8; part_size];
860 let mut bytes_read = 0;
861 // Fill the buffer completely (or until EOF)
862 while bytes_read < part_size {
863 match file.read(&mut buf[bytes_read..]).await {
864 Ok(0) => break,
865 Ok(n) => bytes_read += n,
866 Err(e) => return Err(format!("Failed to read file: {e}")),
867 }
868 }
869
870 if bytes_read == 0 {
871 break;
872 }
873
874 buf.truncate(bytes_read);
875 let part: bytes::Bytes = buf.into();
876
877 // Retry the part upload up to 3 times on transient failures.
878 // S3 part uploads can flake on network blips; aborting the
879 // whole multipart upload because of one timeout means the
880 // caller has to restart from byte 0. Three attempts with
881 // exponential backoff covers the common transient cases
882 // without making a permanent failure (auth, oversize, etc.)
883 // wait forever.
884 let mut attempt: u32 = 0;
885 let resp = loop {
886 attempt += 1;
887 let body = aws_sdk_s3::primitives::ByteStream::from(part.clone());
888 match self
889 .client
890 .upload_part()
891 .bucket(&self.bucket)
892 .key(key)
893 .upload_id(upload_id)
894 .part_number(part_number)
895 .body(body)
896 .send()
897 .await
898 {
899 Ok(resp) => break Ok(resp),
900 Err(e) if attempt < 3 => {
901 // Backoff: 200ms, 800ms. Cheap enough not to
902 // mask a permanent failure; long enough that
903 // a brief network glitch resolves.
904 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
905 tracing::warn!(
906 part_number, attempt, delay_ms, error = ?e,
907 "S3 upload_part transient failure, retrying"
908 );
909 tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
910 }
911 Err(e) => break Err(e),
912 }
913 };
914
915 let resp = resp
916 .map_err(|e| format!("S3 upload part {part_number} failed after retries: {e}"))?;
917 let etag = resp.e_tag().unwrap_or_default().to_string();
918 completed_parts.push((part_number, etag));
919
920 part_number += 1;
921 }
922
923 if completed_parts.is_empty() {
924 return Err("No parts uploaded (empty file)".to_string());
925 }
926
927 // Shared completion path (with its own transient retry). The caller owns
928 // the single abort-on-Err, so a failure here still aborts exactly once.
929 self.complete_multipart_upload(key, upload_id, &completed_parts)
930 .await
931 }
932
933 /// Abort a multipart upload, releasing any uploaded parts.
934 ///
935 /// Public and `Result`-returning so the pending-upload reaper can abort an
936 /// orphaned session and retry on failure. Internal callers that only want
937 /// best-effort cleanup (e.g. [`Self::upload_multipart`] unwinding a failed
938 /// transfer) log the error and move on.
939 pub async fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> Result<(), String> {
940 self.client
941 .abort_multipart_upload()
942 .bucket(&self.bucket)
943 .key(key)
944 .upload_id(upload_id)
945 .send()
946 .await
947 .map(|_| ())
948 .map_err(|e| format!("S3 abort multipart upload for {key} failed: {e}"))
949 }
950
951 /// List the in-progress multipart uploads for exactly `key`, returning their
952 /// upload ids.
953 ///
954 /// The orphan reaper needs this: a session that was started but never
955 /// completed has **no object to delete**, only uploaded parts that S3 bills
956 /// for until they are aborted, so a plain `delete` is a no-op against it. The
957 /// upload id is not recorded durably anywhere, so it is recovered from S3,
958 /// which also catches sessions whose tracking row was lost entirely.
959 ///
960 /// `ListMultipartUploads` matches a *prefix*, so results are filtered to an
961 /// exact key match — otherwise reaping `staging/abc` would also abort a live
962 /// session for `staging/abcdef`.
963 pub async fn list_multipart_uploads_for_key(&self, key: &str) -> Result<Vec<String>, String> {
964 let mut ids = Vec::new();
965 let mut key_marker: Option<String> = None;
966 let mut upload_id_marker: Option<String> = None;
967
968 loop {
969 let mut req = self
970 .client
971 .list_multipart_uploads()
972 .bucket(&self.bucket)
973 .prefix(key);
974 if let Some(ref k) = key_marker {
975 req = req.key_marker(k);
976 }
977 if let Some(ref u) = upload_id_marker {
978 req = req.upload_id_marker(u);
979 }
980
981 let resp = req
982 .send()
983 .await
984 .map_err(|e| format!("S3 list_multipart_uploads for {key} failed: {e}"))?;
985
986 for upload in resp.uploads() {
987 // Exact-key filter: the request matched on prefix.
988 if upload.key() == Some(key)
989 && let Some(id) = upload.upload_id()
990 {
991 ids.push(id.to_string());
992 }
993 }
994
995 if resp.is_truncated().unwrap_or(false) {
996 key_marker = resp.next_key_marker().map(str::to_string);
997 upload_id_marker = resp.next_upload_id_marker().map(str::to_string);
998 // Defensive: a truncated response with no markers would loop forever.
999 if key_marker.is_none() && upload_id_marker.is_none() {
1000 break;
1001 }
1002 } else {
1003 break;
1004 }
1005 }
1006
1007 Ok(ids)
1008 }
1009
1010 /// Server-side copy an object using multipart `UploadPartCopy`, for sources
1011 /// larger than the 5 GiB single-part [`Self::copy_object`]/`CopyObject`
1012 /// limit. No bytes transit the caller — S3 copies each byte range internally.
1013 ///
1014 /// This is the >5 GiB promote path: scan-then-promote lifts a Clean staging
1015 /// object to the served content key, and a single `CopyObject` caps at 5 GiB,
1016 /// so a large video would otherwise succeed at upload and fail at promote —
1017 /// the worst failure position. `src_size` is the source object's size (the
1018 /// caller already reads it at confirm); `part_size` defaults to an
1019 /// auto-chosen size via [`MultipartPlan::auto`]. `content_type` sets the
1020 /// destination's type, since a fresh multipart upload does not inherit the
1021 /// source's metadata the way `CopyObject` does.
1022 ///
1023 /// Abort lives in one place, as in [`Self::upload_multipart`]: any failure
1024 /// after the destination upload is created aborts it exactly once.
1025 pub async fn copy_object_multipart(
1026 &self,
1027 src_bucket: &str,
1028 src_key: &str,
1029 dst_key: &str,
1030 content_type: &str,
1031 src_size: u64,
1032 part_size: Option<usize>,
1033 ) -> Result<(), String> {
1034 // Pre-flight: plan before anything is created, so a bad size strands
1035 // nothing and makes no network call.
1036 let plan = match part_size {
1037 Some(ps) => MultipartPlan::new(src_size, ps)?,
1038 None => MultipartPlan::auto(src_size)?,
1039 };
1040
1041 let upload_id = self.create_multipart_upload(dst_key, content_type).await?;
1042
1043 match self
1044 .run_multipart_copy(src_bucket, src_key, dst_key, &plan, &upload_id)
1045 .await
1046 {
1047 Ok(()) => Ok(()),
1048 Err(e) => {
1049 if let Err(abort_err) = self.abort_multipart_upload(dst_key, &upload_id).await {
1050 tracing::warn!("Failed to abort multipart copy for {dst_key}: {abort_err}");
1051 }
1052 Err(e)
1053 }
1054 }
1055 }
1056
1057 /// Issue every `UploadPartCopy` for a ranged multipart copy, then complete
1058 /// it. Returns `Err` (without aborting) on any failure; the caller
1059 /// [`Self::copy_object_multipart`] owns the single abort.
1060 async fn run_multipart_copy(
1061 &self,
1062 src_bucket: &str,
1063 src_key: &str,
1064 dst_key: &str,
1065 plan: &MultipartPlan,
1066 upload_id: &str,
1067 ) -> Result<(), String> {
1068 // `CopySource` is `{bucket}/{key}`; keys are sanitized to
1069 // `[A-Za-z0-9._/-]` upstream, so no percent-encoding is needed (same as
1070 // `copy_object_from`).
1071 let copy_source = format!("{src_bucket}/{src_key}");
1072 let mut completed_parts: Vec<(i32, String)> = Vec::with_capacity(plan.part_count as usize);
1073
1074 for part_number in 1..=plan.part_count {
1075 let (start, end) = plan
1076 .part_range(part_number)
1077 .ok_or_else(|| format!("internal: part {part_number} outside plan range"))?;
1078
1079 // Retry transient copy failures, mirroring the part-upload path.
1080 let mut attempt: u32 = 0;
1081 let resp = loop {
1082 attempt += 1;
1083 match self
1084 .client
1085 .upload_part_copy()
1086 .bucket(&self.bucket)
1087 .key(dst_key)
1088 .upload_id(upload_id)
1089 .part_number(part_number as i32)
1090 .copy_source(&copy_source)
1091 .copy_source_range(format!("bytes={start}-{end}"))
1092 .send()
1093 .await
1094 {
1095 Ok(r) => break Ok(r),
1096 Err(e) if attempt < 3 => {
1097 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
1098 tracing::warn!(
1099 part_number, attempt, delay_ms, error = ?e,
1100 "S3 upload_part_copy transient failure, retrying"
1101 );
1102 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
1103 }
1104 Err(e) => break Err(e),
1105 }
1106 };
1107
1108 let resp = resp.map_err(|e| {
1109 format!("S3 upload_part_copy part {part_number} failed after retries: {e}")
1110 })?;
1111 let etag = resp
1112 .copy_part_result()
1113 .and_then(|r| r.e_tag())
1114 .unwrap_or_default()
1115 .to_string();
1116 completed_parts.push((part_number as i32, etag));
1117 }
1118
1119 self.complete_multipart_upload(dst_key, upload_id, &completed_parts)
1120 .await
1121 }
1122
1123 /// Configure CORS on the bucket for browser uploads.
1124 pub async fn configure_cors(&self, allowed_origin: &str) {
1125 let origin = allowed_origin.trim_end_matches('/').to_string();
1126 let rule = match CorsRule::builder()
1127 .allowed_origins(&origin)
1128 .allowed_methods("PUT")
1129 .allowed_methods("GET")
1130 .allowed_methods("HEAD")
1131 .allowed_headers("Content-Type")
1132 .allowed_headers("Cache-Control")
1133 .allowed_headers("Content-Disposition")
1134 .expose_headers("ETag")
1135 .max_age_seconds(3600)
1136 .build()
1137 {
1138 Ok(r) => r,
1139 Err(e) => {
1140 tracing::warn!("Failed to build CORS rule: {}", e);
1141 return;
1142 }
1143 };
1144
1145 let cors_config = match CorsConfiguration::builder().cors_rules(rule).build() {
1146 Ok(c) => c,
1147 Err(e) => {
1148 tracing::warn!("Failed to build CORS config: {}", e);
1149 return;
1150 }
1151 };
1152
1153 match self
1154 .client
1155 .put_bucket_cors()
1156 .bucket(&self.bucket)
1157 .cors_configuration(cors_config)
1158 .send()
1159 .await
1160 {
1161 Ok(_) => tracing::info!("S3 bucket CORS configured for {}", origin),
1162 Err(e) => tracing::warn!("Failed to configure S3 CORS: {}", e),
1163 }
1164 }
1165
1166 /// Lightweight connectivity check — `list_objects_v2` with `max_keys(0)`.
1167 pub async fn check_connectivity(&self) -> Result<(), String> {
1168 self.client
1169 .list_objects_v2()
1170 .bucket(&self.bucket)
1171 .max_keys(0)
1172 .send()
1173 .await
1174 .map(|_| ())
1175 .map_err(|e| format!("{e}"))
1176 }
1177 }
1178
1179 #[cfg(test)]
1180 mod tests {
1181 use super::*;
1182
1183 fn test_client() -> S3Client {
1184 // `from_conf` is local — no network until a request is sent — so this
1185 // builds a usable client without reaching any endpoint.
1186 let s3_config = aws_sdk_s3::Config::builder()
1187 .behavior_version(BehaviorVersion::latest())
1188 .http_client(https_client())
1189 .region(Region::new("test"))
1190 .endpoint_url("http://127.0.0.1:1")
1191 .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
1192 .force_path_style(true)
1193 .build();
1194 S3Client {
1195 client: Client::from_conf(s3_config),
1196 bucket: "test-bucket".to_string(),
1197 }
1198 }
1199
1200 const MIB: u64 = 1024 * 1024;
1201
1202 #[test]
1203 fn multipart_plan_divides_with_remainder() {
1204 // 25 MiB in 10 MiB parts -> 10 + 10 + 5.
1205 let plan = MultipartPlan::new(25 * MIB, 10 * MIB as usize).unwrap();
1206 assert_eq!(plan.part_count, 3);
1207 assert_eq!(plan.part_len(1), 10 * MIB);
1208 assert_eq!(plan.part_len(2), 10 * MIB);
1209 assert_eq!(plan.part_len(3), 5 * MIB);
1210 assert_eq!(plan.part_len(4), 0, "out-of-range part");
1211 assert_eq!(plan.part_range(1), Some((0, 10 * MIB - 1)));
1212 assert_eq!(plan.part_range(3), Some((20 * MIB, 25 * MIB - 1)));
1213 assert_eq!(plan.part_range(4), None);
1214 }
1215
1216 #[test]
1217 fn multipart_plan_divides_evenly() {
1218 // 20 MiB in 5 MiB parts -> four full parts, last is a full part.
1219 let plan = MultipartPlan::new(20 * MIB, MULTIPART_MIN_PART_SIZE).unwrap();
1220 assert_eq!(plan.part_count, 4);
1221 assert_eq!(plan.part_len(4), 5 * MIB);
1222 assert_eq!(plan.part_range(4), Some((15 * MIB, 20 * MIB - 1)));
1223 }
1224
1225 #[test]
1226 fn multipart_plan_rejects_empty_object() {
1227 let err = MultipartPlan::new(0, MULTIPART_MIN_PART_SIZE).unwrap_err();
1228 assert!(err.contains("non-empty"), "unexpected error: {err}");
1229 }
1230
1231 #[test]
1232 fn multipart_plan_rejects_undersized_part() {
1233 let err = MultipartPlan::new(100 * MIB, MULTIPART_MIN_PART_SIZE - 1).unwrap_err();
1234 assert!(err.contains("5 MiB"), "unexpected error: {err}");
1235 }
1236
1237 #[test]
1238 fn multipart_plan_rejects_oversized_part() {
1239 let err = MultipartPlan::new(10 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1).unwrap_err();
1240 assert!(err.contains("5 GiB"), "unexpected error: {err}");
1241 }
1242
1243 #[test]
1244 fn multipart_plan_rejects_too_many_parts() {
1245 // One 5 MiB part past the 10k limit at the minimum part size.
1246 let total = MULTIPART_MIN_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 1);
1247 let err = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap_err();
1248 assert!(err.contains("10000-part"), "unexpected error: {err}");
1249 }
1250
1251 #[test]
1252 fn multipart_plan_accepts_exactly_max_parts() {
1253 let total = MULTIPART_MIN_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64;
1254 let plan = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap();
1255 assert_eq!(plan.part_count, MULTIPART_MAX_PARTS);
1256 }
1257
1258 #[test]
1259 fn multipart_plan_rejects_over_object_ceiling() {
1260 let err = MultipartPlan::new(
1261 MULTIPART_MAX_OBJECT_SIZE + 1,
1262 MULTIPART_MAX_PART_SIZE as usize,
1263 )
1264 .unwrap_err();
1265 assert!(err.contains("5 TiB"), "unexpected error: {err}");
1266 }
1267
1268 #[test]
1269 fn multipart_plan_auto_uses_default_for_small_objects() {
1270 let plan = MultipartPlan::auto(100 * MIB).unwrap();
1271 assert_eq!(plan.part_size, MULTIPART_DEFAULT_PART_SIZE);
1272 // 100 MiB / 16 MiB -> 7 parts (ceil).
1273 assert_eq!(plan.part_count, 7);
1274 }
1275
1276 #[test]
1277 fn multipart_plan_auto_scales_part_size_to_stay_within_part_cap() {
1278 // An object too big for the default part size within 10k parts must get
1279 // a larger part size, and the resulting plan must be valid.
1280 let big = MULTIPART_DEFAULT_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 500);
1281 let plan = MultipartPlan::auto(big).unwrap();
1282 assert!(plan.part_size > MULTIPART_DEFAULT_PART_SIZE);
1283 assert!(plan.part_count <= MULTIPART_MAX_PARTS);
1284 // Whole-MiB part size.
1285 assert_eq!(plan.part_size as u64 % MIB, 0);
1286 }
1287
1288 #[test]
1289 fn multipart_plan_auto_rejects_empty() {
1290 assert!(MultipartPlan::auto(0).is_err());
1291 }
1292
1293 #[tokio::test]
1294 async fn copy_object_multipart_rejects_empty_source_before_any_request() {
1295 // Plan is pre-flight: an empty source fails before the destination
1296 // multipart upload is created, so the unreachable endpoint is untouched.
1297 let client = test_client();
1298 let err = client
1299 .copy_object_multipart("bkt", "src", "dst", "application/octet-stream", 0, None)
1300 .await
1301 .expect_err("empty source must be rejected");
1302 assert!(err.contains("non-empty"), "unexpected error: {err}");
1303 }
1304
1305 /// The `X-Amz-SignedHeaders` list from a presigned URL.
1306 fn signed_headers(url: &str) -> String {
1307 url.split('&')
1308 .find_map(|p| p.strip_prefix("X-Amz-SignedHeaders="))
1309 .map(|v| v.replace("%3B", ";"))
1310 .expect("presigned URL must carry X-Amz-SignedHeaders")
1311 }
1312
1313 #[tokio::test]
1314 async fn presign_upload_signs_content_length_when_bound() {
1315 // Callers rely on `max_bytes` being enforced, and it is enforced only
1316 // because it lands in SignedHeaders: a client sending a different
1317 // Content-Length then fails the signature. That also makes the declared
1318 // size a hard contract — a caller that declares anything other than the
1319 // exact body length breaks every upload — so pin it here rather than
1320 // discovering it against production S3.
1321 let client = test_client();
1322
1323 let bound = client
1324 .presign_upload("k", "application/octet-stream", 900, None, Some(12_345))
1325 .await
1326 .unwrap();
1327 let headers = signed_headers(&bound);
1328 assert!(
1329 headers.contains("content-length"),
1330 "max_bytes must be signed, got: {headers}"
1331 );
1332
1333 let unbound = client
1334 .presign_upload("k", "application/octet-stream", 900, None, None)
1335 .await
1336 .unwrap();
1337 assert!(
1338 !signed_headers(&unbound).contains("content-length"),
1339 "without max_bytes the client is free to send any length"
1340 );
1341 }
1342
1343 #[tokio::test]
1344 async fn presign_upload_part_rejects_out_of_range_part_number() {
1345 // Pre-flight range check: fires before any network call, so the
1346 // unreachable dummy endpoint is never touched.
1347 let client = test_client();
1348 for bad in [0, MULTIPART_MAX_PARTS as i32 + 1] {
1349 let err = client
1350 .presign_upload_part("k", "uid", bad, 3600, None, None)
1351 .await
1352 .expect_err("out-of-range part number must be rejected");
1353 assert!(err.contains("out of range"), "unexpected error: {err}");
1354 }
1355 }
1356
1357 #[tokio::test]
1358 async fn presign_upload_part_signs_the_checksum_when_bound() {
1359 // S3 enforces a bound checksum by rehashing the part, but only if the
1360 // client sends the header — which it must, because signing it makes it
1361 // mandatory. Both halves of that live in SignedHeaders.
1362 let client = test_client();
1363
1364 let bound = client
1365 .presign_upload_part("k", "uid", 1, 900, Some(64), Some("Zm9vYmFyYmF6"))
1366 .await
1367 .unwrap();
1368 let headers = signed_headers(&bound);
1369 assert!(
1370 headers.contains("x-amz-checksum-sha256"),
1371 "a bound checksum must be signed, got: {headers}"
1372 );
1373
1374 let unbound = client
1375 .presign_upload_part("k", "uid", 1, 900, Some(64), None)
1376 .await
1377 .unwrap();
1378 assert!(
1379 !signed_headers(&unbound).contains("checksum"),
1380 "no checksum bound means no checksum header is required"
1381 );
1382 }
1383
1384 #[tokio::test]
1385 async fn complete_multipart_rejects_empty_parts() {
1386 let client = test_client();
1387 let err = client
1388 .complete_multipart_upload("k", "uid", &[])
1389 .await
1390 .expect_err("empty parts must be rejected");
1391 assert!(err.contains("no parts"), "unexpected error: {err}");
1392 }
1393
1394 #[tokio::test]
1395 async fn upload_multipart_rejects_undersized_part_before_any_request() {
1396 // The minimum-part-size guard is pre-flight: it must fire before the
1397 // multipart upload is created, so there is nothing to strand and no
1398 // network call (the dummy endpoint is unreachable — reaching it would
1399 // hang/error instead of returning this exact message).
1400 let client = test_client();
1401 let path = std::path::Path::new("/nonexistent");
1402 let err = client
1403 .upload_multipart("k", "application/octet-stream", path, Some(1024))
1404 .await
1405 .expect_err("undersized part size must be rejected");
1406 assert!(err.contains("at least 5 MB"), "unexpected error: {err}");
1407 }
1408 }
1409