Skip to main content

max / makenotwork

Add blob_upload_streaming to synckit-client The bounded-memory upload path, and the only one the server accepts above its one-shot PUT ceiling. Peak memory is one part plus one chunk regardless of blob size, where blob_upload buffers the whole ciphertext. It declares the ciphertext length up front (blob_encrypted_len of the file size), so the server can sign every part boundary before a byte is read. It then seals 1 MiB v3 chunks straight out of the file, drains a full part the moment one is complete, and PUTs it to a URL signed for that exact Content-Length. Part URLs are pulled 100 at a time as the upload progresses. The SHA-256 is recomputed off the same read: the caller hashed the file in an earlier pass, and a file edited in between would otherwise land under an address that does not describe it, which every later download would re-hash and reject. Any failure past start aborts the session, since uploaded parts bill until released. Not resumable across a process restart. Persisting upload_id plus the completed part ETags is a consumer-facing API decision, left open. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 14:52 UTC
Commit: 99d418bcbd59a68ec3e055bfea0336ade0964a06
Parent: 7b530ac
6 files changed, +725 insertions, -13 deletions
M shared/synckit-client/Cargo.toml +1 -1
@@ -33,7 +33,7 @@ sha2 = "0.10"
33 33 # HTTP
34 34 reqwest = { version = "0.13", features = ["json", "native-tls", "stream", "form"] }
35 35 bytes = "1"
36 - tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
36 + tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "fs", "io-util"] }
37 37 tokio-stream = "0.1"
38 38
39 39 # Serialization
M shared/synckit-client/docs/architecture.md +22
@@ -251,6 +251,28 @@ base64) to avoid the ~33% base64 overhead on large files. v2 overhead is 44 byte
251 251 per blob: a 4-byte `sk2:` wire tag + 24-byte XChaCha20 nonce + 16-byte Poly1305
252 252 tag.
253 253
254 + ### Large blobs: the multipart path
255 +
256 + Steps 1 and 2 buffer the whole ciphertext, and the server refuses a one-shot PUT
257 + above its own ceiling. For a blob already on disk, `blob_upload_streaming(hash,
258 + path)` replaces both with a multipart session and holds one part at a time:
259 +
260 + 1. It computes the exact ciphertext length from the file size alone
261 + (`blob_encrypted_len`) and declares it to `blobs/multipart/start`, which
262 + returns the part geometry both sides then derive identically.
263 + 2. It seals 1 MiB v3 chunks straight out of the file, staging them until a full
264 + part is ready, and PUTs each part to a presigned URL signed for that exact
265 + `Content-Length`. Presigned part URLs are pulled 100 at a time as the upload
266 + progresses.
267 + 3. It recomputes the SHA-256 as it reads, so a file that changed since the
268 + caller hashed it fails rather than landing under a stale content address.
269 + 4. Any failure past `start` aborts the session, since uploaded parts bill until
270 + released. `blob_confirm` is still the caller's next step, unchanged.
271 +
272 + It works at any size (a small blob is a single part), so prefer it whenever the
273 + blob is file-backed. Not yet resumable across a process restart: the session id
274 + is not persisted.
275 +
254 276 ## Keychain integration
255 277
256 278 The `keychain` feature (default on) uses the `keyring` crate (v3) to store
M shared/synckit-client/src/client/blob.rs +320 -12
@@ -5,8 +5,12 @@
5 5 //! downloads are re-hashed on arrival and discarded on any mismatch, so a
6 6 //! server that swaps or rolls back a blob cannot slip it past the client.
7 7
8 + use std::collections::VecDeque;
9 + use std::path::Path;
10 +
8 11 use bytes::{Buf, Bytes, BytesMut};
9 12 use sha2::{Digest, Sha256};
13 + use tokio::io::AsyncReadExt;
10 14 use tokio_stream::StreamExt;
11 15 use tracing::instrument;
12 16
@@ -24,6 +28,13 @@ use super::helpers::{check_response, Idempotency};
24 28 /// Generous (4 GiB) so legitimate large media still flow.
25 29 const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;
26 30
31 + /// How many presigned part URLs [`SyncKitClient::blob_upload_streaming`] pulls
32 + /// per request. Must not exceed the server's own window bound
33 + /// (`BLOB_MULTIPART_PART_URL_WINDOW`), which rejects a larger `count` outright.
34 + /// Pulling as the upload progresses keeps the number of live upload credentials
35 + /// small for a transfer that may never finish.
36 + const PART_URL_WINDOW: u32 = 100;
37 +
27 38 impl SyncKitClient {
28 39 /// Request a presigned upload URL for a blob.
29 40 /// Returns (upload_url, already_exists). If `already_exists` is true,
@@ -69,16 +80,11 @@ impl SyncKitClient {
69 80 /// content address. The plaintext is never sent to the server, preserving
70 81 /// the E2E encryption guarantee.
71 82 ///
72 - /// The input is capped at [`MAX_BLOB_BYTES`]: without it, a caller (or a
73 - /// hostile local integration) could hand in an arbitrarily large `data`,
74 - /// which this method would encrypt into an equally large ciphertext and
75 - /// buffer whole — an unbounded memory (OOM) lever. Bounded-memory uploads of
76 - /// blobs *larger* than the cap need chunk-at-a-time streaming to S3, which in
77 - /// turn needs multipart upload (a single presigned `PUT` requires a known,
78 - /// signed `Content-Length`, and `reqwest`'s streaming body sends chunked with
79 - /// none). That multipart path is a separate, deferred server+client feature;
80 - /// until it lands, blobs above the cap are rejected here rather than risking
81 - /// the OOM.
83 + /// This is the in-memory path: it buffers the whole ciphertext, so the input
84 + /// is capped at [`MAX_BLOB_BYTES`] rather than risking an unbounded
85 + /// allocation. For anything file-backed, prefer
86 + /// [`Self::blob_upload_streaming`], which holds one part at a time and is
87 + /// the only path the server accepts above its one-shot PUT ceiling.
82 88 #[instrument(skip(self, presigned_url, data))]
83 89 pub async fn blob_upload(&self, hash: &str, presigned_url: &str, data: Vec<u8>) -> Result<()> {
84 90 if data.len() > MAX_BLOB_BYTES {
@@ -94,8 +100,8 @@ impl SyncKitClient {
94 100 // Release the plaintext before the (potentially long, multi-GB) network
95 101 // send so the sustained peak during transfer is ~1x the blob, not ~2x.
96 102 // The per-attempt `encrypted.clone()` below is a cheap `Bytes` refcount
97 - // bump, not a copy. True 1x streaming (never materializing the full
98 - // ciphertext) arrives with the resumable multipart upload in a later pass.
103 + // bump, not a copy. Never materializing the full ciphertext at all is
104 + // what `blob_upload_streaming` does, for callers holding a file.
99 105 drop(data);
100 106
101 107 self.retry_request(Idempotency::IdempotentWrite { on: "content hash (S3 key is content-addressed)" }, || {
@@ -111,6 +117,308 @@ impl SyncKitClient {
111 117 Ok(())
112 118 }
113 119
120 + /// Upload a file-backed blob through the multipart session, never holding
121 + /// more than one part in memory.
122 + ///
123 + /// This is the bounded-memory counterpart to [`Self::blob_upload`], and the
124 + /// only path for blobs above the server's one-shot PUT ceiling. It works at
125 + /// any size (a small blob is a single part), so it is the better default
126 + /// whenever the blob is already a file on disk.
127 + ///
128 + /// The v3 chunk geometry is what makes it possible. A presigned PUT needs a
129 + /// signed `Content-Length` up front, and `blob_encrypted_len` gives the
130 + /// exact ciphertext length from the plaintext length alone — so the server
131 + /// signs every part boundary before a single byte has been read, and this
132 + /// method fills those boundaries by sealing 1 MiB chunks as it streams the
133 + /// file. Peak memory is one part plus one chunk, not the whole blob.
134 + ///
135 + /// `hash` must be the SHA-256 of the file's plaintext, as elsewhere. It is
136 + /// recomputed while streaming and the upload is abandoned on mismatch: the
137 + /// caller hashed the file in an earlier pass, so a file edited in between
138 + /// would otherwise be stored under an address that does not describe it.
139 + ///
140 + /// The caller still calls [`Self::blob_confirm`] afterwards, exactly as with
141 + /// the one-shot path — this method replaces the transport only.
142 + #[instrument(skip(self, path), fields(path = %path.display()))]
143 + pub async fn blob_upload_streaming(&self, hash: &str, path: &Path) -> Result<()> {
144 + let master_key = self.require_master_key()?;
145 +
146 + let meta = tokio::fs::metadata(path).await.map_err(|e| {
147 + SyncKitError::Internal(format!("stat blob file {}: {e}", path.display()))
148 + })?;
149 + let plaintext_len = usize::try_from(meta.len())
150 + .map_err(|_| SyncKitError::InvalidArgument("blob file is larger than usize".into()))?;
151 + if plaintext_len > MAX_BLOB_BYTES {
152 + return Err(SyncKitError::InvalidArgument(format!(
153 + "blob is {plaintext_len} bytes, over the {MAX_BLOB_BYTES}-byte client cap"
154 + )));
155 + }
156 + // The session is sized in ciphertext, which is knowable from the
157 + // plaintext length alone — that is the whole trick.
158 + let size_bytes = i64::try_from(crypto::blob_encrypted_len(plaintext_len))
159 + .map_err(|_| SyncKitError::InvalidArgument("blob ciphertext exceeds i64".into()))?;
160 +
161 + let token = self.require_token()?;
162 + let start_body = Bytes::from(serde_json::to_vec(&BlobMultipartStartRequest {
163 + hash: hash.to_string(),
164 + size_bytes,
165 + })?);
166 + let start: BlobMultipartStartResponse = self
167 + .retry_request_json(Idempotency::ReadOnly, || {
168 + let req = self
169 + .http
170 + .post(&self.endpoints.blobs_multipart_start)
171 + .bearer_auth(&token)
172 + .header("content-type", "application/json")
173 + .body(start_body.clone());
174 + async move { check_response(req.send().await?).await }
175 + })
176 + .await?;
177 +
178 + if start.already_exists {
179 + return Ok(());
180 + }
181 +
182 + // Every failure past this point leaves uploaded parts that S3 bills for
183 + // until the session is aborted, so the abort lives in exactly one place:
184 + // here, around the whole transfer. The server-side orphan reaper is the
185 + // backstop for a client that dies outright.
186 + let transfer = self
187 + .stream_blob_parts(hash, &start, path, plaintext_len, size_bytes, &master_key)
188 + .await;
189 + let parts = match transfer {
190 + Ok(parts) => parts,
191 + Err(e) => {
192 + if let Err(abort_err) = self.blob_multipart_abort(hash, &start.upload_id).await {
193 + tracing::warn!(
194 + error = %abort_err,
195 + "failed to abort multipart blob session; the server reaper will release it"
196 + );
197 + }
198 + return Err(e);
199 + }
200 + };
201 +
202 + let complete_body = Bytes::from(serde_json::to_vec(&BlobMultipartCompleteRequest {
203 + hash: hash.to_string(),
204 + upload_id: start.upload_id.clone(),
205 + parts,
206 + })?);
207 + self.retry_request(
208 + Idempotency::IdempotentWrite { on: "upload_id (completing twice assembles the same object)" },
209 + || {
210 + let req = self
211 + .http
212 + .post(&self.endpoints.blobs_multipart_complete)
213 + .bearer_auth(&token)
214 + .header("content-type", "application/json")
215 + .body(complete_body.clone());
216 + async move { check_response(req.send().await?).await }
217 + },
218 + )
219 + .await?;
220 +
221 + Ok(())
222 + }
223 +
224 + /// Seal the file chunk by chunk, cutting the ciphertext at the server's
225 + /// signed part boundaries and PUTting each part. Returns the completed
226 + /// `(part_number, etag)` pairs.
227 + ///
228 + /// Peak memory is one part plus one chunk: the staging buffer never holds
229 + /// more than a part's worth, because a full part is drained and sent the
230 + /// moment it is complete.
231 + async fn stream_blob_parts(
232 + &self,
233 + hash: &str,
234 + start: &BlobMultipartStartResponse,
235 + path: &Path,
236 + plaintext_len: usize,
237 + size_bytes: i64,
238 + master_key: &[u8; 32],
239 + ) -> Result<Vec<BlobMultipartCompletedPart>> {
240 + let part_size = usize::try_from(start.part_size).map_err(|_| {
241 + SyncKitError::Internal(format!("server part_size {} exceeds usize", start.part_size))
242 + })?;
243 + if part_size == 0 || start.part_count == 0 {
244 + return Err(SyncKitError::Internal(
245 + "server returned an empty multipart plan".into(),
246 + ));
247 + }
248 +
249 + let mut file = tokio::fs::File::open(path).await.map_err(|e| {
250 + SyncKitError::Internal(format!("open blob file {}: {e}", path.display()))
251 + })?;
252 +
253 + let chunk_count = crypto::blob_chunk_count_for(plaintext_len);
254 + let mut hasher = Sha256::new();
255 + let mut plain = vec![0u8; crypto::BLOB_CHUNK_SIZE];
256 + let mut staged = BytesMut::with_capacity(part_size + crypto::BLOB_CHUNK_SIZE);
257 + staged.extend_from_slice(&crypto::blob_header_bytes(plaintext_len));
258 +
259 + let mut urls: VecDeque<BlobMultipartPartUrl> = VecDeque::new();
260 + let mut completed = Vec::with_capacity(start.part_count as usize);
261 + let mut next_part = 1u32;
262 +
263 + for index in 0..chunk_count {
264 + let offset = index as usize * crypto::BLOB_CHUNK_SIZE;
265 + let want = (plaintext_len - offset).min(crypto::BLOB_CHUNK_SIZE);
266 + file.read_exact(&mut plain[..want]).await.map_err(|e| {
267 + SyncKitError::Internal(format!(
268 + "read blob file {} at offset {offset}: {e}",
269 + path.display()
270 + ))
271 + })?;
272 + hasher.update(&plain[..want]);
273 + staged.extend_from_slice(&crypto::seal_blob_chunk(
274 + &plain[..want],
275 + master_key,
276 + hash,
277 + index,
278 + chunk_count,
279 + )?);
280 +
281 + // Every part but the last is exactly `part_size`; the remainder is
282 + // the final part, so it is never sent from inside this loop.
283 + while staged.len() >= part_size && next_part < start.part_count {
284 + let body = staged.split_to(part_size).freeze();
285 + let etag = self
286 + .upload_blob_part(hash, start, size_bytes, next_part, &mut urls, body)
287 + .await?;
288 + completed.push(BlobMultipartCompletedPart {
289 + part_number: next_part as i32,
290 + etag,
291 + });
292 + next_part += 1;
293 + }
294 + }
295 +
296 + // Verify the content address before the object can be assembled. A
297 + // mismatch here aborts the session, so a file that changed after the
298 + // caller hashed it is never stored under the stale address.
299 + let actual = format!("{:x}", hasher.finalize());
300 + if actual != hash {
301 + return Err(SyncKitError::IntegrityFailed {
302 + expected: hash.to_string(),
303 + actual,
304 + });
305 + }
306 +
307 + if next_part != start.part_count {
308 + return Err(SyncKitError::Internal(format!(
309 + "sealed stream produced {} parts, server planned {}",
310 + next_part, start.part_count
311 + )));
312 + }
313 + let body = staged.split().freeze();
314 + let etag = self
315 + .upload_blob_part(hash, start, size_bytes, next_part, &mut urls, body)
316 + .await?;
317 + completed.push(BlobMultipartCompletedPart {
318 + part_number: next_part as i32,
319 + etag,
320 + });
321 +
322 + Ok(completed)
323 + }
324 +
325 + /// PUT one part to its presigned URL, refilling the URL window as needed,
326 + /// and return the ETag S3 assigned it.
327 + async fn upload_blob_part(
328 + &self,
329 + hash: &str,
330 + start: &BlobMultipartStartResponse,
331 + size_bytes: i64,
332 + part_number: u32,
333 + urls: &mut VecDeque<BlobMultipartPartUrl>,
334 + body: Bytes,
335 + ) -> Result<String> {
336 + if urls.is_empty() {
337 + let token = self.require_token()?;
338 + let req_body = Bytes::from(serde_json::to_vec(&BlobMultipartPartsRequest {
339 + hash: hash.to_string(),
340 + upload_id: start.upload_id.clone(),
341 + size_bytes,
342 + first_part: part_number,
343 + count: PART_URL_WINDOW.min(start.part_count - part_number + 1),
344 + })?);
345 + let window: BlobMultipartPartsResponse = self
346 + .retry_request_json(Idempotency::ReadOnly, || {
347 + let req = self
348 + .http
349 + .post(&self.endpoints.blobs_multipart_parts)
350 + .bearer_auth(&token)
351 + .header("content-type", "application/json")
352 + .body(req_body.clone());
353 + async move { check_response(req.send().await?).await }
354 + })
355 + .await?;
356 + urls.extend(window.parts);
357 + }
358 +
359 + let part = urls.pop_front().ok_or_else(|| {
360 + SyncKitError::Internal(format!("server returned no URL for part {part_number}"))
361 + })?;
362 + // The two sides derive the plan independently, so a disagreement means
363 + // one of them is wrong about the geometry. Sending anyway would fail the
364 + // signed Content-Length at S3 with a far less legible error.
365 + if part.part_number != part_number as i32 || part.content_length != body.len() as u64 {
366 + return Err(SyncKitError::Internal(format!(
367 + "part geometry mismatch: server signed part {} for {} bytes, client has part {part_number} of {} bytes",
368 + part.part_number,
369 + part.content_length,
370 + body.len()
371 + )));
372 + }
373 +
374 + let resp = self
375 + .retry_request(
376 + Idempotency::IdempotentWrite { on: "(upload_id, part_number) — re-PUTting a part replaces it" },
377 + || {
378 + let req = self
379 + .http_stream
380 + .put(&part.url)
381 + .header("content-type", "application/octet-stream")
382 + .body(body.clone());
383 + async move { check_response(req.send().await?).await }
384 + },
385 + )
386 + .await?;
387 +
388 + // S3 identifies each part by the ETag it returns; complete cannot be
389 + // assembled without it.
390 + resp.headers()
391 + .get(reqwest::header::ETAG)
392 + .and_then(|v| v.to_str().ok())
393 + .map(|s| s.to_string())
394 + .ok_or_else(|| {
395 + SyncKitError::Internal(format!("part {part_number} upload returned no ETag"))
396 + })
397 + }
398 +
399 + /// Release an abandoned multipart session and the parts it holds.
400 + async fn blob_multipart_abort(&self, hash: &str, upload_id: &str) -> Result<()> {
401 + let token = self.require_token()?;
402 + let body = Bytes::from(serde_json::to_vec(&BlobMultipartAbortRequest {
403 + hash: hash.to_string(),
404 + upload_id: upload_id.to_string(),
405 + })?);
406 + self.retry_request(
407 + Idempotency::IdempotentWrite { on: "upload_id (aborting twice is a no-op)" },
408 + || {
409 + let req = self
410 + .http
411 + .post(&self.endpoints.blobs_multipart_abort)
412 + .bearer_auth(&token)
413 + .header("content-type", "application/json")
414 + .body(body.clone());
415 + async move { check_response(req.send().await?).await }
416 + },
417 + )
418 + .await?;
419 + Ok(())
420 + }
421 +
114 422 /// Confirm that a blob upload completed successfully.
115 423 /// The server verifies the object exists in S3 and records it.
116 424 #[instrument(skip(self))]
M shared/synckit-client/src/client/mod.rs +8
@@ -114,6 +114,10 @@ struct Endpoints {
114 114 blobs_upload: String,
115 115 blobs_confirm: String,
116 116 blobs_download: String,
117 + blobs_multipart_start: String,
118 + blobs_multipart_parts: String,
119 + blobs_multipart_complete: String,
120 + blobs_multipart_abort: String,
117 121 subscription: String,
118 122 subscription_checkout: String,
119 123 subscription_quote: String,
@@ -143,6 +147,10 @@ impl Endpoints {
143 147 blobs_upload: format!("{base}/api/v1/sync/blobs/upload"),
144 148 blobs_confirm: format!("{base}/api/v1/sync/blobs/confirm"),
145 149 blobs_download: format!("{base}/api/v1/sync/blobs/download"),
150 + blobs_multipart_start: format!("{base}/api/v1/sync/blobs/multipart/start"),
151 + blobs_multipart_parts: format!("{base}/api/v1/sync/blobs/multipart/parts"),
152 + blobs_multipart_complete: format!("{base}/api/v1/sync/blobs/multipart/complete"),
153 + blobs_multipart_abort: format!("{base}/api/v1/sync/blobs/multipart/abort"),
146 154 subscription: format!("{base}/api/v1/sync/subscription"),
147 155 subscription_checkout: format!("{base}/api/v1/sync/subscription/checkout"),
148 156 subscription_quote: format!("{base}/api/v1/sync/subscription/quote"),
M shared/synckit-client/src/types.rs +65
@@ -530,6 +530,71 @@ pub(crate) struct BlobConfirmRequest {
530 530 pub size_bytes: i64,
531 531 }
532 532
533 + // ── Blob multipart session ──
534 + //
535 + // The chunked transport for blobs above the server's one-shot PUT ceiling.
536 + // `size_bytes` throughout is the *ciphertext* length (`blob_encrypted_len` of
537 + // the plaintext), which both sides need to derive the same part geometry.
538 +
539 + #[derive(Serialize)]
540 + pub(crate) struct BlobMultipartStartRequest {
541 + pub hash: String,
542 + pub size_bytes: i64,
543 + }
544 +
545 + #[derive(Deserialize)]
546 + pub(crate) struct BlobMultipartStartResponse {
547 + pub upload_id: String,
548 + /// Ciphertext bytes per part; every part but the last is exactly this long.
549 + pub part_size: u64,
550 + pub part_count: u32,
551 + /// Server already holds this content address — no session was opened. Same
552 + /// consumer contract as [`BlobUploadUrlResponse::already_exists`].
553 + pub already_exists: bool,
554 + }
555 +
556 + #[derive(Serialize)]
557 + pub(crate) struct BlobMultipartPartsRequest {
558 + pub hash: String,
559 + pub upload_id: String,
560 + pub size_bytes: i64,
561 + pub first_part: u32,
562 + pub count: u32,
563 + }
564 +
565 + #[derive(Deserialize)]
566 + pub(crate) struct BlobMultipartPartsResponse {
567 + pub parts: Vec<BlobMultipartPartUrl>,
568 + }
569 +
570 + #[derive(Deserialize)]
571 + pub(crate) struct BlobMultipartPartUrl {
572 + pub part_number: i32,
573 + /// The `Content-Length` this URL was signed for. The PUT must send exactly
574 + /// this many bytes.
575 + pub content_length: u64,
576 + pub url: String,
577 + }
578 +
579 + #[derive(Serialize)]
580 + pub(crate) struct BlobMultipartCompleteRequest {
581 + pub hash: String,
582 + pub upload_id: String,
583 + pub parts: Vec<BlobMultipartCompletedPart>,
584 + }
585 +
586 + #[derive(Serialize)]
587 + pub(crate) struct BlobMultipartCompletedPart {
588 + pub part_number: i32,
589 + pub etag: String,
590 + }
591 +
592 + #[derive(Serialize)]
593 + pub(crate) struct BlobMultipartAbortRequest {
594 + pub hash: String,
595 + pub upload_id: String,
596 + }
597 +
533 598 #[derive(Serialize)]
534 599 pub(crate) struct BlobDownloadUrlRequest {
535 600 pub hash: String,
M shared/synckit-client/tests/integration.rs +309
@@ -3124,3 +3124,312 @@ mod rotation_e2e {
3124 3124 );
3125 3125 }
3126 3126 }
3127 +
3128 + // ── Multipart blob upload (streaming) ──
3129 + //
3130 + // The transport for blobs above the server's one-shot PUT ceiling. What matters
3131 + // here is that the client never buffers the whole ciphertext yet still produces
3132 + // a byte-exact v3 blob: it seals 1 MiB chunks as it reads the file and cuts the
3133 + // sealed stream at the part boundaries the server signed, which it could only
3134 + // pick because `blob_encrypted_len` predicts the ciphertext size up front.
3135 + mod blob_multipart {
3136 + use super::*;
3137 + use std::path::PathBuf;
3138 +
3139 + const START_PATH: &str = "/api/v1/sync/blobs/multipart/start";
3140 + const PARTS_PATH: &str = "/api/v1/sync/blobs/multipart/parts";
3141 + const COMPLETE_PATH: &str = "/api/v1/sync/blobs/multipart/complete";
3142 + const ABORT_PATH: &str = "/api/v1/sync/blobs/multipart/abort";
3143 + const PART_PUT_PATH: &str = "/s3/part";
3144 +
3145 + fn temp_blob(name: &str, contents: &[u8]) -> PathBuf {
3146 + use std::sync::atomic::{AtomicU64, Ordering};
3147 + static N: AtomicU64 = AtomicU64::new(0);
3148 + let mut p = std::env::temp_dir();
3149 + p.push(format!(
3150 + "synckit_mp_{}_{}_{name}",
3151 + std::process::id(),
3152 + N.fetch_add(1, Ordering::Relaxed)
3153 + ));
3154 + std::fs::write(&p, contents).unwrap();
3155 + p
3156 + }
3157 +
3158 + /// Mount the whole session: start (with the given plan), a parts window
3159 + /// whose URLs all point at one PUT path, and complete.
3160 + async fn mount_session(server: &MockServer, cipher_len: usize, part_size: usize) -> u32 {
3161 + let part_count = cipher_len.div_ceil(part_size) as u32;
3162 + let parts: Vec<serde_json::Value> = (1..=part_count)
3163 + .map(|n| {
3164 + let content_length = if n == part_count {
3165 + cipher_len - part_size * (part_count as usize - 1)
3166 + } else {
3167 + part_size
3168 + };
3169 + json!({
3170 + "part_number": n,
3171 + "content_length": content_length,
3172 + "url": format!("{}{PART_PUT_PATH}?partNumber={n}", server.uri()),
3173 + })
3174 + })
3175 + .collect();
3176 +
3177 + Mock::given(method("POST"))
3178 + .and(path(START_PATH))
3179 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3180 + "upload_id": "test-upload-id",
3181 + "part_size": part_size,
3182 + "part_count": part_count,
3183 + "already_exists": false,
3184 + })))
3185 + .mount(server)
3186 + .await;
3187 + Mock::given(method("POST"))
3188 + .and(path(PARTS_PATH))
3189 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "parts": parts })))
3190 + .mount(server)
3191 + .await;
3192 + Mock::given(method("PUT"))
3193 + .and(path(PART_PUT_PATH))
3194 + .respond_with(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
3195 + .mount(server)
3196 + .await;
3197 + Mock::given(method("POST"))
3198 + .and(path(COMPLETE_PATH))
3199 + .respond_with(ResponseTemplate::new(204))
3200 + .mount(server)
3201 + .await;
3202 +
3203 + part_count
3204 + }
3205 +
3206 + fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
3207 + reqs.iter().filter(|r| r.url.path() == p).count()
3208 + }
3209 +
3210 + #[tokio::test]
3211 + async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
3212 + let server = MockServer::start().await;
3213 + let client = authed_client(&server);
3214 + let key = synckit_client::crypto::generate_master_key();
3215 + client.set_master_key_raw(key);
3216 +
3217 + // Spans four 1 MiB chunks (three full plus a remainder), so sealed
3218 + // chunks straddle part boundaries rather than lining up with them.
3219 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
3220 + .map(|i| i as u8)
3221 + .collect();
3222 + let hash = format!("{:x}", sha2::Sha256::digest(&plaintext));
3223 + let file = temp_blob("big.bin", &plaintext);
3224 +
3225 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3226 + let part_size = 1024 * 1024;
3227 + let part_count = mount_session(&server, cipher_len, part_size).await;
3228 + assert!(part_count > 1, "the fixture must actually be multipart");
3229 +
3230 + client.blob_upload_streaming(&hash, &file).await.unwrap();
3231 +
3232 + let reqs = server.received_requests().await.unwrap();
3233 +
3234 + // The session was sized in ciphertext, predicted from the plaintext.
3235 + let start: serde_json::Value = reqs
3236 + .iter()
3237 + .find(|r| r.url.path() == START_PATH)
3238 + .unwrap()
3239 + .body_json()
3240 + .unwrap();
3241 + assert_eq!(start["size_bytes"].as_u64().unwrap(), cipher_len as u64);
3242 + assert_eq!(start["hash"].as_str().unwrap(), hash);
3243 +
3244 + // Every part carried exactly the length the server signed for it.
3245 + let puts: Vec<&wiremock::Request> =
3246 + reqs.iter().filter(|r| r.url.path() == PART_PUT_PATH).collect();
3247 + assert_eq!(puts.len() as u32, part_count, "one PUT per planned part");
3248 + for (i, put) in puts.iter().enumerate() {
3249 + let expected = if i as u32 == part_count - 1 {
3250 + cipher_len - part_size * (part_count as usize - 1)
3251 + } else {
3252 + part_size
3253 + };
3254 + assert_eq!(put.body.len(), expected, "part {} length", i + 1);
3255 + }
3256 +
3257 + // The concatenated parts are a valid v3 blob for this content address —
3258 + // proof that streaming produced the same wire format as the in-memory
3259 + // encrypt, boundaries and all.
3260 + let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
3261 + assert_eq!(assembled.len(), cipher_len);
3262 + let decrypted =
3263 + synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap();
3264 + assert_eq!(decrypted, plaintext, "streamed blob must round-trip");
3265 +
3266 + // Complete named every part, in order, with the ETag S3 returned.
3267 + let complete: serde_json::Value = reqs
3268 + .iter()
3269 + .find(|r| r.url.path() == COMPLETE_PATH)
3270 + .unwrap()
3271 + .body_json()
3272 + .unwrap();
3273 + let named = complete["parts"].as_array().unwrap();
3274 + assert_eq!(named.len() as u32, part_count);
3275 + for (i, part) in named.iter().enumerate() {
3276 + assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1);
3277 + assert_eq!(part["etag"].as_str().unwrap(), "\"part-etag\"");
3278 + }
3279 + assert_eq!(hits(&reqs, ABORT_PATH), 0, "a clean upload must not abort");
3280 +
3281 + std::fs::remove_file(&file).ok();
3282 + }
3283 +
3284 + #[tokio::test]
3285 + async fn streaming_upload_handles_an_empty_file() {
3286 + let server = MockServer::start().await;
3287 + let client = authed_client(&server);
3288 + let key = synckit_client::crypto::generate_master_key();
3289 + client.set_master_key_raw(key);
3290 +
3291 + let hash = format!("{:x}", sha2::Sha256::digest(b""));
3292 + let file = temp_blob("empty.bin", b"");
3293 + let cipher_len = synckit_client::crypto::blob_encrypted_len(0);
3294 + mount_session(&server, cipher_len, 1024 * 1024).await;
3295 +
3296 + client.blob_upload_streaming(&hash, &file).await.unwrap();
3297 +
3298 + let reqs = server.received_requests().await.unwrap();
3299 + let put = reqs.iter().find(|r| r.url.path() == PART_PUT_PATH).unwrap();
3300 + assert_eq!(put.body.len(), cipher_len, "one part carries the whole blob");
3301 + assert_eq!(
3302 + synckit_client::crypto::decrypt_blob_chunked(&put.body, &key, &hash).unwrap(),
3303 + Vec::<u8>::new(),
3304 + "an empty blob is still an authenticated single chunk"
3305 + );
3306 +
3307 + std::fs::remove_file(&file).ok();
3308 + }
3309 +
3310 + #[tokio::test]
3311 + async fn streaming_upload_skips_when_the_server_already_has_the_content() {
3312 + let server = MockServer::start().await;
3313 + let client = authed_client(&server);
3314 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3315 +
3316 + Mock::given(method("POST"))
3317 + .and(path(START_PATH))
3318 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3319 + "upload_id": "",
3320 + "part_size": 0,
3321 + "part_count": 0,
3322 + "already_exists": true,
3323 + })))
3324 + .mount(&server)
3325 + .await;
3326 +
3327 + let plaintext = b"content the server already holds";
3328 + let hash = format!("{:x}", sha2::Sha256::digest(plaintext));
3329 + let file = temp_blob("dedup.bin", plaintext);
3330 +
3331 + client.blob_upload_streaming(&hash, &file).await.unwrap();
3332 +
3333 + // Dedup must cost nothing: no file bytes read out to the wire, no
3334 + // session to clean up.
3335 + let reqs = server.received_requests().await.unwrap();
3336 + assert_eq!(hits(&reqs, PART_PUT_PATH), 0, "dedup must not upload parts");
3337 + assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
3338 + assert_eq!(hits(&reqs, ABORT_PATH), 0);
3339 +
3340 + std::fs::remove_file(&file).ok();
3341 + }
3342 +
3343 + #[tokio::test]
3344 + async fn streaming_upload_aborts_when_the_file_no_longer_matches_its_hash() {
3345 + // The caller hashed the file in an earlier pass. If it changed since,
3346 + // storing it under the stale content address would poison the address:
3347 + // every later download would re-hash and reject it. Fail here instead,
3348 + // and release the parts.
3349 + let server = MockServer::start().await;
3350 + let client = authed_client(&server);
3351 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3352 +
3353 + let plaintext = b"the bytes actually on disk";
3354 + let stale_hash = format!("{:x}", sha2::Sha256::digest(b"what the caller hashed earlier"));
3355 + let file = temp_blob("changed.bin", plaintext);
3356 +
3357 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3358 + mount_session(&server, cipher_len, 1024 * 1024).await;
3359 + Mock::given(method("POST"))
3360 + .and(path(ABORT_PATH))
3361 + .respond_with(ResponseTemplate::new(204))
3362 + .mount(&server)
3363 + .await;
3364 +
3365 + let err = client
3366 + .blob_upload_streaming(&stale_hash, &file)
3367 + .await
3368 + .expect_err("a hash mismatch must not be uploaded");
3369 + assert!(
3370 + matches!(err, SyncKitError::IntegrityFailed { .. }),
3371 + "expected IntegrityFailed, got {err:?}"
3372 + );
3373 +
3374 + let reqs = server.received_requests().await.unwrap();
3375 + assert_eq!(hits(&reqs, COMPLETE_PATH), 0, "a mismatched blob must not be assembled");
3376 + assert_eq!(hits(&reqs, ABORT_PATH), 1, "the session must be released");
3377 +
3378 + std::fs::remove_file(&file).ok();
3379 + }
3380 +
3381 + #[tokio::test]
3382 + async fn streaming_upload_aborts_when_a_part_upload_fails() {
3383 + // Parts already sent are billed until the session is aborted, so any
3384 + // failure past `start` has to release it.
3385 + let server = MockServer::start().await;
3386 + let client = authed_client(&server);
3387 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3388 +
3389 + let plaintext = b"a blob whose part upload will fail";
3390 + let hash = format!("{:x}", sha2::Sha256::digest(plaintext));
3391 + let file = temp_blob("failing.bin", plaintext);
3392 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3393 +
3394 + Mock::given(method("POST"))
3395 + .and(path(START_PATH))
3396 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3397 + "upload_id": "test-upload-id",
3398 + "part_size": cipher_len,
3399 + "part_count": 1,
3400 + "already_exists": false,
3401 + })))
3402 + .mount(&server)
3403 + .await;
3404 + Mock::given(method("POST"))
3405 + .and(path(PARTS_PATH))
3406 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3407 + "parts": [{
3408 + "part_number": 1,
3409 + "content_length": cipher_len,
3410 + "url": format!("{}{PART_PUT_PATH}", server.uri()),
3411 + }]
3412 + })))
3413 + .mount(&server)
3414 + .await;
3415 + Mock::given(method("PUT"))
3416 + .and(path(PART_PUT_PATH))
3417 + .respond_with(ResponseTemplate::new(403))
3418 + .mount(&server)
3419 + .await;
3420 + Mock::given(method("POST"))
3421 + .and(path(ABORT_PATH))
3422 + .respond_with(ResponseTemplate::new(204))
3423 + .mount(&server)
3424 + .await;
3425 +
3426 + let err = client.blob_upload_streaming(&hash, &file).await.unwrap_err();
3427 + assert!(matches!(err, SyncKitError::Server { status: 403, .. }), "got {err:?}");
3428 +
3429 + let reqs = server.received_requests().await.unwrap();
3430 + assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
3431 + assert_eq!(hits(&reqs, ABORT_PATH), 1, "a failed transfer must release its parts");
3432 +
3433 + std::fs::remove_file(&file).ok();
3434 + }
3435 + }