Skip to main content

max / makenotwork

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