Stream multipart assembly instead of buffering entire file in memory

CompleteMultipartUpload previously read all temp parts into a single Vec,
then wrote the assembled data via put_object (sequential 64KB writes). For
a 1GB file: 1GB memory allocation + 16,384 serial round-trips.

Now streams each part through a WalWriter: pipelined reads (64 per batch)
from the temp part → pipelined writes (64 per batch) to a WAL temp file →
atomic rename. Memory usage is bounded to one pipeline buffer (~4MB)
regardless of total file size.

Author: lukekim

src/s3/router.rs

@@ -1075,21 +1075,14 @@ async fn handle_complete_multipart_upload(
         }
     }
 
-    // Concatenate parts in order and write the final object
-    let mut final_data = Vec::new();
-    for pn in &part_numbers {
-        if let Some(part) = upload.parts.get(pn) {
-            match state.share.read_temp(&part.temp_path).await {
-                Ok(data) => final_data.extend_from_slice(&data),
-                Err(e) => {
-                    return io_to_s3_error(&e);
-                }
-            }
-        }
-    }
+    // Stream parts through a WAL writer (pipelined reads → pipelined writes
+    // → atomic rename). Never buffers the whole file in memory.
+    let temp_paths: Vec<&str> = part_numbers
+        .iter()
+        .filter_map(|pn| upload.parts.get(pn).map(|p| p.temp_path.as_str()))
+        .collect();
 
-    // Write the assembled object
-    let meta = match state.share.put_object(key, &final_data).await {
+    let meta = match state.share.assemble_parts(key, &temp_paths).await {
         Ok(m) => m,
         Err(e) => return io_to_s3_error(&e),
     };

src/smb/ops.rs

@@ -461,6 +461,61 @@ impl ShareSession {
         Ok(data)
     }
 
+    /// Assemble multipart upload parts into a single file via streaming.
+    ///
+    /// Reads each temp part using pipelined reads and writes through a WalWriter
+    /// (pipelined writes + atomic rename). Never holds more than one pipeline
+    /// buffer in memory — supports arbitrarily large files.
+    pub async fn assemble_parts(
+        &self,
+        key: &str,
+        temp_paths: &[&str],
+    ) -> io::Result<ObjectMeta> {
+        let mut wal = self.open_wal_write(key).await?;
+        let max_read = self.pool.max_read_size;
+
+        for &temp_path in temp_paths {
+            // Open the part file for reading on any pool connection
+            let (client, tree_id) = self.pick();
+            let cr = client
+                .create(
+                    tree_id,
+                    temp_path,
+                    DesiredAccess::GenericRead as u32,
+                    ShareAccess::All as u32,
+                    CreateDisposition::Open as u32,
+                    CreateOptions::NonDirectoryFile as u32,
+                )
+                .await?;
+
+            let file_size = cr.file_size;
+            let file_id = cr.file_id;
+
+            // Stream part data to the WalWriter using pipelined reads
+            let mut offset = 0u64;
+            while offset < file_size {
+                let remaining = file_size - offset;
+                let batch = remaining
+                    .div_ceil(max_read as u64)
+                    .min(PIPELINE_DEPTH as u64) as usize;
+                let chunks = client
+                    .pipelined_read(tree_id, &file_id, offset, max_read, batch)
+                    .await?;
+                if chunks.is_empty() {
+                    break;
+                }
+                for chunk in &chunks {
+                    wal.write(chunk).await?;
+                    offset += chunk.len() as u64;
+                }
+            }
+
+            let _ = client.close(tree_id, &file_id).await;
+        }
+
+        wal.commit(self).await
+    }
+
     /// Delete a temp file (best effort).
     pub async fn delete_temp(&self, smb_path: &str) {
         let (client, tree_id) = self.pick();