Skip to main content

max / makenotwork

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