Skip to main content

max / makenotwork

s3-storage: add client-direct multipart primitives Add the client-direct multipart surface to S3Client: create_multipart_upload, presign_upload_part (1-based, range-checked, signs Content-Length), and a shared complete_multipart_upload taking (part_number, etag) pairs. Promote the former private best-effort abort_multipart_upload to a public Result-returning method so the pending-upload reaper can abort orphaned sessions and retry. Add copy_object_multipart (ranged UploadPartCopy) for server-side copies above the 5 GiB single-part CopyObject limit, the >5 GiB promote path. Add the S3 multipart limits as constants with a compile-time guard, and a pure MultipartPlan (new/auto/part_len/part_range) as the one geometry source the blob and creator-media session endpoints and the client all share. Route the existing upload_multipart/run_multipart_upload through the new create and complete methods, deduplicating the create and completion-retry logic. 15 tests, clippy --all-targets clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 02:46 UTC
Commit: 7774b6bd856828e1b4e2b20fe01a50c2d206b1fb
Parent: 1365b8f
1 file changed, +393 insertions, -61 deletions
M shared/s3-storage/src/lib.rs +393 -61
@@ -46,6 +46,134 @@ pub struct S3Client {
46 46 /// opaque signing error — and it bounds how long any single minted URL can live.
47 47 const MAX_PRESIGN_EXPIRY_SECS: u64 = 7 * 24 * 60 * 60;
48 48
49 + /// S3 multipart-upload limits, straight from the S3 API contract. Every layer
50 + /// that plans a client-direct multipart upload (the blob and creator-media
51 + /// session endpoints) validates against these, so the client is never handed a
52 + /// plan S3 would reject at complete time.
53 + ///
54 + /// Minimum size of every part *except the last* (5 MiB). The final part may be
55 + /// anything down to 1 byte.
56 + pub const MULTIPART_MIN_PART_SIZE: usize = 5 * 1024 * 1024;
57 + /// Maximum number of parts in a single multipart upload.
58 + pub const MULTIPART_MAX_PARTS: u32 = 10_000;
59 + /// Maximum size of a single part (5 GiB).
60 + pub const MULTIPART_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
61 + /// Maximum size of an object assembled from a multipart upload (5 TiB).
62 + pub const MULTIPART_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024;
63 + /// Default/floor part size for an auto-planned multipart upload (16 MiB): big
64 + /// enough to keep the per-part round-trip overhead low, small enough that a
65 + /// resumable client re-sends little on a retry. Used when the object is small
66 + /// enough not to force larger parts to stay within the part-count limit.
67 + pub const MULTIPART_DEFAULT_PART_SIZE: usize = 16 * 1024 * 1024;
68 +
69 + // The part budget at max part size must be able to cover the object ceiling, or
70 + // a large-but-legal object would be unplannable at any part size. 10,000 x 5 GiB
71 + // ~= 48.8 TiB, comfortably over the 5 TiB object cap. (The 5 MiB *minimum* part
72 + // deliberately does NOT cover 5 TiB — a large object simply needs bigger parts,
73 + // which `MultipartPlan::new` enforces via the part-count limit.)
74 + const _: () = assert!(
75 + MULTIPART_MAX_PARTS as u128 * MULTIPART_MAX_PART_SIZE as u128
76 + >= MULTIPART_MAX_OBJECT_SIZE as u128
77 + );
78 +
79 + /// A validated multipart-upload plan: the part size to use and how many parts a
80 + /// body of `total_size` bytes splits into. Pure arithmetic with no S3 call, so
81 + /// the blob and creator-media session endpoints share one source of truth for
82 + /// part geometry (and the client computes the same boundaries independently).
83 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
84 + pub struct MultipartPlan {
85 + /// Total object size in bytes.
86 + pub total_size: u64,
87 + /// Size of every part except the last.
88 + pub part_size: usize,
89 + /// Number of parts (1..=[`MULTIPART_MAX_PARTS`]).
90 + pub part_count: u32,
91 + }
92 +
93 + impl MultipartPlan {
94 + /// Plan a multipart upload of `total_size` bytes into `part_size`-byte parts,
95 + /// the last part taking the remainder.
96 + ///
97 + /// Errors if the object is empty (use a single PUT), exceeds the 5 TiB
98 + /// multipart ceiling, the part size is below S3's 5 MiB floor or above its
99 + /// 5 GiB ceiling, or the resulting part count would exceed 10,000 — the last
100 + /// meaning the part size is too small for this object and the caller should
101 + /// pick a larger one.
102 + pub fn new(total_size: u64, part_size: usize) -> Result<Self, String> {
103 + if total_size == 0 {
104 + return Err(
105 + "multipart upload needs a non-empty object; use a single PUT for empty objects"
106 + .to_string(),
107 + );
108 + }
109 + if total_size > MULTIPART_MAX_OBJECT_SIZE {
110 + return Err(format!(
111 + "object is {total_size} bytes, over the {MULTIPART_MAX_OBJECT_SIZE}-byte (5 TiB) multipart ceiling"
112 + ));
113 + }
114 + if part_size < MULTIPART_MIN_PART_SIZE {
115 + return Err(format!(
116 + "part size {part_size} is below the {MULTIPART_MIN_PART_SIZE}-byte (5 MiB) S3 minimum"
117 + ));
118 + }
119 + if part_size as u64 > MULTIPART_MAX_PART_SIZE {
120 + return Err(format!(
121 + "part size {part_size} is above the {MULTIPART_MAX_PART_SIZE}-byte (5 GiB) S3 maximum"
122 + ));
123 + }
124 + let part_count = total_size.div_ceil(part_size as u64);
125 + if part_count > MULTIPART_MAX_PARTS as u64 {
126 + return Err(format!(
127 + "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"
128 + ));
129 + }
130 + Ok(Self { total_size, part_size, part_count: part_count as u32 })
131 + }
132 +
133 + /// Plan a multipart upload of `total_size` bytes, choosing the part size
134 + /// automatically: [`MULTIPART_DEFAULT_PART_SIZE`] when that keeps the object
135 + /// within [`MULTIPART_MAX_PARTS`] parts, otherwise the smallest whole-MiB
136 + /// part size that does. Errors only if the object is empty or over the 5 TiB
137 + /// ceiling — a valid non-empty object always yields a plan.
138 + pub fn auto(total_size: u64) -> Result<Self, String> {
139 + const MIB: u64 = 1024 * 1024;
140 + // Smallest part size that fits the object within the part-count cap,
141 + // rounded up to a whole MiB, then floored at the default part size.
142 + let needed = total_size.div_ceil(MULTIPART_MAX_PARTS as u64);
143 + let rounded = needed.div_ceil(MIB) * MIB;
144 + let part_size = (rounded as usize).max(MULTIPART_DEFAULT_PART_SIZE);
145 + Self::new(total_size, part_size)
146 + }
147 +
148 + /// Byte length of part `part_number` (1-based). The last part is the
149 + /// remainder; every earlier part is exactly `part_size`. Returns 0 for a
150 + /// part number outside `1..=part_count`.
151 + pub fn part_len(&self, part_number: u32) -> u64 {
152 + if part_number == 0 || part_number > self.part_count {
153 + return 0;
154 + }
155 + if part_number < self.part_count {
156 + return self.part_size as u64;
157 + }
158 + // Last part: the remainder, or a full part when the size divides evenly.
159 + match self.total_size % self.part_size as u64 {
160 + 0 => self.part_size as u64,
161 + rem => rem,
162 + }
163 + }
164 +
165 + /// Inclusive `[start, end]` byte range of part `part_number` (1-based), the
166 + /// form S3 `UploadPartCopy` wants as `bytes=start-end`. `None` for a part
167 + /// number outside `1..=part_count`.
168 + pub fn part_range(&self, part_number: u32) -> Option<(u64, u64)> {
169 + if part_number == 0 || part_number > self.part_count {
170 + return None;
171 + }
172 + let start = (part_number as u64 - 1) * self.part_size as u64;
173 + Some((start, start + self.part_len(part_number) - 1))
174 + }
175 + }
176 +
49 177 impl S3Client {
50 178 /// Create a new S3 client from configuration.
51 179 pub async fn new(config: &S3Config) -> Result<Self, String> {
@@ -503,6 +631,35 @@ impl S3Client {
503 631 return Err("Multipart part size must be at least 5 MB".to_string());
504 632 }
505 633
634 + let upload_id = self.create_multipart_upload(key, content_type).await?;
635 +
636 + // Single abort site: any failure past this point aborts exactly once.
637 + match self.run_multipart_upload(key, file_path, part_size, &upload_id).await {
638 + Ok(()) => Ok(()),
639 + Err(e) => {
640 + // Best-effort abort; the create succeeded, so on failure surface
641 + // the original error and just log if the cleanup also fails.
642 + if let Err(abort_err) = self.abort_multipart_upload(key, &upload_id).await {
643 + tracing::warn!("Failed to abort multipart upload for {key}: {abort_err}");
644 + }
645 + Err(e)
646 + }
647 + }
648 + }
649 +
650 + /// Begin a client-direct multipart upload, returning the `upload_id` that
651 + /// [`Self::presign_upload_part`], [`Self::complete_multipart_upload`], and
652 + /// [`Self::abort_multipart_upload`] key on.
653 + ///
654 + /// The counterpart to [`Self::upload_multipart`], which drives the whole
655 + /// transfer server-side from a local file. Here the server only mints the id
656 + /// and (via `presign_upload_part`) the per-part URLs; the client streams the
657 + /// parts straight to S3, so no object bytes transit the server.
658 + pub async fn create_multipart_upload(
659 + &self,
660 + key: &str,
661 + content_type: &str,
662 + ) -> Result<String, String> {
506 663 let create = self
507 664 .client
508 665 .create_multipart_upload()
@@ -513,17 +670,113 @@ impl S3Client {
513 670 .await
514 671 .map_err(|e| format!("S3 create multipart upload failed: {e}"))?;
515 672
516 - let upload_id = create
673 + create
517 674 .upload_id()
518 - .ok_or("S3 create multipart upload returned no upload_id")?
519 - .to_string();
675 + .map(str::to_string)
676 + .ok_or_else(|| "S3 create multipart upload returned no upload_id".to_string())
677 + }
520 678
521 - // Single abort site: any failure past this point aborts exactly once.
522 - match self.run_multipart_upload(key, file_path, part_size, &upload_id).await {
523 - Ok(()) => Ok(()),
524 - Err(e) => {
525 - self.abort_multipart_upload(key, &upload_id).await;
526 - Err(e)
679 + /// Presign an `UploadPart` request for one part of an in-progress multipart
680 + /// upload. `part_number` is 1-based (`1..=`[`MULTIPART_MAX_PARTS`]).
681 + ///
682 + /// When `max_bytes` is set it is signed as `Content-Length`, the same
683 + /// defense-in-depth (not a hard, server-enforced ceiling) as
684 + /// [`Self::presign_upload`] — see its docs for why the authoritative size
685 + /// check still lives at confirm time.
686 + pub async fn presign_upload_part(
687 + &self,
688 + key: &str,
689 + upload_id: &str,
690 + part_number: i32,
691 + expiry_secs: u64,
692 + max_bytes: Option<i64>,
693 + ) -> Result<String, String> {
694 + if !(1..=MULTIPART_MAX_PARTS as i32).contains(&part_number) {
695 + return Err(format!(
696 + "part number {part_number} out of range 1..={MULTIPART_MAX_PARTS}"
697 + ));
698 + }
699 + let presigning_config = PresigningConfig::builder()
700 + .expires_in(Duration::from_secs(expiry_secs.min(MAX_PRESIGN_EXPIRY_SECS)))
701 + .build()
702 + .map_err(|e| format!("Presigning config error: {e}"))?;
703 +
704 + let mut req = self
705 + .client
706 + .upload_part()
707 + .bucket(&self.bucket)
708 + .key(key)
709 + .upload_id(upload_id)
710 + .part_number(part_number);
711 +
712 + if let Some(n) = max_bytes {
713 + req = req.content_length(n);
714 + }
715 +
716 + let presigned = req
717 + .presigned(presigning_config)
718 + .await
719 + .map_err(|e| format!("Failed to generate upload part URL: {e}"))?;
720 +
721 + Ok(presigned.uri().to_string())
722 + }
723 +
724 + /// Complete a multipart upload from the collected `(part_number, etag)`
725 + /// pairs. Parts are sorted by number before assembly (S3 requires ascending
726 + /// order), so the caller may pass them in completion order.
727 + ///
728 + /// Retries the completion on transient failure: by this point every part is
729 + /// uploaded and paid for, so losing the completion call would orphan the
730 + /// whole upload and force a restart from byte 0 (ultra-fuzz Run 11 Storage
731 + /// HIGH). The shared completion path for both the server-driven
732 + /// [`Self::upload_multipart`] and the client-direct session flow.
733 + pub async fn complete_multipart_upload(
734 + &self,
735 + key: &str,
736 + upload_id: &str,
737 + parts: &[(i32, String)],
738 + ) -> Result<(), String> {
739 + if parts.is_empty() {
740 + return Err("cannot complete a multipart upload with no parts".to_string());
741 + }
742 + let mut parts = parts.to_vec();
743 + parts.sort_by_key(|(n, _)| *n);
744 + let completed_parts: Vec<CompletedPart> = parts
745 + .into_iter()
746 + .map(|(n, etag)| CompletedPart::builder().e_tag(etag).part_number(n).build())
747 + .collect();
748 +
749 + let completed = CompletedMultipartUpload::builder()
750 + .set_parts(Some(completed_parts))
751 + .build();
752 +
753 + let mut attempt: u32 = 0;
754 + loop {
755 + attempt += 1;
756 + match self
757 + .client
758 + .complete_multipart_upload()
759 + .bucket(&self.bucket)
760 + .key(key)
761 + .upload_id(upload_id)
762 + .multipart_upload(completed.clone())
763 + .send()
764 + .await
765 + {
766 + Ok(_) => return Ok(()),
767 + Err(e) if attempt < 3 => {
768 + let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
769 + tracing::warn!(
770 + attempt, delay_ms, error = ?e,
771 + "S3 complete_multipart_upload transient failure, retrying"
772 + );
773 + tokio::time::sleep(Duration::from_millis(delay_ms)).await;
774 + }
775 + Err(e) => {
776 + return Err(format!(
777 + "S3 complete multipart upload failed after retries: {e}"
778 + ));
779 + }
527 780 }
528 781 }
529 782 }
@@ -547,7 +800,7 @@ impl S3Client {
547 800 .map_err(|e| format!("Failed to open file for multipart upload: {e}"))?;
548 801
549 802 let mut part_number: i32 = 1;
550 - let mut completed_parts: Vec<CompletedPart> = Vec::new();
803 + let mut completed_parts: Vec<(i32, String)> = Vec::new();
551 804
552 805 loop {
553 806 // One owned buffer per part, frozen into `Bytes` so each retry attempt
@@ -616,12 +869,7 @@ impl S3Client {
616 869 let resp = resp
617 870 .map_err(|e| format!("S3 upload part {part_number} failed after retries: {e}"))?;
618 871 let etag = resp.e_tag().unwrap_or_default().to_string();
619 - completed_parts.push(
620 - CompletedPart::builder()
621 - .e_tag(etag)
622 - .part_number(part_number)
623 - .build(),
624 - );
872 + completed_parts.push((part_number, etag));
625 873
626 874 part_number += 1;
627 875 }
@@ -630,61 +878,145 @@ impl S3Client {
630 878 return Err("No parts uploaded (empty file)".to_string());
631 879 }
632 880
633 - let completed = CompletedMultipartUpload::builder()
634 - .set_parts(Some(completed_parts))
635 - .build();
636 -
637 - // Retry completion on transient failures, same as the part uploads.
638 - // Completion is the most expensive call to lose — every part is already
639 - // uploaded and paid for, so a single flake here would otherwise orphan
640 - // the whole multipart upload and force a restart from byte 0 (ultra-fuzz
641 - // Run 11 Storage HIGH).
642 - let mut attempt: u32 = 0;
643 - loop {
644 - attempt += 1;
645 - match self
646 - .client
647 - .complete_multipart_upload()
648 - .bucket(&self.bucket)
649 - .key(key)
650 - .upload_id(upload_id)
651 - .multipart_upload(completed.clone())
652 - .send()
653 - .await
654 - {
655 - Ok(_) => break,
656 - Err(e) if attempt < 3 => {
657 - let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
658 - tracing::warn!(
659 - attempt, delay_ms, error = ?e,
660 - "S3 complete_multipart_upload transient failure, retrying"
661 - );
662 - tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
663 - }
664 - Err(e) => {
665 - return Err(format!(
666 - "S3 complete multipart upload failed after retries: {e}"
667 - ));
668 - }
669 - }
670 - }
671 -
672 - Ok(())
881 + // Shared completion path (with its own transient retry). The caller owns
882 + // the single abort-on-Err, so a failure here still aborts exactly once.
883 + self.complete_multipart_upload(key, upload_id, &completed_parts).await
673 884 }
674 885
675 - /// Abort a multipart upload (best-effort, logs on failure).
676 - async fn abort_multipart_upload(&self, key: &str, upload_id: &str) {
677 - if let Err(e) = self
678 - .client
886 + /// Abort a multipart upload, releasing any uploaded parts.
887 + ///
888 + /// Public and `Result`-returning so the pending-upload reaper can abort an
889 + /// orphaned session and retry on failure. Internal callers that only want
890 + /// best-effort cleanup (e.g. [`Self::upload_multipart`] unwinding a failed
891 + /// transfer) log the error and move on.
892 + pub async fn abort_multipart_upload(
893 + &self,
894 + key: &str,
895 + upload_id: &str,
896 + ) -> Result<(), String> {
897 + self.client
679 898 .abort_multipart_upload()
680 899 .bucket(&self.bucket)
681 900 .key(key)
682 901 .upload_id(upload_id)
683 902 .send()
684 903 .await
904 + .map(|_| ())
905 + .map_err(|e| format!("S3 abort multipart upload for {key} failed: {e}"))
906 + }
907 +
908 + /// Server-side copy an object using multipart `UploadPartCopy`, for sources
909 + /// larger than the 5 GiB single-part [`Self::copy_object`]/`CopyObject`
910 + /// limit. No bytes transit the caller — S3 copies each byte range internally.
911 + ///
912 + /// This is the >5 GiB promote path: scan-then-promote lifts a Clean staging
913 + /// object to the served content key, and a single `CopyObject` caps at 5 GiB,
914 + /// so a large video would otherwise succeed at upload and fail at promote —
915 + /// the worst failure position. `src_size` is the source object's size (the
916 + /// caller already reads it at confirm); `part_size` defaults to an
917 + /// auto-chosen size via [`MultipartPlan::auto`]. `content_type` sets the
918 + /// destination's type, since a fresh multipart upload does not inherit the
919 + /// source's metadata the way `CopyObject` does.
920 + ///
921 + /// Abort lives in one place, as in [`Self::upload_multipart`]: any failure
922 + /// after the destination upload is created aborts it exactly once.
923 + pub async fn copy_object_multipart(
924 + &self,
925 + src_bucket: &str,
926 + src_key: &str,
927 + dst_key: &str,
928 + content_type: &str,
929 + src_size: u64,
930 + part_size: Option<usize>,
931 + ) -> Result<(), String> {
932 + // Pre-flight: plan before anything is created, so a bad size strands
933 + // nothing and makes no network call.
934 + let plan = match part_size {
935 + Some(ps) => MultipartPlan::new(src_size, ps)?,
936 + None => MultipartPlan::auto(src_size)?,
937 + };
938 +
939 + let upload_id = self.create_multipart_upload(dst_key, content_type).await?;
940 +
941 + match self
942 + .run_multipart_copy(src_bucket, src_key, dst_key, &plan, &upload_id)
943 + .await
685 944 {
686 - tracing::warn!("Failed to abort multipart upload for {key}: {e}");
945 + Ok(()) => Ok(()),
946 + Err(e) => {
947 + if let Err(abort_err) = self.abort_multipart_upload(dst_key, &upload_id).await {
948 + tracing::warn!("Failed to abort multipart copy for {dst_key}: {abort_err}");
949 + }
950 + Err(e)
951 + }
952 + }
953 + }
954 +
955 + /// Issue every `UploadPartCopy` for a ranged multipart copy, then complete
956 + /// it. Returns `Err` (without aborting) on any failure; the caller
957 + /// [`Self::copy_object_multipart`] owns the single abort.
958 + async fn run_multipart_copy(
959 + &self,
960 + src_bucket: &str,
961 + src_key: &str,
962 + dst_key: &str,
963 + plan: &MultipartPlan,
964 + upload_id: &str,
965 + ) -> Result<(), String> {
966 + // `CopySource` is `{bucket}/{key}`; keys are sanitized to
967 + // `[A-Za-z0-9._/-]` upstream, so no percent-encoding is needed (same as
968 + // `copy_object_from`).
969 + let copy_source = format!("{src_bucket}/{src_key}");
970 + let mut completed_parts: Vec<(i32, String)> = Vec::with_capacity(plan.part_count as usize);
971 +
972 + for part_number in 1..=plan.part_count {
973 + let (start, end) = plan
974 + .part_range(part_number)
975 + .ok_or_else(|| format!("internal: part {part_number} outside plan range"))?;
976 +
977 + // Retry transient copy failures, mirroring the part-upload path.
978 + let mut attempt: u32 = 0;
979 + let resp = loop {
980 + attempt += 1;
981 + match self
982 + .client
983 + .upload_part_copy()
984 + .bucket(&self.bucket)
985 + .key(dst_key)
986 + .upload_id(upload_id)
987 + .part_number(part_number as i32)
988 + .copy_source(&copy_source)
989 + .copy_source_range(format!("bytes={start}-{end}"))
990 + .send()
991 + .await
992 + {
993 + Ok(r) => break Ok(r),
994 + Err(e) if attempt < 3 => {
995 + let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
996 + tracing::warn!(
997 + part_number, attempt, delay_ms, error = ?e,
998 + "S3 upload_part_copy transient failure, retrying"
999 + );
1000 + tokio::time::sleep(Duration::from_millis(delay_ms)).await;
1001 + continue;
1002 + }
1003 + Err(e) => break Err(e),
1004 + }
1005 + };
1006 +
1007 + let resp = resp.map_err(|e| {
1008 + format!("S3 upload_part_copy part {part_number} failed after retries: {e}")
1009 + })?;
1010 + let etag = resp
1011 + .copy_part_result()
1012 + .and_then(|r| r.e_tag())
1013 + .unwrap_or_default()
1014 + .to_string();
1015 + completed_parts.push((part_number as i32, etag));
687 1016 }
1017 +
1018 + self.complete_multipart_upload(dst_key, upload_id, &completed_parts)
1019 + .await
688 1020 }
Lines truncated