feat(services/swift): support SLO multipart upload via MultipartWrite

benroeder · benroeder · commit 8c12ed609bc1 · 2026-02-22T17:12:36.000-08:00
Implement oio::MultipartWrite for SwiftWriter using Swift's Static
Large Object (SLO) mechanism, removing the 5GB single-upload ceiling.

SLO flow:
- initiate_part: generate a local UUID (no server call needed)
- write_part: PUT segment to .segments/{path}/{upload_id}/{part:08}
- complete_part: PUT JSON manifest to {path}?multipart-manifest=put
- abort_part: list and delete all segments under the upload_id prefix

Changes:
- Add swift_put_segment, swift_put_slo_manifest, swift_delete_slo,
  slo_segment_path to SwiftCore
- Add SloManifestEntry serde struct for manifest JSON
- Replace OneShotWrite with MultipartWrite on SwiftWriter
- Track per-part sizes in Arc&lt;Mutex&lt;HashMap&gt;&gt; for manifest assembly
- Change Writer type from OneShotWriter to MultipartWriter
- Declare write_can_multi, write_multi_min_size (5MB),
  write_multi_max_size (5GB) capabilities
- Add uuid dependency for upload ID generation
diff --git a/core/Cargo.lock b/core/Cargo.lock
diff --git a/core/core/src/raw/oio/write/multipart_write.rs b/core/core/src/raw/oio/write/multipart_write.rs
@@ -118,6 +118,8 @@ pub struct MultipartPart {
     pub etag: String,
     /// The checksum of the part.
     pub checksum: Option<String>,
+    /// The size of the part in bytes.
+    pub size: Option<u64>,
 }
 
 struct WriteInput<W: MultipartWrite> {
@@ -391,6 +393,7 @@ mod tests {
                 part_number,
                 etag: "etag".to_string(),
                 checksum: None,
+                size: None,
             })
         }
 
diff --git a/core/services/b2/src/writer.rs b/core/services/b2/src/writer.rs
@@ -133,6 +133,7 @@ impl oio::MultipartWrite for B2Writer {
                     etag: result.content_sha1,
                     part_number,
                     checksum: None,
+                    size: None,
                 })
             }
             _ => Err(parse_error(resp)),
diff --git a/core/services/cos/src/writer.rs b/core/services/cos/src/writer.rs
@@ -141,6 +141,7 @@ impl oio::MultipartWrite for CosWriter {
                     part_number,
                     etag,
                     checksum: None,
+                    size: None,
                 })
             }
             _ => Err(parse_error(resp)),
diff --git a/core/services/gcs/src/writer.rs b/core/services/gcs/src/writer.rs
@@ -119,6 +119,7 @@ impl oio::MultipartWrite for GcsWriter {
             part_number,
             etag,
             checksum: None,
+                    size: None,
         })
     }
 
diff --git a/core/services/obs/src/writer.rs b/core/services/obs/src/writer.rs
@@ -136,6 +136,7 @@ impl oio::MultipartWrite for ObsWriter {
                     part_number,
                     etag,
                     checksum: None,
+                    size: None,
                 })
             }
             _ => Err(parse_error(resp)),
diff --git a/core/services/oss/src/writer.rs b/core/services/oss/src/writer.rs
@@ -138,6 +138,7 @@ impl oio::MultipartWrite for OssWriter {
                     part_number,
                     etag,
                     checksum: None,
+                    size: None,
                 })
             }
             _ => Err(parse_error(resp)),
diff --git a/core/services/s3/src/writer.rs b/core/services/s3/src/writer.rs
@@ -143,6 +143,7 @@ impl oio::MultipartWrite for S3Writer {
                     part_number,
                     etag,
                     checksum,
+                    size: None,
                 })
             }
             _ => Err(parse_error(resp)),
diff --git a/core/services/swift/Cargo.toml b/core/services/swift/Cargo.toml
@@ -38,6 +38,7 @@ opendal-core = { path = "../../core", version = "0.55.0", default-features = fal
 quick-xml = { workspace = true, features = ["serialize", "overlapped-lists"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
+uuid = { workspace = true, features = ["v4"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/core/services/swift/src/backend.rs b/core/services/swift/src/backend.rs
@@ -157,6 +157,13 @@ impl Builder for SwiftBuilder {
 
                             write: true,
                             write_can_empty: true,
+                            write_can_multi: true,
+                            write_multi_min_size: Some(5 * 1024 * 1024),
+                            write_multi_max_size: if cfg!(target_pointer_width = "64") {
+                                Some(5 * 1024 * 1024 * 1024)
+                            } else {
+                                Some(usize::MAX)
+                            },
                             write_with_content_type: true,
                             write_with_content_disposition: true,
                             write_with_content_encoding: true,
@@ -194,7 +201,7 @@ pub struct SwiftBackend {
 
 impl Access for SwiftBackend {
     type Reader = HttpBody;
-    type Writer = oio::OneShotWriter<SwiftWriter>;
+    type Writer = oio::MultipartWriter<SwiftWriter>;
     type Lister = oio::PageLister<SwiftLister>;
     type Deleter = oio::BatchDeleter<SwiftDeleter>;
 
@@ -236,9 +243,9 @@ impl Access for SwiftBackend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let concurrent = args.concurrent();
         let writer = SwiftWriter::new(self.core.clone(), args.clone(), path.to_string());
-
-        let w = oio::OneShotWriter::new(writer);
+        let w = oio::MultipartWriter::new(self.core.info.clone(), writer, concurrent);
 
         Ok((RpWrite::default(), w))
     }
diff --git a/core/services/swift/src/core.rs b/core/services/swift/src/core.rs
@@ -26,6 +26,7 @@ use http::header::IF_MODIFIED_SINCE;
 use http::header::IF_NONE_MATCH;
 use http::header::IF_UNMODIFIED_SINCE;
 use serde::Deserialize;
+use serde::Serialize;
 
 use opendal_core::raw::*;
 use opendal_core::*;
@@ -313,6 +314,153 @@ impl SwiftCore {
 
         self.info.http_client().send(req).await
     }
+
+    /// Build the segment path for an SLO part.
+    ///
+    /// Segments are stored as: `.segments/{object_path}/{upload_id}/{part_number:08}`
+    pub fn slo_segment_path(&self, path: &str, upload_id: &str, part_number: usize) -> String {
+        let abs = build_abs_path(&self.root, path);
+        format!(
+            ".segments/{}{}/{:08}",
+            abs.trim_end_matches('/'),
+            upload_id,
+            part_number
+        )
+    }
+
+    /// Upload a segment for an SLO multipart upload.
+    ///
+    /// Reference: <https://docs.openstack.org/swift/latest/overview_large_objects.html>
+    pub async fn swift_put_segment(
+        &self,
+        path: &str,
+        upload_id: &str,
+        part_number: usize,
+        size: u64,
+        body: Buffer,
+    ) -> Result<Response<Buffer>> {
+        let segment = self.slo_segment_path(path, upload_id, part_number);
+        let url = format!(
+            "{}/{}/{}",
+            &self.endpoint,
+            &self.container,
+            percent_encode_path(&segment)
+        );
+
+        let mut req = Request::put(&url);
+        req = req.header("X-Auth-Token", &self.token);
+        req = req.header(header::CONTENT_LENGTH, size);
+
+        let req = req
+            .extension(Operation::Write)
+            .body(body)
+            .map_err(new_request_build_error)?;
+
+        self.info.http_client().send(req).await
+    }
+
+    /// Finalize an SLO by uploading the manifest.
+    ///
+    /// PUT {container}/{path}?multipart-manifest=put with a JSON body listing
+    /// each segment's path, etag, and size.
+    ///
+    /// Reference: <https://docs.openstack.org/swift/latest/overview_large_objects.html>
+    pub async fn swift_put_slo_manifest(
+        &self,
+        path: &str,
+        manifest: &[SloManifestEntry],
+        args: &OpWrite,
+    ) -> Result<Response<Buffer>> {
+        let abs = build_abs_path(&self.root, path);
+        let url = format!(
+            "{}/{}/{}?multipart-manifest=put",
+            &self.endpoint,
+            &self.container,
+            percent_encode_path(&abs)
+        );
+
+        let body = serde_json::to_vec(manifest).map_err(new_json_serialize_error)?;
+
+        let mut req = Request::put(&url);
+        req = req.header("X-Auth-Token", &self.token);
+        req = req.header(header::CONTENT_LENGTH, body.len());
+        req = req.header(header::CONTENT_TYPE, "application/json");
+
+        // Forward user metadata to the manifest object.
+        if let Some(user_metadata) = args.user_metadata() {
+            for (k, v) in user_metadata {
+                req = req.header(format!("X-Object-Meta-{k}"), v);
+            }
+        }
+
+        let req = req
+            .extension(Operation::Write)
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+
+        self.info.http_client().send(req).await
+    }
+
+    /// Delete an SLO manifest and all its segments.
+    ///
+    /// DELETE {container}/{path}?multipart-manifest=delete removes the manifest
+    /// and all referenced segments in one call.
+    ///
+    /// Reference: <https://docs.openstack.org/swift/latest/overview_large_objects.html>
+    pub async fn swift_delete_slo(&self, path: &str, upload_id: &str) -> Result<()> {
+        // List segments under the upload_id prefix and delete them individually.
+        // We can't use multipart-manifest=delete because we haven't created
+        // the manifest yet (abort happens before complete).
+        let abs = build_abs_path(&self.root, path);
+        let prefix = format!(".segments/{}{}/", abs.trim_end_matches('/'), upload_id);
+
+        // List all segments with this prefix.
+        let url = QueryPairsWriter::new(&format!("{}/{}/", &self.endpoint, &self.container))
+            .push("prefix", &percent_encode_path(&prefix))
+            .push("format", "json")
+            .finish();
+
+        let mut req = Request::get(&url);
+        req = req.header("X-Auth-Token", &self.token);
+
+        let req = req
+            .extension(Operation::List)
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+
+        let resp = self.info.http_client().send(req).await?;
+        if !resp.status().is_success() {
+            return Ok(());
+        }
+
+        let bs = resp.into_body().to_bytes();
+        let segments: Vec<ListOpResponse> = serde_json::from_slice(&bs).unwrap_or_default();
+
+        // Delete each segment.
+        for seg in segments {
+            if let ListOpResponse::FileInfo { name, .. } = seg {
+                let seg_url = format!(
+                    "{}/{}/{}",
+                    &self.endpoint,
+                    &self.container,
+                    percent_encode_path(&name)
+                );
+
+                let mut req = Request::delete(&seg_url);
+                req = req.header("X-Auth-Token", &self.token);
+
+                let req = req
+                    .extension(Operation::Delete)
+                    .body(Buffer::new())
+                    .map_err(new_request_build_error)?;
+
+                // Best effort — ignore individual segment delete failures.
+                let _ = self.info.http_client().send(req).await;
+            }
+        }
+
+        Ok(())
+    }
 }
 
 #[derive(Debug, Eq, PartialEq, Deserialize)]
@@ -354,6 +502,19 @@ pub struct BulkDeleteResponse {
     pub response_body: Option<String>,
 }
 
+/// Entry in an SLO manifest JSON array.
+///
+/// Reference: <https://docs.openstack.org/swift/latest/overview_large_objects.html>
+#[derive(Debug, Serialize)]
+pub struct SloManifestEntry {
+    /// Path to the segment: `/{container}/{segment_name}`
+    pub path: String,
+    /// MD5 etag of the segment (without quotes).
+    pub etag: String,
+    /// Size of the segment in bytes.
+    pub size_bytes: u64,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/core/services/swift/src/writer.rs b/core/services/swift/src/writer.rs
diff --git a/core/services/upyun/src/writer.rs b/core/services/upyun/src/writer.rs
diff --git a/core/services/vercel-blob/src/writer.rs b/core/services/vercel-blob/src/writer.rs

Original file line number	Diff line number	Diff line change
`@@ -118,6 +118,8 @@ pub struct MultipartPart {`
`118`	`118`	`pub etag: String,`
`119`	`119`	`/// The checksum of the part.`
`120`	`120`	`pub checksum: Option<String>,`
	`121`	`+ /// The size of the part in bytes.`
	`122`	`+ pub size: Option<u64>,`
`121`	`123`	`}`
`122`	`124`
`123`	`125`	`struct WriteInput<W: MultipartWrite> {`
`@@ -391,6 +393,7 @@ mod tests {`
`391`	`393`	`part_number,`
`392`	`394`	`etag: "etag".to_string(),`
`393`	`395`	`checksum: None,`
	`396`	`+ size: None,`
`394`	`397`	`})`
`395`	`398`	`}`
`396`	`399`
Original file line number	Diff line number	Diff line change
`@@ -133,6 +133,7 @@ impl oio::MultipartWrite for B2Writer {`
`133`	`133`	`etag: result.content_sha1,`
`134`	`134`	`part_number,`
`135`	`135`	`checksum: None,`
	`136`	`+ size: None,`
`136`	`137`	`})`
`137`	`138`	`}`
`138`	`139`	`_ => Err(parse_error(resp)),`
Original file line number	Diff line number	Diff line change
`@@ -141,6 +141,7 @@ impl oio::MultipartWrite for CosWriter {`
`141`	`141`	`part_number,`
`142`	`142`	`etag,`
`143`	`143`	`checksum: None,`
	`144`	`+ size: None,`
`144`	`145`	`})`
`145`	`146`	`}`
`146`	`147`	`_ => Err(parse_error(resp)),`
Original file line number	Diff line number	Diff line change
`@@ -119,6 +119,7 @@ impl oio::MultipartWrite for GcsWriter {`
`119`	`119`	`part_number,`
`120`	`120`	`etag,`
`121`	`121`	`checksum: None,`
	`122`	`+ size: None,`
`122`	`123`	`})`
`123`	`124`	`}`
`124`	`125`
Original file line number	Diff line number	Diff line change
`@@ -136,6 +136,7 @@ impl oio::MultipartWrite for ObsWriter {`
`136`	`136`	`part_number,`
`137`	`137`	`etag,`
`138`	`138`	`checksum: None,`
	`139`	`+ size: None,`
`139`	`140`	`})`
`140`	`141`	`}`
`141`	`142`	`_ => Err(parse_error(resp)),`
Original file line number	Diff line number	Diff line change
`@@ -138,6 +138,7 @@ impl oio::MultipartWrite for OssWriter {`
`138`	`138`	`part_number,`
`139`	`139`	`etag,`
`140`	`140`	`checksum: None,`
	`141`	`+ size: None,`
`141`	`142`	`})`
`142`	`143`	`}`
`143`	`144`	`_ => Err(parse_error(resp)),`
Original file line number	Diff line number	Diff line change
`@@ -143,6 +143,7 @@ impl oio::MultipartWrite for S3Writer {`
`143`	`143`	`part_number,`
`144`	`144`	`etag,`
`145`	`145`	`checksum,`
	`146`	`+ size: None,`
`146`	`147`	`})`
`147`	`148`	`}`
`148`	`149`	`_ => Err(parse_error(resp)),`