Skip to main content

max / makenotwork

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