Skip to main content

max / makenotwork

server: expose client-direct multipart on StorageBackend Add copy_object_multipart and the four client-direct multipart session methods (create_multipart_upload, presign_upload_part, complete_multipart_upload, abort_multipart_upload) to the StorageBackend trait, forwarding through S3Client to the s3-storage primitives. All five are non-defaulted, matching the existing rationale for upload_multipart and copy_object: a silent default would mint a session no client could complete, or drop the abort that stops orphaned parts billing indefinitely. copy_object_multipart always takes src_bucket, collapsing the copy_object/copy_object_from pair into one method; a same-bucket promote passes the backend's own bucket. Implement both backends. InMemoryStorage does the multipart copy for real (the flat map has no 5 GiB limit, so promote observes the same result down either branch) and stubs the upload session the way presigned single PUTs already are, since its URLs are fake and no client bytes flow back. 1878 lib tests green, clippy clean (the one dead_code warning in discover.rs is pre-existing and unrelated). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 02:55 UTC
Commit: abd4f414c890b4bb96a77eb287eff7bd78b7f51d
Parent: 7774b6b
2 files changed, +149 insertions, -0 deletions
M server/src/storage.rs +105
@@ -506,6 +506,43 @@ pub trait StorageBackend: Send + Sync {
506 506 /// public backend with the private bucket name as `src_bucket`. Required
507 507 /// (not defaulted) for the same reason as `copy_object`.
508 508 async fn copy_object_from(&self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key) -> Result<()>;
509 + /// Server-side multipart copy (`UploadPartCopy`) for sources over the 5 GiB
510 + /// single-part `CopyObject` limit — the >5 GiB half of scan-then-promote.
511 + /// Always takes `src_bucket` explicitly, collapsing the
512 + /// `copy_object`/`copy_object_from` pair into one method (pass this
513 + /// backend's own bucket for a same-bucket promote). `content_type` sets the
514 + /// destination's type, since a fresh multipart upload does not inherit the
515 + /// source's metadata the way `CopyObject` does. Required (not defaulted) for
516 + /// the same reason as `copy_object`.
517 + async fn copy_object_multipart(&self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, content_type: &str, src_size: u64, part_size: Option<usize>) -> Result<()>;
518 +
519 + // -- Client-direct multipart sessions --
520 + //
521 + // The counterpart to `upload_multipart`, which drives a whole transfer
522 + // server-side from a local file. Here the server only mints the session and
523 + // the per-part presigned URLs; the client streams parts straight to S3, so
524 + // no object bytes transit the server. This is the path large CLI/desktop
525 + // uploads take (a browser stays on the single-PUT `presign_upload`).
526 + //
527 + // All four are required (not defaulted): a no-op default would mint a
528 + // session no client could complete, or silently drop the cleanup that
529 + // stops orphaned parts billing forever.
530 +
531 + /// Begin a client-direct multipart upload, returning the `upload_id` the
532 + /// part/complete/abort calls key on.
533 + async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result<String>;
534 + /// Presign an `UploadPart` request for one part (1-based `part_number`).
535 + /// `max_bytes`, when set, is signed as `Content-Length`, the same
536 + /// defense-in-depth as [`Self::presign_upload`] — the authoritative size
537 + /// check still happens at confirm time.
538 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String>;
539 + /// Complete a multipart upload from the collected `(part_number, etag)`
540 + /// pairs. Parts may be passed in any order; the backend sorts them.
541 + async fn complete_multipart_upload(&self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()>;
542 + /// Abort a multipart upload, releasing its uploaded parts. The
543 + /// pending-upload reaper calls this on sessions that were never confirmed —
544 + /// incomplete multipart uploads bill for their parts indefinitely.
545 + async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()>;
509 546 async fn check_connectivity(&self) -> std::result::Result<(), String>;
510 547 fn bucket(&self) -> &str;
511 548 }
@@ -918,6 +955,53 @@ impl S3Client {
918 955 .map_err(AppError::Storage)
919 956 }
920 957
958 + /// Server-side multipart copy for sources over the 5 GiB single-part
959 + /// `CopyObject` limit. See the [`StorageBackend::copy_object_multipart`]
960 + /// trait method for the promote rationale. `part_size` of `None` lets the
961 + /// storage layer auto-size parts for `src_size`.
962 + pub async fn copy_object_multipart(&self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, content_type: &str, src_size: u64, part_size: Option<usize>) -> Result<()> {
963 + self.inner
964 + .copy_object_multipart(src_bucket, src_key.as_str(), dst_key.as_str(), content_type, src_size, part_size)
965 + .await
966 + .map_err(AppError::Storage)
967 + }
968 +
969 + /// Begin a client-direct multipart upload. See the
970 + /// [`StorageBackend::create_multipart_upload`] trait method.
971 + pub async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result<String> {
972 + self.inner
973 + .create_multipart_upload(s3_key.as_str(), content_type)
974 + .await
975 + .map_err(AppError::Storage)
976 + }
977 +
978 + /// Presign one `UploadPart` request. See the
979 + /// [`StorageBackend::presign_upload_part`] trait method.
980 + pub async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String> {
981 + self.inner
982 + .presign_upload_part(s3_key.as_str(), upload_id, part_number, expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS), max_bytes)
983 + .await
984 + .map_err(AppError::Storage)
985 + }
986 +
987 + /// Complete a multipart upload. See the
988 + /// [`StorageBackend::complete_multipart_upload`] trait method.
989 + pub async fn complete_multipart_upload(&self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
990 + self.inner
991 + .complete_multipart_upload(s3_key.as_str(), upload_id, parts)
992 + .await
993 + .map_err(AppError::Storage)
994 + }
995 +
996 + /// Abort a multipart upload. See the
997 + /// [`StorageBackend::abort_multipart_upload`] trait method.
998 + pub async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
999 + self.inner
1000 + .abort_multipart_upload(s3_key.as_str(), upload_id)
1001 + .await
1002 + .map_err(AppError::Storage)
1003 + }
1004 +
921 1005 /// Lightweight connectivity check — issues a list with max_keys(0).
922 1006 pub async fn check_connectivity(&self) -> std::result::Result<(), String> {
923 1007 self.inner.check_connectivity().await
@@ -1145,6 +1229,27 @@ impl StorageBackend for S3Client {
1145 1229 self.upload_multipart(s3_key, content_type, file_path).await
1146 1230 }
1147 1231
1232 + async fn copy_object_multipart(&self, src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, content_type: &str, src_size: u64, part_size: Option<usize>) -> Result<()> {
1233 + // Inherent method; delegate.
1234 + self.copy_object_multipart(src_bucket, src_key, dst_key, content_type, src_size, part_size).await
1235 + }
1236 +
1237 + async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result<String> {
1238 + self.create_multipart_upload(s3_key, content_type).await
1239 + }
1240 +
1241 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String> {
1242 + self.presign_upload_part(s3_key, upload_id, part_number, expiry_secs, max_bytes).await
1243 + }
1244 +
1245 + async fn complete_multipart_upload(&self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
1246 + self.complete_multipart_upload(s3_key, upload_id, parts).await
1247 + }
1248 +
1249 + async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
1250 + self.abort_multipart_upload(s3_key, upload_id).await
1251 + }
1252 +
1148 1253 async fn check_connectivity(&self) -> std::result::Result<(), String> {
1149 1254 self.check_connectivity().await
1150 1255 }
M server/tests/harness/storage.rs +44
@@ -114,6 +114,50 @@ impl StorageBackend for InMemoryStorage {
114 114 self.copy_object(src_key, dst_key).await
115 115 }
116 116
117 + async fn copy_object_multipart(&self, _src_bucket: &str, src_key: &S3Key, dst_key: &S3Key, _content_type: &str, _src_size: u64, _part_size: Option<usize>) -> Result<()> {
118 + // The flat map has no 5 GiB single-copy limit, so the multipart promote
119 + // is the same copy. Implemented for real (not stubbed) so a promote test
120 + // gets the same observable result down either branch.
121 + self.copy_object(src_key, dst_key).await
122 + }
123 +
124 + // -- Client-direct multipart sessions --
125 + //
126 + // Stubbed the same way presigned single-PUT uploads already are: the URLs
127 + // this backend hands out are fake, so no client bytes can flow back into the
128 + // map. Tests simulate the finished object with `put()` and then exercise the
129 + // confirm path, exactly as they do for `presign_upload` + confirm.
130 +
131 + async fn create_multipart_upload(&self, s3_key: &S3Key, _content_type: &str) -> Result<String> {
132 + Ok(format!("test-upload-id/{}", s3_key))
133 + }
134 +
135 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, _expiry_secs: Option<u64>, _max_bytes: Option<i64>) -> Result<String> {
136 + // Mirror the production range check so a bad part number fails in tests
137 + // the same way it would against S3.
138 + if !(1..=s3_storage::MULTIPART_MAX_PARTS as i32).contains(&part_number) {
139 + return Err(AppError::Storage(format!(
140 + "part number {part_number} out of range 1..={}",
141 + s3_storage::MULTIPART_MAX_PARTS
142 + )));
143 + }
144 + Ok(format!("http://test-storage/{s3_key}?uploadId={upload_id}&partNumber={part_number}"))
145 + }
146 +
147 + async fn complete_multipart_upload(&self, _s3_key: &S3Key, _upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
148 + if parts.is_empty() {
149 + return Err(AppError::Storage(
150 + "cannot complete a multipart upload with no parts".to_string(),
151 + ));
152 + }
153 + Ok(())
154 + }
155 +
156 + async fn abort_multipart_upload(&self, _s3_key: &S3Key, _upload_id: &str) -> Result<()> {
157 + // Idempotent: nothing is staged, so aborting an unknown session is fine.
158 + Ok(())
159 + }
160 +
117 161 async fn check_connectivity(&self) -> std::result::Result<(), String> {
118 162 Ok(())
119 163 }