Skip to main content

max / makenotwork

81.1 KB · 2049 lines History Blame Raw
1 //! Shared S3-compatible storage client.
2 //!
3 //! Overview: maintainer wiki.
4 //! <!-- wiki: s3-storage-overview -->
5 //!
6 //! Thin wrapper around the AWS SDK providing upload, download, delete,
7 //! presigned URL generation, and bucket management. Used by MNW and
8 //! Multithreaded to avoid duplicating S3 initialization and operations.
9
10 #![warn(missing_docs)]
11
12 use aws_config::BehaviorVersion;
13 use aws_sdk_s3::Client;
14 use aws_sdk_s3::config::{Credentials, Region};
15 use aws_sdk_s3::error::ProvideErrorMetadata;
16 use aws_sdk_s3::presigning::PresigningConfig;
17 use aws_sdk_s3::types::{
18 CompletedMultipartUpload, CompletedPart, CorsConfiguration, CorsRule, Delete, ObjectIdentifier,
19 };
20 use std::time::Duration;
21
22 pub use aws_sdk_s3::primitives::ByteStream;
23
24 /// The HTTPS client every S3 client is built on.
25 ///
26 /// Built here rather than taken from the SDK's `default-https-client` feature,
27 /// which is a hard alias for the aws-lc-rs (C) crypto backend. Ring is the
28 /// pure-Rust provider the rest of the tree uses, and the SDK will accept it only
29 /// through an explicitly constructed client.
30 fn https_client() -> aws_sdk_s3::config::SharedHttpClient {
31 aws_smithy_http_client::Builder::new()
32 .tls_provider(aws_smithy_http_client::tls::Provider::Rustls(
33 aws_smithy_http_client::tls::rustls_provider::CryptoMode::Ring,
34 ))
35 .build_https()
36 }
37
38 /// S3 connection configuration.
39 #[derive(Debug, Clone)]
40 pub struct S3Config {
41 /// Endpoint URL (e.g., `https://fsn1.your-objectstorage.com`)
42 pub endpoint: String,
43 /// Bucket name
44 pub bucket: String,
45 /// Access key ID
46 pub access_key: String,
47 /// Secret access key
48 pub secret_key: String,
49 /// Region (e.g., `fsn1`)
50 pub region: String,
51 }
52
53 /// One CORS rule as the bucket reports it, in this crate's own types.
54 ///
55 /// A caller that wants to know what CORS the bucket is set to has nowhere to
56 /// ask otherwise: [`S3Client::configure_cors`] returns `()` and the SDK's
57 /// `CorsRule` is not re-exported, so the setting was write-only from outside
58 /// this crate. Owning the shape here also keeps the SDK out of the callers'
59 /// signatures, which is the reason the wrapper exists at all.
60 #[derive(Debug, Clone, PartialEq, Eq, Default)]
61 pub struct CorsRuleView {
62 /// Origins the rule allows, verbatim as stored.
63 pub allowed_origins: Vec<String>,
64 /// Methods the rule allows.
65 pub allowed_methods: Vec<String>,
66 /// Request headers the rule allows.
67 pub allowed_headers: Vec<String>,
68 /// Response headers the rule exposes to the browser.
69 pub expose_headers: Vec<String>,
70 /// Preflight cache lifetime in seconds, if the rule sets one.
71 pub max_age_seconds: Option<i32>,
72 }
73
74 /// S3 client wrapper.
75 #[derive(Clone)]
76 pub struct S3Client {
77 client: Client,
78 bucket: String,
79 }
80
81 /// SigV4's hard maximum presign lifetime (7 days). A presign request above this
82 /// is rejected by the signer, so we clamp callers to it rather than surface an
83 /// opaque signing error — and it bounds how long any single minted URL can live.
84 const MAX_PRESIGN_EXPIRY_SECS: u64 = 7 * 24 * 60 * 60;
85
86 /// S3 multipart-upload limits, straight from the S3 API contract. Every layer
87 /// that plans a client-direct multipart upload (the blob and creator-media
88 /// session endpoints) validates against these, so the client is never handed a
89 /// plan S3 would reject at complete time.
90 ///
91 /// Minimum size of every part *except the last* (5 MiB). The final part may be
92 /// anything down to 1 byte.
93 pub const MULTIPART_MIN_PART_SIZE: usize = 5 * 1024 * 1024;
94 /// Maximum number of parts in a single multipart upload.
95 pub const MULTIPART_MAX_PARTS: u32 = 10_000;
96 /// Maximum size of a single part (5 GiB).
97 pub const MULTIPART_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
98 /// Maximum size of an object assembled from a multipart upload (5 TiB).
99 pub const MULTIPART_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024;
100 /// Default/floor part size for an auto-planned multipart upload (16 MiB): big
101 /// enough to keep the per-part round-trip overhead low, small enough that a
102 /// resumable client re-sends little on a retry. Used when the object is small
103 /// enough not to force larger parts to stay within the part-count limit.
104 pub const MULTIPART_DEFAULT_PART_SIZE: usize = 16 * 1024 * 1024;
105
106 // The part budget at max part size must be able to cover the object ceiling, or
107 // a large-but-legal object would be unplannable at any part size. 10,000 x 5 GiB
108 // ~= 48.8 TiB, comfortably over the 5 TiB object cap. (The 5 MiB *minimum* part
109 // deliberately does NOT cover 5 TiB — a large object simply needs bigger parts,
110 // which `MultipartPlan::new` enforces via the part-count limit.)
111 const _: () = assert!(
112 MULTIPART_MAX_PARTS as u128 * MULTIPART_MAX_PART_SIZE as u128
113 >= MULTIPART_MAX_OBJECT_SIZE as u128
114 );
115
116 /// A validated multipart-upload plan: the part size to use and how many parts a
117 /// body of `total_size` bytes splits into. Pure arithmetic with no S3 call, so
118 /// the blob and creator-media session endpoints share one source of truth for
119 /// part geometry (and the client computes the same boundaries independently).
120 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
121 pub struct MultipartPlan {
122 /// Total object size in bytes.
123 pub total_size: u64,
124 /// Size of every part except the last.
125 pub part_size: usize,
126 /// Number of parts (1..=[`MULTIPART_MAX_PARTS`]).
127 pub part_count: u32,
128 }
129
130 impl MultipartPlan {
131 /// Plan a multipart upload of `total_size` bytes into `part_size`-byte parts,
132 /// the last part taking the remainder.
133 ///
134 /// Errors if the object is empty (use a single PUT), exceeds the 5 TiB
135 /// multipart ceiling, the part size is below S3's 5 MiB floor or above its
136 /// 5 GiB ceiling, or the resulting part count would exceed 10,000 — the last
137 /// meaning the part size is too small for this object and the caller should
138 /// pick a larger one.
139 pub fn new(total_size: u64, part_size: usize) -> Result<Self, String> {
140 if total_size == 0 {
141 return Err(
142 "multipart upload needs a non-empty object; use a single PUT for empty objects"
143 .to_string(),
144 );
145 }
146 if total_size > MULTIPART_MAX_OBJECT_SIZE {
147 return Err(format!(
148 "object is {total_size} bytes, over the {MULTIPART_MAX_OBJECT_SIZE}-byte (5 TiB) multipart ceiling"
149 ));
150 }
151 if part_size < MULTIPART_MIN_PART_SIZE {
152 return Err(format!(
153 "part size {part_size} is below the {MULTIPART_MIN_PART_SIZE}-byte (5 MiB) S3 minimum"
154 ));
155 }
156 if part_size as u64 > MULTIPART_MAX_PART_SIZE {
157 return Err(format!(
158 "part size {part_size} is above the {MULTIPART_MAX_PART_SIZE}-byte (5 GiB) S3 maximum"
159 ));
160 }
161 let part_count = total_size.div_ceil(part_size as u64);
162 if part_count > MULTIPART_MAX_PARTS as u64 {
163 return Err(format!(
164 "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"
165 ));
166 }
167 Ok(Self {
168 total_size,
169 part_size,
170 part_count: part_count as u32,
171 })
172 }
173
174 /// Plan a multipart upload of `total_size` bytes, choosing the part size
175 /// automatically: [`MULTIPART_DEFAULT_PART_SIZE`] when that keeps the object
176 /// within [`MULTIPART_MAX_PARTS`] parts, otherwise the smallest whole-MiB
177 /// part size that does. Errors only if the object is empty or over the 5 TiB
178 /// ceiling — a valid non-empty object always yields a plan.
179 pub fn auto(total_size: u64) -> Result<Self, String> {
180 const MIB: u64 = 1024 * 1024;
181 // Smallest part size that fits the object within the part-count cap,
182 // rounded up to a whole MiB, then floored at the default part size.
183 let needed = total_size.div_ceil(MULTIPART_MAX_PARTS as u64);
184 let rounded = needed.div_ceil(MIB) * MIB;
185 let part_size = (rounded as usize).max(MULTIPART_DEFAULT_PART_SIZE);
186 Self::new(total_size, part_size)
187 }
188
189 /// Byte length of part `part_number` (1-based). The last part is the
190 /// remainder; every earlier part is exactly `part_size`. Returns 0 for a
191 /// part number outside `1..=part_count`.
192 pub fn part_len(&self, part_number: u32) -> u64 {
193 if part_number == 0 || part_number > self.part_count {
194 return 0;
195 }
196 if part_number < self.part_count {
197 return self.part_size as u64;
198 }
199 // Last part: the remainder, or a full part when the size divides evenly.
200 match self.total_size % self.part_size as u64 {
201 0 => self.part_size as u64,
202 rem => rem,
203 }
204 }
205
206 /// Inclusive `[start, end]` byte range of part `part_number` (1-based), the
207 /// form S3 `UploadPartCopy` wants as `bytes=start-end`. `None` for a part
208 /// number outside `1..=part_count`.
209 pub fn part_range(&self, part_number: u32) -> Option<(u64, u64)> {
210 if part_number == 0 || part_number > self.part_count {
211 return None;
212 }
213 let start = (part_number as u64 - 1) * self.part_size as u64;
214 Some((start, start + self.part_len(part_number) - 1))
215 }
216 }
217
218 /// What a [`MultipartPlan`] must be true of, whatever it was asked for.
219 ///
220 /// **The oracle lives in the crate**, so `tests/regressions.rs` on stable and
221 /// the soak tier's libFuzzer target on nightly assert the same things and
222 /// cannot drift into checking less than each other. The shape is
223 /// `MNW/shared/git-command`'s, copied deliberately.
224 ///
225 /// ## The doors, counted before this was written
226 ///
227 /// Infra `8910f917` instructs it, and the answer decided the shape. Almost all
228 /// of this crate is async I/O against a live endpoint and is not fuzzable
229 /// without one. What IS pure, and what creator media actually depends on, is
230 /// the part geometry: [`MultipartPlan::auto`] has four call sites, all in MNW
231 /// server (`routes/synckit/blobs.rs` twice, `routes/api/internal/uploads.rs`
232 /// twice), and one implementation behind them.
233 ///
234 /// `MultipartPlan`'s own docs say "the client computes the same boundaries
235 /// independently", which reads like the two-parsers-of-one-grammar shape that
236 /// made a shared crate the right answer for `git_ssh`. It is not. SyncKit's
237 /// client check (`synckit-client/src/client/blob.rs`) is a one-line restatement
238 /// -- `size_bytes.div_ceil(part_size) == part_count` -- so a differential
239 /// between the two would measure nothing that asserting the property does not.
240 /// That restatement is property (1) below, asserted against the plan itself.
241 ///
242 /// ## Why the properties are the ones they are
243 ///
244 /// This is the path creator media travels, and it carries two distinct failure
245 /// modes. A plan whose parts do not tile the object exactly means **corrupted
246 /// or truncated media** -- a gap loses bytes, an overlap duplicates them, and
247 /// neither shows up until someone plays the file back. A plan that is refused
248 /// when it should not be means a **legitimate upload rejected**, which is why
249 /// the error paths are asserted as tightly as the success path: every rejection
250 /// has to name a condition that actually holds.
251 pub mod oracle {
252 use super::{
253 MULTIPART_MAX_OBJECT_SIZE, MULTIPART_MAX_PART_SIZE, MULTIPART_MAX_PARTS,
254 MULTIPART_MIN_PART_SIZE, MultipartPlan,
255 };
256
257 /// Assert everything that must hold for one `(total_size, part_size)` pair.
258 ///
259 /// # Panics
260 /// By design. It is an oracle, and a panic is how it reports.
261 pub fn check_plan(total_size: u64, part_size: usize) {
262 let outcome = MultipartPlan::new(total_size, part_size);
263
264 // Whether a plan is possible at all, stated independently of the code
265 // under test rather than read back out of it.
266 let possible = total_size > 0
267 && total_size <= MULTIPART_MAX_OBJECT_SIZE
268 && part_size >= MULTIPART_MIN_PART_SIZE
269 && part_size as u64 <= MULTIPART_MAX_PART_SIZE
270 && total_size.div_ceil(part_size as u64) <= MULTIPART_MAX_PARTS as u64;
271
272 match outcome {
273 Err(_) => assert!(
274 !possible,
275 "a legitimate plan was refused: total_size {total_size}, part_size {part_size}"
276 ),
277 Ok(plan) => {
278 assert!(
279 possible,
280 "an impossible plan was accepted: total_size {total_size}, \
281 part_size {part_size} -> {plan:?}"
282 );
283 check_geometry(&plan);
284 }
285 }
286 }
287
288 /// Assert what [`MultipartPlan::auto`] promises: a plan for every non-empty
289 /// object within the ceiling, with no part size for the caller to pick.
290 ///
291 /// # Panics
292 /// By design.
293 pub fn check_auto(total_size: u64) {
294 let outcome = MultipartPlan::auto(total_size);
295 let possible = total_size > 0 && total_size <= MULTIPART_MAX_OBJECT_SIZE;
296
297 match outcome {
298 Err(_) => assert!(
299 !possible,
300 "auto refused a legitimate object of {total_size} bytes"
301 ),
302 Ok(plan) => {
303 assert!(
304 possible,
305 "auto planned an impossible object of {total_size} bytes: {plan:?}"
306 );
307 check_geometry(&plan);
308 }
309 }
310 }
311
312 /// The geometry every accepted plan must satisfy.
313 ///
314 /// # Panics
315 /// By design.
316 pub(crate) fn check_geometry(plan: &MultipartPlan) {
317 let MultipartPlan {
318 total_size,
319 part_size,
320 part_count,
321 } = *plan;
322
323 // (1) The client's independent check, asserted here so the two cannot
324 // disagree. SyncKit refuses a session whose part_count does not equal
325 // this, so a plan that fails it is an upload that cannot start.
326 assert_eq!(
327 u64::from(part_count),
328 total_size.div_ceil(part_size as u64),
329 "part_count disagrees with div_ceil for {plan:?}"
330 );
331
332 // (2) S3's own limits, restated because they are what the remote will
333 // enforce whatever we believe.
334 assert!(part_count >= 1, "a plan with no parts: {plan:?}");
335 assert!(
336 part_count <= MULTIPART_MAX_PARTS,
337 "over the part limit: {plan:?}"
338 );
339 assert!(
340 part_size >= MULTIPART_MIN_PART_SIZE,
341 "part below the S3 floor: {plan:?}"
342 );
343 assert!(
344 part_size as u64 <= MULTIPART_MAX_PART_SIZE,
345 "part above the S3 ceiling: {plan:?}"
346 );
347
348 // (3) THE PARTS TILE THE OBJECT EXACTLY. This is the corruption
349 // property: a gap loses bytes and an overlap duplicates them, and
350 // neither is visible until someone plays the file back.
351 let mut covered: u64 = 0;
352 let mut expected_start: u64 = 0;
353 for n in 1..=part_count {
354 let len = plan.part_len(n);
355 assert!(len > 0, "part {n} is empty in {plan:?}");
356 assert!(
357 len <= part_size as u64,
358 "part {n} is longer than the part size in {plan:?}"
359 );
360
361 let (start, end) = plan
362 .part_range(n)
363 .unwrap_or_else(|| panic!("part {n} has no range in {plan:?}"));
364 assert_eq!(
365 start,
366 expected_start,
367 "part {n} does not abut part {} in {plan:?}",
368 n - 1
369 );
370 assert_eq!(
371 end - start + 1,
372 len,
373 "part {n}'s range and length disagree in {plan:?}"
374 );
375
376 covered += len;
377 expected_start = end + 1;
378 }
379 assert_eq!(
380 covered, total_size,
381 "the parts do not cover the object exactly: {plan:?}"
382 );
383 assert_eq!(
384 expected_start, total_size,
385 "the last part does not end at the object's end: {plan:?}"
386 );
387
388 // (4) Out-of-range part numbers are refused rather than answered. A
389 // caller that loops off the end must get nothing, not byte zero.
390 assert_eq!(plan.part_len(0), 0, "part 0 has a length in {plan:?}");
391 assert_eq!(plan.part_range(0), None, "part 0 has a range in {plan:?}");
392 let past = part_count + 1;
393 assert_eq!(
394 plan.part_len(past),
395 0,
396 "part {past} has a length in {plan:?}"
397 );
398 assert_eq!(
399 plan.part_range(past),
400 None,
401 "part {past} has a range in {plan:?}"
402 );
403 }
404 }
405
406 impl S3Client {
407 /// Create a new S3 client from configuration.
408 // Public async constructor: kept async for API stability across callers.
409 #[allow(clippy::unused_async)]
410 pub async fn new(config: &S3Config) -> Result<Self, String> {
411 let credentials = Credentials::new(
412 &config.access_key,
413 &config.secret_key,
414 None,
415 None,
416 "s3-storage",
417 );
418
419 // Bound every S3 operation so a hung endpoint (e.g. a stalled
420 // HeadObject on a blob-confirm path) can't wedge a caller forever. These
421 // apply to establishing the connection and to a single attempt's
422 // round-trip to first byte of the response — they do NOT cap the time
423 // spent streaming a large object body, so big uploads/downloads are
424 // unaffected. The SDK's default retry policy still applies per attempt.
425 let timeout_config = aws_sdk_s3::config::timeout::TimeoutConfig::builder()
426 .connect_timeout(Duration::from_secs(10))
427 .operation_attempt_timeout(Duration::from_mins(1))
428 .build();
429
430 let s3_config = aws_sdk_s3::Config::builder()
431 .behavior_version(BehaviorVersion::latest())
432 .http_client(https_client())
433 .region(Region::new(config.region.clone()))
434 .endpoint_url(&config.endpoint)
435 .credentials_provider(credentials)
436 .timeout_config(timeout_config)
437 .force_path_style(true)
438 .build();
439
440 let client = Client::from_conf(s3_config);
441
442 Ok(Self {
443 client,
444 bucket: config.bucket.clone(),
445 })
446 }
447
448 /// Bucket name accessor.
449 pub fn bucket(&self) -> &str {
450 &self.bucket
451 }
452
453 /// Upload bytes to S3.
454 pub async fn upload(
455 &self,
456 key: &str,
457 content_type: &str,
458 data: Vec<u8>,
459 cache_control: Option<&str>,
460 ) -> Result<(), String> {
461 let mut req = self
462 .client
463 .put_object()
464 .bucket(&self.bucket)
465 .key(key)
466 .content_type(content_type)
467 .body(data.into());
468
469 if let Some(cc) = cache_control {
470 req = req.cache_control(cc);
471 }
472
473 req.send()
474 .await
475 .map_err(|e| format!("S3 upload failed: {e}"))?;
476
477 Ok(())
478 }
479
480 /// Download bytes from S3. Returns `(data, content_type)`.
481 ///
482 /// Convenience `Vec<u8>` form; for the zero-extra-copy path use
483 /// [`download_buf`](Self::download_buf), which returns the aggregated
484 /// `Bytes` directly.
485 pub async fn download(&self, key: &str) -> Result<(Vec<u8>, String), String> {
486 let (bytes, content_type) = self.download_buf(key).await?;
487 Ok((bytes.to_vec(), content_type))
488 }
489
490 /// Download an object as `bytes::Bytes`, returning `(data, content_type)`.
491 ///
492 /// Unlike [`download`](Self::download) this does not copy the aggregated
493 /// body into a fresh `Vec` — the caller gets the SDK's buffer directly. Use
494 /// it on memory-sensitive paths (e.g. the scanner's buffered branch) where
495 /// the extra `to_vec` would transiently double the footprint.
496 pub async fn download_buf(&self, key: &str) -> Result<(bytes::Bytes, String), String> {
497 let resp = self
498 .client
499 .get_object()
500 .bucket(&self.bucket)
501 .key(key)
502 .send()
503 .await
504 .map_err(|e| format!("S3 download failed: {e}"))?;
505
506 let content_type = resp
507 .content_type()
508 .unwrap_or("application/octet-stream")
509 .to_string();
510
511 let bytes = resp
512 .body
513 .collect()
514 .await
515 .map_err(|e| format!("S3 read body failed: {e}"))?;
516
517 Ok((bytes.into_bytes(), content_type))
518 }
519
520 /// Stream an object's body from S3 without buffering. Caller drives the
521 /// `ByteStream` to disk or hands it to a layer that wants chunks.
522 pub async fn download_stream(
523 &self,
524 key: &str,
525 ) -> Result<aws_sdk_s3::primitives::ByteStream, String> {
526 let resp = self
527 .client
528 .get_object()
529 .bucket(&self.bucket)
530 .key(key)
531 .send()
532 .await
533 .map_err(|e| format!("S3 download failed: {e}"))?;
534
535 Ok(resp.body)
536 }
537
538 /// Download only the first `len` bytes of an object via a ranged
539 /// `GetObject` (`Range: bytes=0-{len-1}`). Used for content sniffing so a
540 /// 4 KB header read doesn't initiate a transfer of the whole object. Returns
541 /// fewer bytes if the object is smaller than `len`.
542 pub async fn download_head(&self, key: &str, len: usize) -> Result<Vec<u8>, String> {
543 if len == 0 {
544 return Ok(Vec::new());
545 }
546 let resp = self
547 .client
548 .get_object()
549 .bucket(&self.bucket)
550 .key(key)
551 .range(format!("bytes=0-{}", len - 1))
552 .send()
553 .await
554 .map_err(|e| format!("S3 ranged download failed: {e}"))?;
555 let data = resp
556 .body
557 .collect()
558 .await
559 .map_err(|e| format!("S3 ranged body read failed: {e}"))?;
560 Ok(data.to_vec())
561 }
562
563 /// Delete an object from S3.
564 pub async fn delete(&self, key: &str) -> Result<(), String> {
565 self.client
566 .delete_object()
567 .bucket(&self.bucket)
568 .key(key)
569 .send()
570 .await
571 .map_err(|e| format!("S3 delete failed: {e}"))?;
572
573 Ok(())
574 }
575
576 /// Server-side copy an object from `src_key` to `dst_key` within this
577 /// bucket. No bytes transit the caller — S3 performs the copy internally.
578 ///
579 /// This is the primitive behind scan-then-promote: an upload is scanned at a
580 /// staging key the client can overwrite, and only a Clean object is copied to
581 /// the served key (which the client holds no presign for), so the served
582 /// bytes are provably the scanned bytes (ultra-fuzz Run #24 Storage HIGH).
583 ///
584 /// `CopySource` is `{bucket}/{key}`. Keys in this system are sanitized to
585 /// `[A-Za-z0-9._/-]` (see the server's `sanitize_filename`), none of which
586 /// require percent-encoding, so the source is formed directly.
587 pub async fn copy_object(&self, src_key: &str, dst_key: &str) -> Result<(), String> {
588 self.copy_object_from(&self.bucket, src_key, dst_key).await
589 }
590
591 /// Server-side copy from an arbitrary source bucket into THIS client's
592 /// bucket. Used by the scan-then-promote path to lift a Clean object from
593 /// the private staging bucket into the public (CDN-served) bucket in one
594 /// server-side operation (no bytes transit the process). The credentials
595 /// this client holds must have read on `src_bucket`; in this deployment all
596 /// buckets share one Hetzner project/key, so cross-bucket copy is permitted.
597 ///
598 /// `CopySource` is `{src_bucket}/{src_key}`. Keys are sanitized to
599 /// `[A-Za-z0-9._/-]` (server `sanitize_filename`), so no percent-encoding is
600 /// needed and the source is formed directly. When `src_bucket` equals this
601 /// client's bucket the copy is the ordinary same-bucket promote.
602 pub async fn copy_object_from(
603 &self,
604 src_bucket: &str,
605 src_key: &str,
606 dst_key: &str,
607 ) -> Result<(), String> {
608 self.client
609 .copy_object()
610 .bucket(&self.bucket)
611 .copy_source(format!("{src_bucket}/{src_key}"))
612 .key(dst_key)
613 .send()
614 .await
615 .map_err(|e| {
616 format!(
617 "S3 copy_object {src_bucket}/{src_key} -> {}/{dst_key} failed: {e}",
618 self.bucket
619 )
620 })?;
621 Ok(())
622 }
623
624 /// Delete a batch of objects in a single S3 `DeleteObjects` request.
625 ///
626 /// S3 accepts up to 1000 keys per call; the caller is responsible for
627 /// chunking. Returns the keys that failed (if any) along with their
628 /// per-object error message. A successful response with `Errors` is
629 /// not a hard error — partial success is normal for batched deletes.
630 pub async fn delete_objects(&self, keys: &[String]) -> Result<Vec<(String, String)>, String> {
631 if keys.is_empty() {
632 return Ok(Vec::new());
633 }
634 // A key that won't build an ObjectIdentifier is reported as a failure
635 // rather than silently dropped — otherwise the caller believes it was
636 // deleted and the object leaks forever (ultra-fuzz Run 11 Storage MED).
637 let mut failures: Vec<(String, String)> = Vec::new();
638 let mut objects: Vec<ObjectIdentifier> = Vec::with_capacity(keys.len());
639 for k in keys {
640 match ObjectIdentifier::builder().key(k).build() {
641 Ok(o) => objects.push(o),
642 Err(e) => failures.push((k.clone(), format!("malformed key: {e}"))),
643 }
644 }
645 if objects.is_empty() {
646 return Ok(failures);
647 }
648 let delete = Delete::builder()
649 .set_objects(Some(objects))
650 .quiet(true)
651 .build()
652 .map_err(|e| format!("S3 delete_objects build failed: {e}"))?;
653 let resp = self
654 .client
655 .delete_objects()
656 .bucket(&self.bucket)
657 .delete(delete)
658 .send()
659 .await
660 .map_err(|e| format!("S3 delete_objects failed: {e}"))?;
661 failures.extend(
662 resp.errors
663 .unwrap_or_default()
664 .into_iter()
665 .filter_map(|err| {
666 let key = err.key?;
667 let msg = err.message.unwrap_or_else(|| "<no message>".into());
668 Some((key, msg))
669 }),
670 );
671 Ok(failures)
672 }
673
674 /// Delete all objects under a given key prefix.
675 ///
676 /// Lists in pages of 1000 and deletes each page in a single batched
677 /// `DeleteObjects` call (S3's max). For a 50k-key prefix this is 50
678 /// round-trips instead of 50,000.
679 pub async fn delete_prefix(&self, prefix: &str) -> Result<(), String> {
680 let mut continuation_token: Option<String> = None;
681 loop {
682 let mut req = self
683 .client
684 .list_objects_v2()
685 .bucket(&self.bucket)
686 .prefix(prefix)
687 .max_keys(1000);
688 if let Some(ref token) = continuation_token {
689 req = req.continuation_token(token);
690 }
691 let resp = req
692 .send()
693 .await
694 .map_err(|e| format!("S3 list objects failed: {e}"))?;
695
696 let keys: Vec<String> = resp
697 .contents
698 .unwrap_or_default()
699 .into_iter()
700 .filter_map(|obj| obj.key)
701 .collect();
702
703 if !keys.is_empty() {
704 let failures = self.delete_objects(&keys).await?;
705 if !failures.is_empty() {
706 // Surface per-key failures to the caller (and abort further
707 // pages): the durable deletion queue is the convergence
708 // backstop, but a silent partial wipe would hide a bucket
709 // policy / permissions problem. Preview the first few keys.
710 let preview: Vec<String> = failures
711 .iter()
712 .take(5)
713 .map(|(k, e)| format!("{k}: {e}"))
714 .collect();
715 return Err(format!(
716 "S3 delete_prefix partial failure: {} keys failed (first 5: {})",
717 failures.len(),
718 preview.join(", ")
719 ));
720 }
721 }
722
723 if resp.is_truncated.unwrap_or(false) {
724 continuation_token = resp.next_continuation_token;
725 } else {
726 break;
727 }
728 }
729 Ok(())
730 }
731
732 /// Check if an object exists in S3.
733 pub async fn object_exists(&self, key: &str) -> Result<bool, String> {
734 match self
735 .client
736 .head_object()
737 .bucket(&self.bucket)
738 .key(key)
739 .send()
740 .await
741 {
742 Ok(_) => Ok(true),
743 Err(e) => {
744 let service_error = e.into_service_error();
745 if service_error.is_not_found() {
746 Ok(false)
747 } else {
748 Err(format!("S3 head_object failed: {service_error}"))
749 }
750 }
751 }
752 }
753
754 /// Get the size of an object in bytes, or `None` if not found.
755 pub async fn object_size(&self, key: &str) -> Result<Option<i64>, String> {
756 match self
757 .client
758 .head_object()
759 .bucket(&self.bucket)
760 .key(key)
761 .send()
762 .await
763 {
764 Ok(resp) => Ok(resp.content_length()),
765 Err(e) => {
766 let service_error = e.into_service_error();
767 if service_error.is_not_found() {
768 Ok(None)
769 } else {
770 Err(format!("S3 head_object failed: {service_error}"))
771 }
772 }
773 }
774 }
775
776 /// Generate a presigned URL for uploading.
777 ///
778 /// When `max_bytes` is set, the value is signed into the request as
779 /// `Content-Length` (a SignedHeader). A client that sends a *different*
780 /// `Content-Length` than the one signed produces a signature mismatch, so
781 /// the common "lie about the size" case fails. This is NOT a hard,
782 /// server-enforced ceiling, though: it relies on the client sending an
783 /// honest `Content-Length`, and S3-compatible backends (Ceph/MinIO/Garage)
784 /// vary in how strictly they reconcile the declared length with the actual
785 /// body. Treat it as defense-in-depth, not the boundary — the authoritative
786 /// per-file cap is enforced at confirm time, where the handler reads the
787 /// object's real size via [`Self::object_size`] before crediting storage.
788 /// (A presigned POST with a `content-length-range` policy would be a true
789 /// server-side range check; it is not used here because every upload path is
790 /// a single PUT.)
791 pub async fn presign_upload(
792 &self,
793 key: &str,
794 content_type: &str,
795 expiry_secs: u64,
796 cache_control: Option<&str>,
797 max_bytes: Option<i64>,
798 ) -> Result<String, String> {
799 let presigning_config = PresigningConfig::builder()
800 .expires_in(Duration::from_secs(
801 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
802 ))
803 .build()
804 .map_err(|e| format!("Presigning config error: {e}"))?;
805
806 let mut req = self
807 .client
808 .put_object()
809 .bucket(&self.bucket)
810 .key(key)
811 .content_type(content_type);
812
813 if let Some(cc) = cache_control {
814 req = req.cache_control(cc);
815 }
816
817 if let Some(n) = max_bytes {
818 req = req.content_length(n);
819 }
820
821 let presigned = req
822 .presigned(presigning_config)
823 .await
824 .map_err(|e| format!("Failed to generate upload URL: {e}"))?;
825
826 Ok(presigned.uri().to_string())
827 }
828
829 /// Generate a presigned URL for downloading.
830 pub async fn presign_download(&self, key: &str, expiry_secs: u64) -> Result<String, String> {
831 let presigning_config = PresigningConfig::builder()
832 .expires_in(Duration::from_secs(
833 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
834 ))
835 .build()
836 .map_err(|e| format!("Presigning config error: {e}"))?;
837
838 let presigned = self
839 .client
840 .get_object()
841 .bucket(&self.bucket)
842 .key(key)
843 .presigned(presigning_config)
844 .await
845 .map_err(|e| format!("Failed to generate download URL: {e}"))?;
846
847 Ok(presigned.uri().to_string())
848 }
849
850 /// Upload a file to S3 using multipart upload.
851 ///
852 /// Reads the file in `part_size` chunks (default 10 MB, minimum 5 MB) and
853 /// uploads each as a part. Aborts the multipart upload on any failure.
854 ///
855 /// The abort lives in exactly ONE place: once the multipart upload has been
856 /// created, all the fallible part/complete work runs in
857 /// the private `run_multipart_upload`, and this wrapper aborts on any `Err` it
858 /// returns. So a future failure path added inside the inner method aborts by
859 /// construction — it cannot forget to (the Sto-S1 fix, made structural rather
860 /// than per-branch).
861 pub async fn upload_multipart(
862 &self,
863 key: &str,
864 content_type: &str,
865 file_path: &std::path::Path,
866 part_size: Option<usize>,
867 ) -> Result<(), String> {
868 let part_size = part_size.unwrap_or(10 * 1024 * 1024); // 10 MB default
869 if part_size < 5 * 1024 * 1024 {
870 // Pre-flight: nothing created yet, nothing to abort.
871 return Err("Multipart part size must be at least 5 MB".to_string());
872 }
873
874 let upload_id = self.create_multipart_upload(key, content_type).await?;
875
876 // Single abort site: any failure past this point aborts exactly once.
877 match self
878 .run_multipart_upload(key, file_path, part_size, &upload_id)
879 .await
880 {
881 Ok(()) => Ok(()),
882 Err(e) => {
883 // Best-effort abort; the create succeeded, so on failure surface
884 // the original error and just log if the cleanup also fails.
885 if let Err(abort_err) = self.abort_multipart_upload(key, &upload_id).await {
886 tracing::warn!("Failed to abort multipart upload for {key}: {abort_err}");
887 }
888 Err(e)
889 }
890 }
891 }
892
893 /// Begin a client-direct multipart upload, returning the `upload_id` that
894 /// [`Self::presign_upload_part`], [`Self::complete_multipart_upload`], and
895 /// [`Self::abort_multipart_upload`] key on.
896 ///
897 /// The counterpart to [`Self::upload_multipart`], which drives the whole
898 /// transfer server-side from a local file. Here the server only mints the id
899 /// and (via `presign_upload_part`) the per-part URLs; the client streams the
900 /// parts straight to S3, so no object bytes transit the server.
901 pub async fn create_multipart_upload(
902 &self,
903 key: &str,
904 content_type: &str,
905 ) -> Result<String, String> {
906 let create = self
907 .client
908 .create_multipart_upload()
909 .bucket(&self.bucket)
910 .key(key)
911 .content_type(content_type)
912 .send()
913 .await
914 .map_err(|e| format!("S3 create multipart upload failed: {e}"))?;
915
916 create
917 .upload_id()
918 .map(str::to_string)
919 .ok_or_else(|| "S3 create multipart upload returned no upload_id".to_string())
920 }
921
922 /// Presign an `UploadPart` request for one part of an in-progress multipart
923 /// upload. `part_number` is 1-based (`1..=`[`MULTIPART_MAX_PARTS`]).
924 ///
925 /// When `max_bytes` is set it is signed as `Content-Length`, the same
926 /// defense-in-depth (not a hard, server-enforced ceiling) as
927 /// [`Self::presign_upload`] — see its docs for why the authoritative size
928 /// check still lives at confirm time.
929 ///
930 /// `checksum_sha256` (base64 of the raw 32-byte digest) is signed as
931 /// `x-amz-checksum-sha256`, and unlike the length this one S3 *does*
932 /// enforce: it hashes the received part and rejects a mismatch with
933 /// `BadDigest` before the bytes are durable. The caller must therefore send
934 /// the header — it is in `SignedHeaders`, so omitting it fails the
935 /// signature. This is transport integrity (the bytes S3 wrote are the bytes
936 /// the client hashed), not a statement about what those bytes mean.
937 pub async fn presign_upload_part(
938 &self,
939 key: &str,
940 upload_id: &str,
941 part_number: i32,
942 expiry_secs: u64,
943 max_bytes: Option<i64>,
944 checksum_sha256: Option<&str>,
945 ) -> Result<String, String> {
946 if !(1..=MULTIPART_MAX_PARTS as i32).contains(&part_number) {
947 return Err(format!(
948 "part number {part_number} out of range 1..={MULTIPART_MAX_PARTS}"
949 ));
950 }
951 let presigning_config = PresigningConfig::builder()
952 .expires_in(Duration::from_secs(
953 expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS),
954 ))
955 .build()
956 .map_err(|e| format!("Presigning config error: {e}"))?;
957
958 let mut req = self
959 .client
960 .upload_part()
961 .bucket(&self.bucket)
962 .key(key)
963 .upload_id(upload_id)
964 .part_number(part_number);
965
966 if let Some(n) = max_bytes {
967 req = req.content_length(n);
968 }
969 if let Some(c) = checksum_sha256 {
970 req = req.checksum_sha256(c);
971 }
972
973 let presigned = req
974 .presigned(presigning_config)
975 .await
976 .map_err(|e| format!("Failed to generate upload part URL: {e}"))?;
977
978 Ok(presigned.uri().to_string())
979 }
980
981 /// Complete a multipart upload from the collected `(part_number, etag)`
982 /// pairs. Parts are sorted by number before assembly (S3 requires ascending
983 /// order), so the caller may pass them in completion order.
984 ///
985 /// Retries the completion on transient failure: by this point every part is
986 /// uploaded and paid for, so losing the completion call would orphan the
987 /// whole upload and force a restart from byte 0 (ultra-fuzz Run 11 Storage
988 /// HIGH). The shared completion path for both the server-driven
989 /// [`Self::upload_multipart`] and the client-direct session flow.
990 pub async fn complete_multipart_upload(
991 &self,
992 key: &str,
993 upload_id: &str,
994 parts: &[(i32, String)],
995 ) -> Result<(), String> {
996 if parts.is_empty() {
997 return Err("cannot complete a multipart upload with no parts".to_string());
998 }
999 let mut parts = parts.to_vec();
1000 parts.sort_by_key(|(n, _)| *n);
1001 let completed_parts: Vec<CompletedPart> = parts
1002 .into_iter()
1003 .map(|(n, etag)| CompletedPart::builder().e_tag(etag).part_number(n).build())
1004 .collect();
1005
1006 let completed = CompletedMultipartUpload::builder()
1007 .set_parts(Some(completed_parts))
1008 .build();
1009
1010 let mut attempt: u32 = 0;
1011 loop {
1012 attempt += 1;
1013 match self
1014 .client
1015 .complete_multipart_upload()
1016 .bucket(&self.bucket)
1017 .key(key)
1018 .upload_id(upload_id)
1019 .multipart_upload(completed.clone())
1020 .send()
1021 .await
1022 {
1023 Ok(_) => return Ok(()),
1024 Err(e) if attempt < 3 => {
1025 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
1026 tracing::warn!(
1027 attempt, delay_ms, error = ?e,
1028 "S3 complete_multipart_upload transient failure, retrying"
1029 );
1030 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
1031 }
1032 Err(e) => {
1033 return Err(format!(
1034 "S3 complete multipart upload failed after retries: {e}"
1035 ));
1036 }
1037 }
1038 }
1039 }
1040
1041 /// Read the file and upload every part, then complete the multipart upload.
1042 ///
1043 /// Returns `Err` (without aborting) on any failure; the caller
1044 /// [`Self::upload_multipart`] owns the single abort. Keep all abort handling
1045 /// out of here so the "abort on failure" contract can't be partially applied.
1046 async fn run_multipart_upload(
1047 &self,
1048 key: &str,
1049 file_path: &std::path::Path,
1050 part_size: usize,
1051 upload_id: &str,
1052 ) -> Result<(), String> {
1053 use tokio::io::AsyncReadExt;
1054
1055 let mut file = tokio::fs::File::open(file_path)
1056 .await
1057 .map_err(|e| format!("Failed to open file for multipart upload: {e}"))?;
1058
1059 let mut part_number: i32 = 1;
1060 let mut completed_parts: Vec<(i32, String)> = Vec::new();
1061
1062 loop {
1063 // One owned buffer per part, frozen into `Bytes` so each retry attempt
1064 // clones a refcount instead of re-copying the part. Previously the
1065 // body was `buf[..n].to_vec()` *inside* the retry loop, so every part
1066 // (and every retry) paid a full part_size heap copy on top of the
1067 // resident staging buffer — ~2x part_size live (ultra-fuzz Run 11
1068 // Storage HIGH). `Bytes::from(Vec)` takes ownership without copying.
1069 let mut buf = vec![0u8; part_size];
1070 let mut bytes_read = 0;
1071 // Fill the buffer completely (or until EOF)
1072 while bytes_read < part_size {
1073 match file.read(&mut buf[bytes_read..]).await {
1074 Ok(0) => break,
1075 Ok(n) => bytes_read += n,
1076 Err(e) => return Err(format!("Failed to read file: {e}")),
1077 }
1078 }
1079
1080 if bytes_read == 0 {
1081 break;
1082 }
1083
1084 buf.truncate(bytes_read);
1085 let part: bytes::Bytes = buf.into();
1086
1087 // Retry the part upload up to 3 times on transient failures.
1088 // S3 part uploads can flake on network blips; aborting the
1089 // whole multipart upload because of one timeout means the
1090 // caller has to restart from byte 0. Three attempts with
1091 // exponential backoff covers the common transient cases
1092 // without making a permanent failure (auth, oversize, etc.)
1093 // wait forever.
1094 let mut attempt: u32 = 0;
1095 let resp = loop {
1096 attempt += 1;
1097 let body = aws_sdk_s3::primitives::ByteStream::from(part.clone());
1098 match self
1099 .client
1100 .upload_part()
1101 .bucket(&self.bucket)
1102 .key(key)
1103 .upload_id(upload_id)
1104 .part_number(part_number)
1105 .body(body)
1106 .send()
1107 .await
1108 {
1109 Ok(resp) => break Ok(resp),
1110 Err(e) if attempt < 3 => {
1111 // Backoff: 200ms, 800ms. Cheap enough not to
1112 // mask a permanent failure; long enough that
1113 // a brief network glitch resolves.
1114 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
1115 tracing::warn!(
1116 part_number, attempt, delay_ms, error = ?e,
1117 "S3 upload_part transient failure, retrying"
1118 );
1119 tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
1120 }
1121 Err(e) => break Err(e),
1122 }
1123 };
1124
1125 let resp = resp
1126 .map_err(|e| format!("S3 upload part {part_number} failed after retries: {e}"))?;
1127 let etag = resp.e_tag().unwrap_or_default().to_string();
1128 completed_parts.push((part_number, etag));
1129
1130 part_number += 1;
1131 }
1132
1133 if completed_parts.is_empty() {
1134 return Err("No parts uploaded (empty file)".to_string());
1135 }
1136
1137 // Shared completion path (with its own transient retry). The caller owns
1138 // the single abort-on-Err, so a failure here still aborts exactly once.
1139 self.complete_multipart_upload(key, upload_id, &completed_parts)
1140 .await
1141 }
1142
1143 /// Abort a multipart upload, releasing any uploaded parts.
1144 ///
1145 /// Public and `Result`-returning so the pending-upload reaper can abort an
1146 /// orphaned session and retry on failure. Internal callers that only want
1147 /// best-effort cleanup (e.g. [`Self::upload_multipart`] unwinding a failed
1148 /// transfer) log the error and move on.
1149 pub async fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> Result<(), String> {
1150 self.client
1151 .abort_multipart_upload()
1152 .bucket(&self.bucket)
1153 .key(key)
1154 .upload_id(upload_id)
1155 .send()
1156 .await
1157 .map(|_| ())
1158 .map_err(|e| format!("S3 abort multipart upload for {key} failed: {e}"))
1159 }
1160
1161 /// List the in-progress multipart uploads for exactly `key`, returning their
1162 /// upload ids.
1163 ///
1164 /// The orphan reaper needs this: a session that was started but never
1165 /// completed has **no object to delete**, only uploaded parts that S3 bills
1166 /// for until they are aborted, so a plain `delete` is a no-op against it. The
1167 /// upload id is not recorded durably anywhere, so it is recovered from S3,
1168 /// which also catches sessions whose tracking row was lost entirely.
1169 ///
1170 /// `ListMultipartUploads` matches a *prefix*, so results are filtered to an
1171 /// exact key match — otherwise reaping `staging/abc` would also abort a live
1172 /// session for `staging/abcdef`.
1173 pub async fn list_multipart_uploads_for_key(&self, key: &str) -> Result<Vec<String>, String> {
1174 let mut ids = Vec::new();
1175 let mut key_marker: Option<String> = None;
1176 let mut upload_id_marker: Option<String> = None;
1177
1178 loop {
1179 let mut req = self
1180 .client
1181 .list_multipart_uploads()
1182 .bucket(&self.bucket)
1183 .prefix(key);
1184 if let Some(ref k) = key_marker {
1185 req = req.key_marker(k);
1186 }
1187 if let Some(ref u) = upload_id_marker {
1188 req = req.upload_id_marker(u);
1189 }
1190
1191 let resp = req
1192 .send()
1193 .await
1194 .map_err(|e| format!("S3 list_multipart_uploads for {key} failed: {e}"))?;
1195
1196 for upload in resp.uploads() {
1197 // Exact-key filter: the request matched on prefix.
1198 if upload.key() == Some(key)
1199 && let Some(id) = upload.upload_id()
1200 {
1201 ids.push(id.to_string());
1202 }
1203 }
1204
1205 if resp.is_truncated().unwrap_or(false) {
1206 key_marker = resp.next_key_marker().map(str::to_string);
1207 upload_id_marker = resp.next_upload_id_marker().map(str::to_string);
1208 // Defensive: a truncated response with no markers would loop forever.
1209 if key_marker.is_none() && upload_id_marker.is_none() {
1210 break;
1211 }
1212 } else {
1213 break;
1214 }
1215 }
1216
1217 Ok(ids)
1218 }
1219
1220 /// Server-side copy an object using multipart `UploadPartCopy`, for sources
1221 /// larger than the 5 GiB single-part [`Self::copy_object`]/`CopyObject`
1222 /// limit. No bytes transit the caller — S3 copies each byte range internally.
1223 ///
1224 /// This is the >5 GiB promote path: scan-then-promote lifts a Clean staging
1225 /// object to the served content key, and a single `CopyObject` caps at 5 GiB,
1226 /// so a large video would otherwise succeed at upload and fail at promote —
1227 /// the worst failure position. `src_size` is the source object's size (the
1228 /// caller already reads it at confirm); `part_size` defaults to an
1229 /// auto-chosen size via [`MultipartPlan::auto`]. `content_type` sets the
1230 /// destination's type, since a fresh multipart upload does not inherit the
1231 /// source's metadata the way `CopyObject` does.
1232 ///
1233 /// Abort lives in one place, as in [`Self::upload_multipart`]: any failure
1234 /// after the destination upload is created aborts it exactly once.
1235 pub async fn copy_object_multipart(
1236 &self,
1237 src_bucket: &str,
1238 src_key: &str,
1239 dst_key: &str,
1240 content_type: &str,
1241 src_size: u64,
1242 part_size: Option<usize>,
1243 ) -> Result<(), String> {
1244 // Pre-flight: plan before anything is created, so a bad size strands
1245 // nothing and makes no network call.
1246 let plan = match part_size {
1247 Some(ps) => MultipartPlan::new(src_size, ps)?,
1248 None => MultipartPlan::auto(src_size)?,
1249 };
1250
1251 let upload_id = self.create_multipart_upload(dst_key, content_type).await?;
1252
1253 match self
1254 .run_multipart_copy(src_bucket, src_key, dst_key, &plan, &upload_id)
1255 .await
1256 {
1257 Ok(()) => Ok(()),
1258 Err(e) => {
1259 if let Err(abort_err) = self.abort_multipart_upload(dst_key, &upload_id).await {
1260 tracing::warn!("Failed to abort multipart copy for {dst_key}: {abort_err}");
1261 }
1262 Err(e)
1263 }
1264 }
1265 }
1266
1267 /// Issue every `UploadPartCopy` for a ranged multipart copy, then complete
1268 /// it. Returns `Err` (without aborting) on any failure; the caller
1269 /// [`Self::copy_object_multipart`] owns the single abort.
1270 async fn run_multipart_copy(
1271 &self,
1272 src_bucket: &str,
1273 src_key: &str,
1274 dst_key: &str,
1275 plan: &MultipartPlan,
1276 upload_id: &str,
1277 ) -> Result<(), String> {
1278 // `CopySource` is `{bucket}/{key}`; keys are sanitized to
1279 // `[A-Za-z0-9._/-]` upstream, so no percent-encoding is needed (same as
1280 // `copy_object_from`).
1281 let copy_source = format!("{src_bucket}/{src_key}");
1282 let mut completed_parts: Vec<(i32, String)> = Vec::with_capacity(plan.part_count as usize);
1283
1284 for part_number in 1..=plan.part_count {
1285 let (start, end) = plan
1286 .part_range(part_number)
1287 .ok_or_else(|| format!("internal: part {part_number} outside plan range"))?;
1288
1289 // Retry transient copy failures, mirroring the part-upload path.
1290 let mut attempt: u32 = 0;
1291 let resp = loop {
1292 attempt += 1;
1293 match self
1294 .client
1295 .upload_part_copy()
1296 .bucket(&self.bucket)
1297 .key(dst_key)
1298 .upload_id(upload_id)
1299 .part_number(part_number as i32)
1300 .copy_source(&copy_source)
1301 .copy_source_range(format!("bytes={start}-{end}"))
1302 .send()
1303 .await
1304 {
1305 Ok(r) => break Ok(r),
1306 Err(e) if attempt < 3 => {
1307 let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
1308 tracing::warn!(
1309 part_number, attempt, delay_ms, error = ?e,
1310 "S3 upload_part_copy transient failure, retrying"
1311 );
1312 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
1313 }
1314 Err(e) => break Err(e),
1315 }
1316 };
1317
1318 let resp = resp.map_err(|e| {
1319 format!("S3 upload_part_copy part {part_number} failed after retries: {e}")
1320 })?;
1321 let etag = resp
1322 .copy_part_result()
1323 .and_then(|r| r.e_tag())
1324 .unwrap_or_default()
1325 .to_string();
1326 completed_parts.push((part_number as i32, etag));
1327 }
1328
1329 self.complete_multipart_upload(dst_key, upload_id, &completed_parts)
1330 .await
1331 }
1332
1333 /// Configure CORS on the bucket for browser uploads.
1334 pub async fn configure_cors(&self, allowed_origin: &str) {
1335 let origin = allowed_origin.trim_end_matches('/').to_string();
1336 let rule = match CorsRule::builder()
1337 .allowed_origins(&origin)
1338 .allowed_methods("PUT")
1339 .allowed_methods("GET")
1340 .allowed_methods("HEAD")
1341 .allowed_headers("Content-Type")
1342 .allowed_headers("Cache-Control")
1343 .allowed_headers("Content-Disposition")
1344 .expose_headers("ETag")
1345 .max_age_seconds(3600)
1346 .build()
1347 {
1348 Ok(r) => r,
1349 Err(e) => {
1350 tracing::warn!("Failed to build CORS rule: {}", e);
1351 return;
1352 }
1353 };
1354
1355 let cors_config = match CorsConfiguration::builder().cors_rules(rule).build() {
1356 Ok(c) => c,
1357 Err(e) => {
1358 tracing::warn!("Failed to build CORS config: {}", e);
1359 return;
1360 }
1361 };
1362
1363 match self
1364 .client
1365 .put_bucket_cors()
1366 .bucket(&self.bucket)
1367 .cors_configuration(cors_config)
1368 .send()
1369 .await
1370 {
1371 Ok(_) => tracing::info!("S3 bucket CORS configured for {}", origin),
1372 Err(e) => tracing::warn!("Failed to configure S3 CORS: {}", e),
1373 }
1374 }
1375
1376 /// Read the bucket's CORS configuration back.
1377 ///
1378 /// A bucket with no CORS configuration is `Ok(vec![])`, not an error: S3
1379 /// answers `GetBucketCors` with `NoSuchCORSConfiguration` in that case, and
1380 /// "no rules" is the honest reading of it rather than a failure. Anything
1381 /// else is an error.
1382 pub async fn bucket_cors(&self) -> Result<Vec<CorsRuleView>, String> {
1383 match self
1384 .client
1385 .get_bucket_cors()
1386 .bucket(&self.bucket)
1387 .send()
1388 .await
1389 {
1390 Ok(resp) => Ok(resp
1391 .cors_rules()
1392 .iter()
1393 .map(|r| CorsRuleView {
1394 allowed_origins: r.allowed_origins().to_vec(),
1395 allowed_methods: r.allowed_methods().to_vec(),
1396 allowed_headers: r.allowed_headers().to_vec(),
1397 expose_headers: r.expose_headers().to_vec(),
1398 max_age_seconds: r.max_age_seconds(),
1399 })
1400 .collect()),
1401 Err(e) => {
1402 let service_error = e.into_service_error();
1403 if service_error.code() == Some("NoSuchCORSConfiguration") {
1404 Ok(Vec::new())
1405 } else {
1406 Err(format!("S3 get_bucket_cors failed: {service_error}"))
1407 }
1408 }
1409 }
1410 }
1411
1412 /// Lightweight connectivity check — `list_objects_v2` with `max_keys(0)`.
1413 pub async fn check_connectivity(&self) -> Result<(), String> {
1414 self.client
1415 .list_objects_v2()
1416 .bucket(&self.bucket)
1417 .max_keys(0)
1418 .send()
1419 .await
1420 .map(|_| ())
1421 .map_err(|e| format!("{e}"))
1422 }
1423 }
1424
1425 #[cfg(test)]
1426 mod tests {
1427 use super::*;
1428
1429 use aws_sdk_s3::config::retry::RetryConfig;
1430 use aws_smithy_http_client::test_util::{ReplayEvent, StaticReplayClient};
1431 use aws_smithy_types::body::SdkBody;
1432
1433 fn test_client() -> S3Client {
1434 // `from_conf` is local — no network until a request is sent — so this
1435 // builds a usable client without reaching any endpoint.
1436 let s3_config = aws_sdk_s3::Config::builder()
1437 .behavior_version(BehaviorVersion::latest())
1438 .http_client(https_client())
1439 .region(Region::new("test"))
1440 .endpoint_url("http://127.0.0.1:1")
1441 .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
1442 .force_path_style(true)
1443 .build();
1444 S3Client {
1445 client: Client::from_conf(s3_config),
1446 bucket: "test-bucket".to_string(),
1447 }
1448 }
1449
1450 /// An `S3Client` whose HTTP layer replays canned responses.
1451 ///
1452 /// WHY THIS EXISTS. Two paths in this crate cannot be reached from a live
1453 /// object store, and both were mutation survivors (infra `f7f13914`,
1454 /// `a536db81`). One is defensive code against a response shape S3 does not
1455 /// currently produce; the other writes a bucket setting the test bucket's
1456 /// credentials cannot read back. A canned response reaches the first and a
1457 /// recorded request reaches the second, with no live infrastructure and no
1458 /// bucket-owner permission.
1459 ///
1460 /// Retries are off so an exhausted replay list fails immediately: a mutant
1461 /// that makes the crate send one request too many should show up as a
1462 /// failed assertion rather than as seconds of SDK backoff.
1463 fn replay_client(events: Vec<ReplayEvent>) -> (S3Client, StaticReplayClient) {
1464 let replay = StaticReplayClient::new(events);
1465 let s3_config = aws_sdk_s3::Config::builder()
1466 .behavior_version(BehaviorVersion::latest())
1467 .http_client(replay.clone())
1468 .retry_config(RetryConfig::disabled())
1469 .region(Region::new("test"))
1470 .endpoint_url("http://127.0.0.1:1")
1471 .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
1472 .force_path_style(true)
1473 .build();
1474 (
1475 S3Client {
1476 client: Client::from_conf(s3_config),
1477 bucket: "test-bucket".to_string(),
1478 },
1479 replay,
1480 )
1481 }
1482
1483 /// A canned 200 with an XML body, which is what every S3 read returns.
1484 fn xml_ok(body: &str) -> ReplayEvent {
1485 ReplayEvent::new(
1486 http::Request::builder()
1487 .uri("http://test-bucket.localhost/")
1488 .body(SdkBody::empty())
1489 .unwrap(),
1490 http::Response::builder()
1491 .status(200)
1492 .header("content-type", "application/xml")
1493 .body(SdkBody::from(body.to_string()))
1494 .unwrap(),
1495 )
1496 }
1497
1498 /// A canned S3 error document, which is how S3 says "no CORS here".
1499 fn xml_err(status: u16, code: &str) -> ReplayEvent {
1500 let body = format!(
1501 "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
1502 <Error><Code>{code}</Code><Message>canned</Message>\
1503 <RequestId>r</RequestId><HostId>h</HostId></Error>"
1504 );
1505 ReplayEvent::new(
1506 http::Request::builder()
1507 .uri("http://test-bucket.localhost/")
1508 .body(SdkBody::empty())
1509 .unwrap(),
1510 http::Response::builder()
1511 .status(status)
1512 .header("content-type", "application/xml")
1513 .body(SdkBody::from(body))
1514 .unwrap(),
1515 )
1516 }
1517
1518 /// infra `f7f13914`. The "truncated but no continuation markers" guard in
1519 /// `list_multipart_uploads_for_key` survived mutation because reaching it
1520 /// needs a response S3 does not produce: truncated, yet naming nothing to
1521 /// continue from. Live MinIO will not build it, and >1000 pending uploads on
1522 /// one key is not a fixture anyone should own. A canned body is.
1523 ///
1524 /// The guard is kept rather than deleted because without it that response
1525 /// is an infinite loop against a real endpoint, and the crate is the last
1526 /// thing standing between the orphan reaper and a spin.
1527 ///
1528 /// The single-request assertion is the one that matters. Removing the
1529 /// `break` makes the loop ask again with the same (absent) markers, and
1530 /// there is no second canned response, so it also fails the `Ok` — either
1531 /// way the mutant dies.
1532 #[tokio::test]
1533 async fn a_truncated_listing_with_no_markers_stops_instead_of_looping() {
1534 let (client, replay) = replay_client(vec![xml_ok(
1535 r#"<?xml version="1.0" encoding="UTF-8"?>
1536 <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1537 <Bucket>test-bucket</Bucket>
1538 <Prefix>staging/abc</Prefix>
1539 <MaxUploads>1000</MaxUploads>
1540 <IsTruncated>true</IsTruncated>
1541 <Upload>
1542 <Key>staging/abc</Key>
1543 <UploadId>upload-one</UploadId>
1544 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1545 </Upload>
1546 <Upload>
1547 <Key>staging/abcdef</Key>
1548 <UploadId>not-ours</UploadId>
1549 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1550 </Upload>
1551 </ListMultipartUploadsResult>"#,
1552 )]);
1553
1554 let ids = client
1555 .list_multipart_uploads_for_key("staging/abc")
1556 .await
1557 .expect("a truncated page with nothing to continue from is a complete answer");
1558
1559 // The exact-key filter travels with it: `ListMultipartUploads` matches a
1560 // prefix, so `staging/abcdef` is a live session the reaper must not abort.
1561 assert_eq!(ids, vec!["upload-one".to_string()]);
1562 assert_eq!(
1563 replay.actual_requests().count(),
1564 1,
1565 "the guard exists to stop a second identical request"
1566 );
1567 }
1568
1569 /// The other half of the guard: a truncated page that DOES name a marker is
1570 /// followed. Without this the test above is satisfied by a loop that never
1571 /// iterates at all, which is a different bug wearing the same result.
1572 #[tokio::test]
1573 async fn a_truncated_listing_with_a_marker_asks_for_the_next_page() {
1574 let (client, replay) = replay_client(vec![
1575 xml_ok(
1576 r#"<?xml version="1.0" encoding="UTF-8"?>
1577 <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1578 <Bucket>test-bucket</Bucket>
1579 <IsTruncated>true</IsTruncated>
1580 <NextKeyMarker>staging/abc</NextKeyMarker>
1581 <NextUploadIdMarker>upload-one</NextUploadIdMarker>
1582 <Upload>
1583 <Key>staging/abc</Key>
1584 <UploadId>upload-one</UploadId>
1585 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1586 </Upload>
1587 </ListMultipartUploadsResult>"#,
1588 ),
1589 xml_ok(
1590 r#"<?xml version="1.0" encoding="UTF-8"?>
1591 <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1592 <Bucket>test-bucket</Bucket>
1593 <IsTruncated>false</IsTruncated>
1594 <Upload>
1595 <Key>staging/abc</Key>
1596 <UploadId>upload-two</UploadId>
1597 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1598 </Upload>
1599 </ListMultipartUploadsResult>"#,
1600 ),
1601 ]);
1602
1603 let ids = client
1604 .list_multipart_uploads_for_key("staging/abc")
1605 .await
1606 .expect("two pages is an ordinary listing");
1607
1608 assert_eq!(
1609 ids,
1610 vec!["upload-one".to_string(), "upload-two".to_string()]
1611 );
1612 let second = replay
1613 .actual_requests()
1614 .nth(1)
1615 .expect("the second page was requested")
1616 .uri()
1617 .to_string();
1618 assert!(
1619 second.contains("upload-id-marker=upload-one"),
1620 "the marker from page one carries into page two: {second}"
1621 );
1622 }
1623
1624 /// A page may name only ONE of the two markers, and that is still a page to
1625 /// follow. The guard reads "neither marker", so it is an `&&`; an `||` there
1626 /// stops on the first page whose key marker happens to be the only one set,
1627 /// silently returning a short list to the orphan reaper -- which then leaves
1628 /// the parts it did not see billing forever.
1629 ///
1630 /// `ListMultipartUploads` returns `NextUploadIdMarker` only when the page
1631 /// splits a key's uploads, so a page ending on a key boundary carries the
1632 /// key marker alone. That is the ordinary case, not a corner.
1633 #[tokio::test]
1634 async fn a_page_naming_only_the_key_marker_is_still_followed() {
1635 let (client, replay) = replay_client(vec![
1636 xml_ok(
1637 r#"<?xml version="1.0" encoding="UTF-8"?>
1638 <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1639 <Bucket>test-bucket</Bucket>
1640 <IsTruncated>true</IsTruncated>
1641 <NextKeyMarker>staging/abc</NextKeyMarker>
1642 <Upload>
1643 <Key>staging/abc</Key>
1644 <UploadId>upload-one</UploadId>
1645 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1646 </Upload>
1647 </ListMultipartUploadsResult>"#,
1648 ),
1649 xml_ok(
1650 r#"<?xml version="1.0" encoding="UTF-8"?>
1651 <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1652 <Bucket>test-bucket</Bucket>
1653 <IsTruncated>false</IsTruncated>
1654 <Upload>
1655 <Key>staging/abc</Key>
1656 <UploadId>upload-two</UploadId>
1657 <Initiated>2026-08-31T00:00:00.000Z</Initiated>
1658 </Upload>
1659 </ListMultipartUploadsResult>"#,
1660 ),
1661 ]);
1662
1663 let ids = client
1664 .list_multipart_uploads_for_key("staging/abc")
1665 .await
1666 .expect("one marker is enough to continue");
1667
1668 assert_eq!(
1669 ids,
1670 vec!["upload-one".to_string(), "upload-two".to_string()]
1671 );
1672 assert_eq!(replay.actual_requests().count(), 2);
1673 }
1674
1675 /// infra `a536db81`. `configure_cors` returns `()` and the crate exposed no
1676 /// readback, so replacing its whole body with `()` was invisible to every
1677 /// test that could be written against it. The request it sends is the
1678 /// observable, and it is the right observable: what the object store ends up
1679 /// configured with is decided entirely by that one PUT.
1680 #[tokio::test]
1681 async fn configuring_cors_sends_the_rule_the_browser_upload_needs() {
1682 let (client, replay) = replay_client(vec![xml_ok("")]);
1683
1684 client.configure_cors("https://example.test/").await;
1685
1686 let requests: Vec<_> = replay.actual_requests().collect();
1687 assert_eq!(requests.len(), 1, "configure_cors must send a PUT");
1688 let uri = requests[0].uri().to_string();
1689 assert!(uri.contains("cors"), "put_bucket_cors, not some other PUT");
1690
1691 let body = String::from_utf8(
1692 requests[0]
1693 .body()
1694 .bytes()
1695 .expect("an in-memory XML body")
1696 .to_vec(),
1697 )
1698 .expect("the CORS document is UTF-8");
1699
1700 // The trailing slash is trimmed: an origin is a scheme-host-port and a
1701 // browser never sends the slash, so a rule carrying one matches nothing.
1702 assert!(body.contains("<AllowedOrigin>https://example.test</AllowedOrigin>"));
1703 for method in ["PUT", "GET", "HEAD"] {
1704 assert!(
1705 body.contains(&format!("<AllowedMethod>{method}</AllowedMethod>")),
1706 "{method} is missing from {body}"
1707 );
1708 }
1709 for header in ["Content-Type", "Cache-Control", "Content-Disposition"] {
1710 assert!(
1711 body.contains(&format!("<AllowedHeader>{header}</AllowedHeader>")),
1712 "{header} is missing from {body}"
1713 );
1714 }
1715 // ETag is what a direct-to-S3 part upload reads back to complete the
1716 // upload; without exposing it the browser cannot finish a multipart.
1717 assert!(body.contains("<ExposeHeader>ETag</ExposeHeader>"));
1718 assert!(body.contains("<MaxAgeSeconds>3600</MaxAgeSeconds>"));
1719 }
1720
1721 /// The readback the wrapper was missing, over a canned response.
1722 #[tokio::test]
1723 async fn cors_rules_come_back_in_the_crates_own_shape() {
1724 let (client, _replay) = replay_client(vec![xml_ok(
1725 r#"<?xml version="1.0" encoding="UTF-8"?>
1726 <CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
1727 <CORSRule>
1728 <AllowedOrigin>https://example.test</AllowedOrigin>
1729 <AllowedMethod>PUT</AllowedMethod>
1730 <AllowedMethod>GET</AllowedMethod>
1731 <AllowedHeader>Content-Type</AllowedHeader>
1732 <ExposeHeader>ETag</ExposeHeader>
1733 <MaxAgeSeconds>3600</MaxAgeSeconds>
1734 </CORSRule>
1735 </CORSConfiguration>"#,
1736 )]);
1737
1738 let rules = client.bucket_cors().await.expect("a configured bucket");
1739 assert_eq!(
1740 rules,
1741 vec![CorsRuleView {
1742 allowed_origins: vec!["https://example.test".to_string()],
1743 allowed_methods: vec!["PUT".to_string(), "GET".to_string()],
1744 allowed_headers: vec!["Content-Type".to_string()],
1745 expose_headers: vec!["ETag".to_string()],
1746 max_age_seconds: Some(3600),
1747 }]
1748 );
1749 }
1750
1751 /// A bucket with no CORS is not a failure. S3 answers `GetBucketCors` with
1752 /// `NoSuchCORSConfiguration`, and a caller asking "what is set" is entitled
1753 /// to hear "nothing" rather than an error it has to pattern-match itself.
1754 #[tokio::test]
1755 async fn a_bucket_with_no_cors_reads_back_as_no_rules() {
1756 let (client, _replay) = replay_client(vec![xml_err(404, "NoSuchCORSConfiguration")]);
1757
1758 let rules = client.bucket_cors().await.expect("absence is not an error");
1759 assert!(rules.is_empty());
1760 }
1761
1762 /// Any other error still is one, or the method above would report a broken
1763 /// endpoint as an unconfigured bucket.
1764 #[tokio::test]
1765 async fn a_cors_read_that_fails_for_another_reason_is_an_error() {
1766 let (client, _replay) = replay_client(vec![xml_err(403, "AccessDenied")]);
1767
1768 let err = client
1769 .bucket_cors()
1770 .await
1771 .expect_err("AccessDenied is not an empty CORS configuration");
1772 assert!(err.contains("get_bucket_cors"), "{err}");
1773 }
1774
1775 const MIB: u64 = 1024 * 1024;
1776
1777 #[test]
1778 fn oracle_accepts_every_plan_the_crate_makes() {
1779 // Spot sizes across the whole legal range, including the exact
1780 // boundaries, since those are where tiling arithmetic goes wrong.
1781 for total in [
1782 1,
1783 MULTIPART_MIN_PART_SIZE as u64 - 1,
1784 MULTIPART_MIN_PART_SIZE as u64,
1785 MULTIPART_MIN_PART_SIZE as u64 + 1,
1786 25 * MIB,
1787 MULTIPART_DEFAULT_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64,
1788 MULTIPART_MAX_OBJECT_SIZE - 1,
1789 MULTIPART_MAX_OBJECT_SIZE,
1790 ] {
1791 oracle::check_auto(total);
1792 oracle::check_plan(total, MULTIPART_MIN_PART_SIZE);
1793 oracle::check_plan(total, MULTIPART_DEFAULT_PART_SIZE);
1794 }
1795 // And the refusals, which must be refusals for a reason that holds.
1796 oracle::check_auto(0);
1797 oracle::check_auto(MULTIPART_MAX_OBJECT_SIZE + 1);
1798 oracle::check_plan(25 * MIB, MULTIPART_MIN_PART_SIZE - 1);
1799 oracle::check_plan(25 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1);
1800 }
1801
1802 #[test]
1803 fn the_limits_are_the_numbers_the_s3_contract_states() {
1804 // Every one of these is written as an arithmetic expression, and an
1805 // expression nothing asserts is a number nothing pins: mutation found
1806 // that `5 * 1024 * 1024 * 1024` could become `5 + 1024 + 1024 + 1024`
1807 // and no test noticed (infra `1498fe2a`). The plans built from them
1808 // would still be self-consistent, which is exactly why the planner's
1809 // own tests cannot catch it. Stated here in bytes, from the S3 API
1810 // contract.
1811 assert_eq!(MULTIPART_MIN_PART_SIZE, 5_242_880, "5 MiB");
1812 assert_eq!(MULTIPART_MAX_PARTS, 10_000);
1813 assert_eq!(MULTIPART_MAX_PART_SIZE, 5_368_709_120, "5 GiB");
1814 assert_eq!(MULTIPART_MAX_OBJECT_SIZE, 5_497_558_138_880, "5 TiB");
1815 assert_eq!(MULTIPART_DEFAULT_PART_SIZE, 16_777_216, "16 MiB");
1816 assert_eq!(MAX_PRESIGN_EXPIRY_SECS, 604_800, "7 days, SigV4's maximum");
1817 }
1818
1819 #[test]
1820 #[should_panic(expected = "disagrees with div_ceil")]
1821 fn oracle_catches_a_part_count_the_client_would_reject() {
1822 // Hand-built, because the crate will not produce it. SyncKit refuses a
1823 // session whose part_count is not div_ceil (synckit-client
1824 // `client/blob.rs`), so a plan like this is an upload that can never
1825 // start.
1826 //
1827 // THE TILING PROPERTY HAS NO SUCH TEST, and deliberately so: it cannot
1828 // be broken by hand. `part_len` and `part_range` both derive from
1829 // `part_size` and `part_count`, so any plan that satisfies the
1830 // div_ceil check above necessarily tiles. What establishes that the
1831 // tiling assertions are observed rather than merely present is the
1832 // mutation run (infra `1498fe2a`), which changes the derivation itself.
1833 // That division is the point of running both.
1834 let plan = MultipartPlan {
1835 total_size: 25 * MIB,
1836 part_size: 10 * MIB as usize,
1837 part_count: 4,
1838 };
1839 oracle::check_geometry(&plan);
1840 }
1841
1842 #[test]
1843 fn multipart_plan_divides_with_remainder() {
1844 // 25 MiB in 10 MiB parts -> 10 + 10 + 5.
1845 let plan = MultipartPlan::new(25 * MIB, 10 * MIB as usize).unwrap();
1846 assert_eq!(plan.part_count, 3);
1847 assert_eq!(plan.part_len(1), 10 * MIB);
1848 assert_eq!(plan.part_len(2), 10 * MIB);
1849 assert_eq!(plan.part_len(3), 5 * MIB);
1850 assert_eq!(plan.part_len(4), 0, "out-of-range part");
1851 assert_eq!(plan.part_range(1), Some((0, 10 * MIB - 1)));
1852 assert_eq!(plan.part_range(3), Some((20 * MIB, 25 * MIB - 1)));
1853 assert_eq!(plan.part_range(4), None);
1854 }
1855
1856 #[test]
1857 fn multipart_plan_divides_evenly() {
1858 // 20 MiB in 5 MiB parts -> four full parts, last is a full part.
1859 let plan = MultipartPlan::new(20 * MIB, MULTIPART_MIN_PART_SIZE).unwrap();
1860 assert_eq!(plan.part_count, 4);
1861 assert_eq!(plan.part_len(4), 5 * MIB);
1862 assert_eq!(plan.part_range(4), Some((15 * MIB, 20 * MIB - 1)));
1863 }
1864
1865 #[test]
1866 fn multipart_plan_rejects_empty_object() {
1867 let err = MultipartPlan::new(0, MULTIPART_MIN_PART_SIZE).unwrap_err();
1868 assert!(err.contains("non-empty"), "unexpected error: {err}");
1869 }
1870
1871 #[test]
1872 fn multipart_plan_rejects_undersized_part() {
1873 let err = MultipartPlan::new(100 * MIB, MULTIPART_MIN_PART_SIZE - 1).unwrap_err();
1874 assert!(err.contains("5 MiB"), "unexpected error: {err}");
1875 }
1876
1877 #[test]
1878 fn multipart_plan_rejects_oversized_part() {
1879 let err = MultipartPlan::new(10 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1).unwrap_err();
1880 assert!(err.contains("5 GiB"), "unexpected error: {err}");
1881 }
1882
1883 #[test]
1884 fn multipart_plan_rejects_too_many_parts() {
1885 // One 5 MiB part past the 10k limit at the minimum part size.
1886 let total = MULTIPART_MIN_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 1);
1887 let err = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap_err();
1888 assert!(err.contains("10000-part"), "unexpected error: {err}");
1889 }
1890
1891 #[test]
1892 fn multipart_plan_accepts_exactly_max_parts() {
1893 let total = MULTIPART_MIN_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64;
1894 let plan = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap();
1895 assert_eq!(plan.part_count, MULTIPART_MAX_PARTS);
1896 }
1897
1898 #[test]
1899 fn multipart_plan_rejects_over_object_ceiling() {
1900 let err = MultipartPlan::new(
1901 MULTIPART_MAX_OBJECT_SIZE + 1,
1902 MULTIPART_MAX_PART_SIZE as usize,
1903 )
1904 .unwrap_err();
1905 assert!(err.contains("5 TiB"), "unexpected error: {err}");
1906 }
1907
1908 #[test]
1909 fn multipart_plan_auto_uses_default_for_small_objects() {
1910 let plan = MultipartPlan::auto(100 * MIB).unwrap();
1911 assert_eq!(plan.part_size, MULTIPART_DEFAULT_PART_SIZE);
1912 // 100 MiB / 16 MiB -> 7 parts (ceil).
1913 assert_eq!(plan.part_count, 7);
1914 }
1915
1916 #[test]
1917 fn multipart_plan_auto_scales_part_size_to_stay_within_part_cap() {
1918 // An object too big for the default part size within 10k parts must get
1919 // a larger part size, and the resulting plan must be valid.
1920 let big = MULTIPART_DEFAULT_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 500);
1921 let plan = MultipartPlan::auto(big).unwrap();
1922 assert!(plan.part_size > MULTIPART_DEFAULT_PART_SIZE);
1923 assert!(plan.part_count <= MULTIPART_MAX_PARTS);
1924 // Whole-MiB part size.
1925 assert_eq!(plan.part_size as u64 % MIB, 0);
1926 }
1927
1928 #[test]
1929 fn multipart_plan_auto_rejects_empty() {
1930 assert!(MultipartPlan::auto(0).is_err());
1931 }
1932
1933 #[tokio::test]
1934 async fn copy_object_multipart_rejects_empty_source_before_any_request() {
1935 // Plan is pre-flight: an empty source fails before the destination
1936 // multipart upload is created, so the unreachable endpoint is untouched.
1937 let client = test_client();
1938 let err = client
1939 .copy_object_multipart("bkt", "src", "dst", "application/octet-stream", 0, None)
1940 .await
1941 .expect_err("empty source must be rejected");
1942 assert!(err.contains("non-empty"), "unexpected error: {err}");
1943 }
1944
1945 /// The `X-Amz-SignedHeaders` list from a presigned URL.
1946 fn signed_headers(url: &str) -> String {
1947 url.split('&')
1948 .find_map(|p| p.strip_prefix("X-Amz-SignedHeaders="))
1949 .map(|v| v.replace("%3B", ";"))
1950 .expect("presigned URL must carry X-Amz-SignedHeaders")
1951 }
1952
1953 #[tokio::test]
1954 async fn presign_upload_signs_content_length_when_bound() {
1955 // Callers rely on `max_bytes` being enforced, and it is enforced only
1956 // because it lands in SignedHeaders: a client sending a different
1957 // Content-Length then fails the signature. That also makes the declared
1958 // size a hard contract — a caller that declares anything other than the
1959 // exact body length breaks every upload — so pin it here rather than
1960 // discovering it against production S3.
1961 let client = test_client();
1962
1963 let bound = client
1964 .presign_upload("k", "application/octet-stream", 900, None, Some(12_345))
1965 .await
1966 .unwrap();
1967 let headers = signed_headers(&bound);
1968 assert!(
1969 headers.contains("content-length"),
1970 "max_bytes must be signed, got: {headers}"
1971 );
1972
1973 let unbound = client
1974 .presign_upload("k", "application/octet-stream", 900, None, None)
1975 .await
1976 .unwrap();
1977 assert!(
1978 !signed_headers(&unbound).contains("content-length"),
1979 "without max_bytes the client is free to send any length"
1980 );
1981 }
1982
1983 #[tokio::test]
1984 async fn presign_upload_part_rejects_out_of_range_part_number() {
1985 // Pre-flight range check: fires before any network call, so the
1986 // unreachable dummy endpoint is never touched.
1987 let client = test_client();
1988 for bad in [0, MULTIPART_MAX_PARTS as i32 + 1] {
1989 let err = client
1990 .presign_upload_part("k", "uid", bad, 3600, None, None)
1991 .await
1992 .expect_err("out-of-range part number must be rejected");
1993 assert!(err.contains("out of range"), "unexpected error: {err}");
1994 }
1995 }
1996
1997 #[tokio::test]
1998 async fn presign_upload_part_signs_the_checksum_when_bound() {
1999 // S3 enforces a bound checksum by rehashing the part, but only if the
2000 // client sends the header — which it must, because signing it makes it
2001 // mandatory. Both halves of that live in SignedHeaders.
2002 let client = test_client();
2003
2004 let bound = client
2005 .presign_upload_part("k", "uid", 1, 900, Some(64), Some("Zm9vYmFyYmF6"))
2006 .await
2007 .unwrap();
2008 let headers = signed_headers(&bound);
2009 assert!(
2010 headers.contains("x-amz-checksum-sha256"),
2011 "a bound checksum must be signed, got: {headers}"
2012 );
2013
2014 let unbound = client
2015 .presign_upload_part("k", "uid", 1, 900, Some(64), None)
2016 .await
2017 .unwrap();
2018 assert!(
2019 !signed_headers(&unbound).contains("checksum"),
2020 "no checksum bound means no checksum header is required"
2021 );
2022 }
2023
2024 #[tokio::test]
2025 async fn complete_multipart_rejects_empty_parts() {
2026 let client = test_client();
2027 let err = client
2028 .complete_multipart_upload("k", "uid", &[])
2029 .await
2030 .expect_err("empty parts must be rejected");
2031 assert!(err.contains("no parts"), "unexpected error: {err}");
2032 }
2033
2034 #[tokio::test]
2035 async fn upload_multipart_rejects_undersized_part_before_any_request() {
2036 // The minimum-part-size guard is pre-flight: it must fire before the
2037 // multipart upload is created, so there is nothing to strand and no
2038 // network call (the dummy endpoint is unreachable — reaching it would
2039 // hang/error instead of returning this exact message).
2040 let client = test_client();
2041 let path = std::path::Path::new("/nonexistent");
2042 let err = client
2043 .upload_multipart("k", "application/octet-stream", path, Some(1024))
2044 .await
2045 .expect_err("undersized part size must be rejected");
2046 assert!(err.contains("at least 5 MB"), "unexpected error: {err}");
2047 }
2048 }
2049