Skip to main content

max / synckit

44.2 KB · 1202 lines History Blame Raw
1 //! Multipart (streaming) blob upload.
2
3 // ── Multipart blob upload (streaming) ──
4 //
5 // The transport for blobs above the server's one-shot PUT ceiling. What matters
6 // here is that the client never buffers the whole ciphertext yet still produces
7 // a byte-exact v3 blob: it seals 1 MiB chunks as it reads the file and cuts the
8 // sealed stream at the part boundaries the server signed, which it could only
9 // pick because `blob_encrypted_len` predicts the ciphertext size up front.
10 use crate::common::*;
11 use std::path::PathBuf;
12
13 const START_PATH: &str = "/api/v1/sync/blobs/multipart/start";
14 const PARTS_PATH: &str = "/api/v1/sync/blobs/multipart/parts";
15 const COMPLETE_PATH: &str = "/api/v1/sync/blobs/multipart/complete";
16 const ABORT_PATH: &str = "/api/v1/sync/blobs/multipart/abort";
17 const PART_PUT_PATH: &str = "/s3/part";
18
19 fn temp_blob(name: &str, contents: &[u8]) -> PathBuf {
20 use std::sync::atomic::{AtomicU64, Ordering};
21 static N: AtomicU64 = AtomicU64::new(0);
22 let mut p = std::env::temp_dir();
23 p.push(format!(
24 "synckit_mp_{}_{}_{name}",
25 std::process::id(),
26 N.fetch_add(1, Ordering::Relaxed)
27 ));
28 std::fs::write(&p, contents).unwrap();
29 p
30 }
31
32 /// Stands in for the server's part-URL minting: answers whatever window the
33 /// client asked for, rather than a fixed list, since the client requests one
34 /// part at a time (it can only checksum a part it has already sealed).
35 struct PartsResponder {
36 cipher_len: usize,
37 part_size: usize,
38 base: String,
39 }
40
41 impl wiremock::Respond for PartsResponder {
42 fn respond(&self, req: &wiremock::Request) -> ResponseTemplate {
43 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
44 let first = body["first_part"].as_u64().unwrap() as usize;
45 let count = body["count"].as_u64().unwrap() as usize;
46 let part_count = self.cipher_len.div_ceil(self.part_size);
47 let last = (first + count - 1).min(part_count);
48
49 let parts: Vec<serde_json::Value> = (first..=last)
50 .map(|n| {
51 let content_length = if n == part_count {
52 self.cipher_len - self.part_size * (part_count - 1)
53 } else {
54 self.part_size
55 };
56 json!({
57 "part_number": n,
58 "content_length": content_length,
59 "url": format!("{}{PART_PUT_PATH}?partNumber={n}", self.base),
60 })
61 })
62 .collect();
63 ResponseTemplate::new(200).set_body_json(json!({ "parts": parts }))
64 }
65 }
66
67 /// Mount the whole session: start (with the given plan), part-URL minting,
68 /// the PUT target, and complete.
69 async fn mount_session(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
70 let part_count = mount_session_without_put(kit, cipher_len, part_size).await;
71 kit.put(PART_PUT_PATH)
72 .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
73 .await;
74 part_count
75 }
76
77 /// The session without the part PUT, for a test that mounts its own (one that
78 /// fails part way, say).
79 async fn mount_session_without_put(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
80 let part_count = cipher_len.div_ceil(part_size) as u32;
81
82 kit.post(START_PATH)
83 .json(json!({
84 "upload_id": "test-upload-id",
85 "part_size": part_size,
86 "part_count": part_count,
87 "already_exists": false,
88 }))
89 .await;
90 kit.post(PARTS_PATH)
91 .responder(PartsResponder {
92 cipher_len,
93 part_size,
94 base: kit.uri(),
95 })
96 .await;
97 kit.post(COMPLETE_PATH).code(204).empty().await;
98
99 part_count
100 }
101
102 #[tokio::test]
103 async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
104 let kit = MockKit::start().await;
105 let (client, key) = kit.keyed();
106
107 // Spans four 1 MiB chunks (three full plus a remainder), so sealed
108 // chunks straddle part boundaries rather than lining up with them.
109 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
110 .map(|i| i as u8)
111 .collect();
112 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
113 let file = temp_blob("big.bin", &plaintext);
114
115 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
116 let part_size = 1024 * 1024;
117 let part_count = mount_session(&kit, cipher_len, part_size).await;
118 assert!(part_count > 1, "the fixture must actually be multipart");
119
120 client.blob_upload_streaming(&hash, &file).await.unwrap();
121
122 // The session was sized in ciphertext, predicted from the plaintext.
123 let start = kit.body(START_PATH).await;
124 assert_eq!(start["size_bytes"].as_u64().unwrap(), cipher_len as u64);
125 assert_eq!(start["hash"].as_str().unwrap(), hash);
126
127 // Every part carried exactly the length the server signed for it.
128 let puts = kit.requests_to(PART_PUT_PATH).await;
129 assert_eq!(puts.len() as u32, part_count, "one PUT per planned part");
130 for (i, put) in puts.iter().enumerate() {
131 let expected = if i as u32 == part_count - 1 {
132 cipher_len - part_size * (part_count as usize - 1)
133 } else {
134 part_size
135 };
136 assert_eq!(put.body.len(), expected, "part {} length", i + 1);
137 }
138
139 // Each part was requested with the SHA-256 of exactly the bytes that
140 // part then carried, which is what S3 rehashes against at write time.
141 // The pairing is what matters: a checksum bound to the wrong part is
142 // worse than none, since it would reject a correct upload.
143 let part_reqs = kit.bodies("POST", PARTS_PATH).await;
144 assert_eq!(
145 part_reqs.len() as u32,
146 part_count,
147 "one URL request per part: a digest exists only once the part is sealed"
148 );
149 for (i, body) in part_reqs.iter().enumerate() {
150 assert_eq!(body["first_part"].as_u64().unwrap(), i as u64 + 1);
151 assert_eq!(body["count"].as_u64().unwrap(), 1);
152 let declared = body["checksums"][0].as_str().unwrap();
153 let expected =
154 base64::engine::general_purpose::STANDARD.encode(sha2::Sha256::digest(&puts[i].body));
155 assert_eq!(
156 declared,
157 expected,
158 "part {} checksum must match its bytes",
159 i + 1
160 );
161 // And the client must actually send it: it is a signed header, so
162 // dropping it would fail SigV4 at S3.
163 assert_eq!(
164 puts[i]
165 .headers
166 .get("x-amz-checksum-sha256")
167 .expect("the PUT must carry the checksum header")
168 .to_str()
169 .unwrap(),
170 declared
171 );
172 }
173
174 // The concatenated parts are a valid v3 blob for this content address:
175 // proof that streaming produced the same wire format as the in-memory
176 // encrypt, boundaries and all.
177 let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
178 assert_eq!(assembled.len(), cipher_len);
179 let decrypted = synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap();
180 assert_eq!(decrypted, plaintext, "streamed blob must round-trip");
181
182 // Complete named every part, in order, with the ETag S3 returned.
183 let complete = kit.body(COMPLETE_PATH).await;
184 let named = complete["parts"].as_array().unwrap();
185 assert_eq!(named.len() as u32, part_count);
186 for (i, part) in named.iter().enumerate() {
187 assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1);
188 assert_eq!(part["etag"].as_str().unwrap(), "\"part-etag\"");
189 }
190 assert_eq!(
191 kit.hits(ABORT_PATH).await,
192 0,
193 "a clean upload must not abort"
194 );
195
196 std::fs::remove_file(&file).ok();
197 }
198
199 #[tokio::test]
200 async fn streaming_upload_rejects_a_server_part_plan_that_lies_about_geometry() {
201 let kit = MockKit::start().await;
202 let (client, _key) = kit.keyed();
203
204 // A blob that genuinely spans several 1 MiB parts.
205 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
206 .map(|i| i as u8)
207 .collect();
208 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
209 let file = temp_blob("liar.bin", &plaintext);
210
211 // Hostile server: claims the whole multi-part blob fits in ONE part.
212 // Trusting it would defeat the one-part-in-memory bound, so the client
213 // must refuse before minting or PUTting anything.
214 kit.post(START_PATH)
215 .json(json!({
216 "upload_id": "test-upload-id",
217 "part_size": 1024 * 1024,
218 "part_count": 1,
219 "already_exists": false,
220 }))
221 .await;
222
223 let err = client
224 .blob_upload_streaming(&hash, &file)
225 .await
226 .unwrap_err();
227 assert!(
228 matches!(err, SyncKitError::Internal(ref m) if m.contains("does not match")),
229 "expected a geometry-mismatch rejection, got {err:?}"
230 );
231 assert_eq!(
232 kit.hits(PART_PUT_PATH).await,
233 0,
234 "no part may be uploaded once the plan is rejected"
235 );
236
237 std::fs::remove_file(&file).ok();
238 }
239
240 #[tokio::test]
241 async fn streaming_upload_handles_an_empty_file() {
242 let kit = MockKit::start().await;
243 let (client, key) = kit.keyed();
244
245 let hash = hex::encode(sha2::Sha256::digest(b""));
246 let file = temp_blob("empty.bin", b"");
247 let cipher_len = synckit_client::crypto::blob_encrypted_len(0);
248 mount_session(&kit, cipher_len, 1024 * 1024).await;
249
250 client.blob_upload_streaming(&hash, &file).await.unwrap();
251
252 let put = kit.raw_body(PART_PUT_PATH).await;
253 assert_eq!(put.len(), cipher_len, "one part carries the whole blob");
254 assert_eq!(
255 synckit_client::crypto::decrypt_blob_chunked(&put, &key, &hash).unwrap(),
256 Vec::<u8>::new(),
257 "an empty blob is still an authenticated single chunk"
258 );
259
260 std::fs::remove_file(&file).ok();
261 }
262
263 #[tokio::test]
264 async fn streaming_upload_skips_when_the_server_already_has_the_content() {
265 let kit = MockKit::start().await;
266 let (client, _key) = kit.keyed();
267
268 kit.post(START_PATH)
269 .json(json!({
270 "upload_id": "",
271 "part_size": 0,
272 "part_count": 0,
273 "already_exists": true,
274 }))
275 .await;
276
277 let plaintext = b"content the server already holds";
278 let hash = hex::encode(sha2::Sha256::digest(plaintext));
279 let file = temp_blob("dedup.bin", plaintext);
280
281 client.blob_upload_streaming(&hash, &file).await.unwrap();
282
283 // Dedup must cost nothing: no file bytes read out to the wire, no
284 // session to clean up.
285 assert_eq!(
286 kit.hits(PART_PUT_PATH).await,
287 0,
288 "dedup must not upload parts"
289 );
290 assert_eq!(kit.hits(COMPLETE_PATH).await, 0);
291 assert_eq!(kit.hits(ABORT_PATH).await, 0);
292
293 std::fs::remove_file(&file).ok();
294 }
295
296 #[tokio::test]
297 async fn streaming_upload_aborts_when_the_file_no_longer_matches_its_hash() {
298 // The caller hashed the file in an earlier pass. If it changed since,
299 // storing it under the stale content address would poison the address:
300 // every later download would re-hash and reject it. Fail here instead,
301 // and release the parts.
302 let kit = MockKit::start().await;
303 let (client, _key) = kit.keyed();
304
305 let plaintext = b"the bytes actually on disk";
306 let stale_hash = hex::encode(sha2::Sha256::digest(b"what the caller hashed earlier"));
307 let file = temp_blob("changed.bin", plaintext);
308
309 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
310 mount_session(&kit, cipher_len, 1024 * 1024).await;
311 kit.post(ABORT_PATH).code(204).empty().await;
312
313 let err = client
314 .blob_upload_streaming(&stale_hash, &file)
315 .await
316 .expect_err("a hash mismatch must not be uploaded");
317 assert!(
318 matches!(err, SyncKitError::IntegrityFailed { .. }),
319 "expected IntegrityFailed, got {err:?}"
320 );
321
322 assert_eq!(
323 kit.hits(COMPLETE_PATH).await,
324 0,
325 "a mismatched blob must not be assembled"
326 );
327 assert_eq!(
328 kit.hits(ABORT_PATH).await,
329 1,
330 "the session must be released"
331 );
332
333 std::fs::remove_file(&file).ok();
334 }
335
336 #[tokio::test]
337 async fn streaming_upload_aborts_when_a_part_upload_fails() {
338 // Parts already sent are billed until the session is aborted, so any
339 // failure past `start` has to release it.
340 let kit = MockKit::start().await;
341 let (client, _key) = kit.keyed();
342
343 let plaintext = b"a blob whose part upload will fail";
344 let hash = hex::encode(sha2::Sha256::digest(plaintext));
345 let file = temp_blob("failing.bin", plaintext);
346 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
347
348 kit.post(START_PATH)
349 .json(json!({
350 "upload_id": "test-upload-id",
351 "part_size": cipher_len,
352 "part_count": 1,
353 "already_exists": false,
354 }))
355 .await;
356 kit.post(PARTS_PATH)
357 .json(json!({
358 "parts": [{
359 "part_number": 1,
360 "content_length": cipher_len,
361 "url": kit.url(PART_PUT_PATH),
362 }]
363 }))
364 .await;
365 kit.put(PART_PUT_PATH).code(403).empty().await;
366 kit.post(ABORT_PATH).code(204).empty().await;
367
368 let err = client
369 .blob_upload_streaming(&hash, &file)
370 .await
371 .unwrap_err();
372 assert!(
373 matches!(err, SyncKitError::Server { status: 403, .. }),
374 "got {err:?}"
375 );
376
377 assert_eq!(kit.hits(COMPLETE_PATH).await, 0);
378 assert_eq!(
379 kit.hits(ABORT_PATH).await,
380 1,
381 "a failed transfer must release its parts"
382 );
383
384 std::fs::remove_file(&file).ok();
385 }
386
387 // ── Resuming an interrupted session ──
388 //
389 // A large blob is a long transfer, and a process killed part way used to throw
390 // all of it away: the parts were still at S3, but nothing on this side
391 // remembered the session. With a resume store installed the next attempt takes
392 // the session over and sends only what is missing.
393 //
394 // The hard part is not the bookkeeping, it is the crypto. Part boundaries come
395 // from the server and have nothing to do with the 1 MiB sealed-chunk geometry,
396 // so a resume almost always restarts inside a chunk whose leading bytes are
397 // already uploaded. Sealing draws a random nonce per chunk, so re-sealing that
398 // chunk with a new one would splice two keystreams together and the assembled
399 // object would never open. These tests are about that boundary.
400
401 use synckit_client::client::resume::BlobResumeStore;
402
403 /// PUTs that succeed for the first `ok` parts and then refuse, standing in for
404 /// a transfer that dies part way. 403 rather than 500 so the client treats it
405 /// as permanent and the test does not sit through the retry backoff.
406 struct DiesAfter {
407 ok: usize,
408 seen: std::sync::atomic::AtomicUsize,
409 }
410
411 impl wiremock::Respond for DiesAfter {
412 fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
413 let n = self.seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
414 if n < self.ok {
415 ResponseTemplate::new(200).append_header("ETag", format!("\"etag-{}\"", n + 1))
416 } else {
417 ResponseTemplate::new(403)
418 }
419 }
420 }
421
422 /// A resume store on its own scratch database, as the engine would install.
423 fn resume_store(name: &str) -> Arc<dyn BlobResumeStore> {
424 use std::sync::atomic::{AtomicU64, Ordering};
425 static N: AtomicU64 = AtomicU64::new(0);
426 let mut p = std::env::temp_dir();
427 p.push(format!(
428 "synckit_resume_{}_{}_{name}",
429 std::process::id(),
430 N.fetch_add(1, Ordering::Relaxed)
431 ));
432 std::fs::create_dir_all(&p).unwrap();
433 synckit_client::store::SqliteResumeStore::shared(synckit_client::store::DbSource::path(
434 p.join("app.db"),
435 ))
436 }
437
438 async fn put_bodies(kit: &MockKit) -> Vec<Vec<u8>> {
439 kit.requests_to(PART_PUT_PATH)
440 .await
441 .into_iter()
442 .map(|r| r.body)
443 .collect()
444 }
445
446 #[tokio::test]
447 async fn a_killed_upload_resumes_and_the_assembled_blob_still_opens() {
448 let kit = MockKit::start().await;
449 let key = synckit_client::crypto::generate_master_key();
450 let store = resume_store("kill");
451
452 // Four 1 MiB chunks against 700 KiB parts: no part boundary can land on a
453 // chunk boundary, so the resume is guaranteed to restart mid-chunk. That is
454 // the case the stored nonces exist for.
455 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
456 .map(|i| i as u8)
457 .collect();
458 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
459 let file = temp_blob("resume.bin", &plaintext);
460 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
461 let part_size = 700 * 1024;
462 let part_count = cipher_len.div_ceil(part_size);
463 assert!(part_count > 3, "the fixture must have parts to resume from");
464
465 // ── First attempt: dies after two parts ──
466 let client = kit.authed();
467 client.set_master_key_raw(key);
468 client.set_resume_store(Arc::clone(&store));
469
470 mount_session_without_put(&kit, cipher_len, part_size).await;
471 kit.put(PART_PUT_PATH)
472 .responder(DiesAfter {
473 ok: 2,
474 seen: std::sync::atomic::AtomicUsize::new(0),
475 })
476 .await;
477 kit.post(ABORT_PATH).code(204).empty().await;
478
479 let err = client
480 .blob_upload_streaming(&hash, &file)
481 .await
482 .unwrap_err();
483 assert!(
484 matches!(err, SyncKitError::Server { status: 403, .. }),
485 "got {err:?}"
486 );
487 // The session is the asset now: aborting it would throw away exactly what
488 // the next attempt is going to reuse.
489 assert_eq!(
490 kit.hits(ABORT_PATH).await,
491 0,
492 "a resumable failure must keep the session"
493 );
494 let first_two: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(2).collect();
495
496 let record = store.load(&hash).unwrap().expect("a session was recorded");
497 assert_eq!(record.usable_parts().len(), 2);
498 assert_eq!(record.session.upload_id, "test-upload-id");
499
500 // ── Second attempt: a fresh client, as a restarted process would have ──
501 kit.reset().await;
502 mount_session(&kit, cipher_len, part_size).await;
503 kit.post(ABORT_PATH).code(204).empty().await;
504
505 let restarted = kit.authed();
506 restarted.set_master_key_raw(key);
507 restarted.set_resume_store(Arc::clone(&store));
508 restarted.blob_upload_streaming(&hash, &file).await.unwrap();
509
510 let resumed = put_bodies(&kit).await;
511 assert_eq!(
512 resumed.len(),
513 part_count - 2,
514 "a resume must not re-send the parts already at S3"
515 );
516 // `start` is unconditional (it carries the dedup answer), so the redundant
517 // session it opens has to be released rather than left to the reaper.
518 assert_eq!(kit.hits(ABORT_PATH).await, 1);
519
520 // The whole point: the two runs' parts concatenate into one valid v3 blob,
521 // which means the chunk straddling the boundary came back byte-identical.
522 let assembled: Vec<u8> = first_two
523 .iter()
524 .chain(resumed.iter())
525 .flat_map(Clone::clone)
526 .collect();
527 assert_eq!(assembled.len(), cipher_len);
528 assert_eq!(
529 synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
530 plaintext,
531 "a resumed blob must decrypt to the original"
532 );
533
534 // Complete named every part, and the kept ones carry the first run's ETags.
535 let complete = kit.body(COMPLETE_PATH).await;
536 let named = complete["parts"].as_array().unwrap();
537 assert_eq!(named.len(), part_count);
538 assert_eq!(named[0]["etag"].as_str().unwrap(), "\"etag-1\"");
539 assert_eq!(named[1]["etag"].as_str().unwrap(), "\"etag-2\"");
540 assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");
541
542 // Assembled means the record describes nothing.
543 assert!(store.load(&hash).unwrap().is_none());
544
545 std::fs::remove_file(&file).ok();
546 }
547
548 #[tokio::test]
549 async fn a_resume_that_fails_again_gives_up_the_session_rather_than_wedging() {
550 // A session the server has already reaped would fail identically on every
551 // future pass. One resume attempt, then a clean slate.
552 let kit = MockKit::start().await;
553 let key = synckit_client::crypto::generate_master_key();
554 let store = resume_store("wedge");
555
556 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 3))
557 .map(|i| i as u8)
558 .collect();
559 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
560 let file = temp_blob("wedge.bin", &plaintext);
561 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
562 let part_size = 700 * 1024;
563
564 let client = kit.authed();
565 client.set_master_key_raw(key);
566 client.set_resume_store(Arc::clone(&store));
567
568 // Attempt one: two parts land, then the transfer dies. The record survives.
569 mount_session_without_put(&kit, cipher_len, part_size).await;
570 kit.put(PART_PUT_PATH)
571 .responder(DiesAfter {
572 ok: 2,
573 seen: std::sync::atomic::AtomicUsize::new(0),
574 })
575 .await;
576 kit.post(ABORT_PATH).code(204).empty().await;
577 client
578 .blob_upload_streaming(&hash, &file)
579 .await
580 .unwrap_err();
581 assert!(store.load(&hash).unwrap().is_some());
582
583 // Attempt two resumes into a session that refuses everything.
584 kit.reset().await;
585 mount_session_without_put(&kit, cipher_len, part_size).await;
586 kit.put(PART_PUT_PATH).code(403).empty().await;
587 kit.post(ABORT_PATH).code(204).empty().await;
588 client
589 .blob_upload_streaming(&hash, &file)
590 .await
591 .unwrap_err();
592
593 assert!(
594 store.load(&hash).unwrap().is_none(),
595 "a failed resume must drop the record so the next pass starts clean"
596 );
597 assert!(
598 kit.hits(ABORT_PATH).await >= 1,
599 "and release the parts it is giving up on"
600 );
601
602 std::fs::remove_file(&file).ok();
603 }
604
605 #[tokio::test]
606 async fn a_file_that_changed_under_the_session_is_refused_rather_than_re_sealed() {
607 // The nonce is the danger. Re-sealing different plaintext under a nonce this
608 // key has already used would leak the XOR of the two chunks, so the resume
609 // path checks a recorded plaintext digest before it re-uses one. A file
610 // edited between attempts must stop the upload, not quietly seal.
611 let kit = MockKit::start().await;
612 let key = synckit_client::crypto::generate_master_key();
613 let store = resume_store("changed");
614
615 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
616 .map(|i| i as u8)
617 .collect();
618 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
619 let file = temp_blob("changed.bin", &plaintext);
620 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
621 let part_size = 700 * 1024;
622
623 let client = kit.authed();
624 client.set_master_key_raw(key);
625 client.set_resume_store(Arc::clone(&store));
626 mount_session_without_put(&kit, cipher_len, part_size).await;
627 kit.put(PART_PUT_PATH)
628 .responder(DiesAfter {
629 ok: 2,
630 seen: std::sync::atomic::AtomicUsize::new(0),
631 })
632 .await;
633 kit.post(ABORT_PATH).code(204).empty().await;
634 client
635 .blob_upload_streaming(&hash, &file)
636 .await
637 .unwrap_err();
638 assert_eq!(store.load(&hash).unwrap().unwrap().usable_parts().len(), 2);
639
640 // Same length, different bytes, and the byte is inside the chunk the resume
641 // has to re-seal under the stored nonce (two 700 KiB parts land the boundary
642 // in chunk 1). Every length check still passes, so the digest is the only
643 // thing between this and a nonce re-use.
644 let mut edited = plaintext.clone();
645 edited[synckit_client::crypto::BLOB_CHUNK_SIZE + 5] ^= 0xff;
646 std::fs::write(&file, &edited).unwrap();
647
648 kit.reset().await;
649 mount_session(&kit, cipher_len, part_size).await;
650 kit.post(ABORT_PATH).code(204).empty().await;
651 let err = client
652 .blob_upload_streaming(&hash, &file)
653 .await
654 .unwrap_err();
655 assert!(
656 matches!(err, SyncKitError::Internal(ref m) if m.contains("changed under an in-flight upload")),
657 "got {err:?}"
658 );
659 assert_eq!(
660 kit.hits(COMPLETE_PATH).await,
661 0,
662 "nothing may be assembled from two different files"
663 );
664
665 std::fs::remove_file(&file).ok();
666 }
667
668 /// A previous attempt that got every part to S3 and died on the assemble call
669 /// must assemble on the next attempt without re-sending a byte.
670 ///
671 /// This is the one resume shape where the streaming loop sends nothing at all:
672 /// the recorded parts cover the whole ciphertext, so the boundary lands past
673 /// the last chunk and every chunk is read for the content-address check and
674 /// then skipped. What the client owes the server is the part list it already
675 /// has, and only `complete` is left to do.
676 ///
677 /// It is reachable because a failed `complete` is the one failure that keeps
678 /// the session and the record: `stream_blob_parts` returned `Ok`, so the abort
679 /// and clear that guard a failed transfer are never run.
680 #[tokio::test]
681 async fn a_resume_that_already_holds_every_part_assembles_without_sending_one() {
682 let kit = MockKit::start().await;
683 let key = synckit_client::crypto::generate_master_key();
684 let store = resume_store("complete-died");
685
686 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 11))
687 .map(|i| i as u8)
688 .collect();
689 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
690 let file = temp_blob("complete-died.bin", &plaintext);
691 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
692 let part_size = 700 * 1024;
693 let part_count = cipher_len.div_ceil(part_size);
694 assert!(
695 part_count > 1,
696 "the fixture must be a real multipart upload"
697 );
698
699 // ── First attempt: every part lands, the assemble call is refused ──
700 let client = kit.authed();
701 client.set_master_key_raw(key);
702 client.set_resume_store(Arc::clone(&store));
703
704 kit.post(START_PATH)
705 .json(json!({
706 "upload_id": "test-upload-id",
707 "part_size": part_size,
708 "part_count": part_count,
709 "already_exists": false,
710 }))
711 .await;
712 kit.post(PARTS_PATH)
713 .responder(PartsResponder {
714 cipher_len,
715 part_size,
716 base: kit.uri(),
717 })
718 .await;
719 kit.put(PART_PUT_PATH)
720 .responder(DiesAfter {
721 // Never dies: this run is about what happens after the parts are up.
722 ok: usize::MAX,
723 seen: std::sync::atomic::AtomicUsize::new(0),
724 })
725 .await;
726 // 403 rather than 500 so the client treats it as permanent and the test does
727 // not sit through the retry backoff.
728 kit.post(COMPLETE_PATH)
729 .code(403)
730 .json(json!({ "message": "assemble refused" }))
731 .await;
732 kit.post(ABORT_PATH).code(204).empty().await;
733
734 let err = client
735 .blob_upload_streaming(&hash, &file)
736 .await
737 .unwrap_err();
738 assert!(
739 matches!(err, SyncKitError::Server { status: 403, .. }),
740 "got {err:?}"
741 );
742 let sent = put_bodies(&kit).await;
743 assert_eq!(sent.len(), part_count, "the first attempt sent every part");
744
745 let record = store
746 .load(&hash)
747 .unwrap()
748 .expect("a failed complete keeps the session: it is what the retry needs");
749 assert_eq!(
750 record.usable_parts().len(),
751 part_count,
752 "every part must be recorded, or this is a different resume shape"
753 );
754
755 // ── Second attempt: nothing left to send ──
756 kit.reset().await;
757 mount_session(&kit, cipher_len, part_size).await;
758 kit.post(ABORT_PATH).code(204).empty().await;
759
760 let restarted = kit.authed();
761 restarted.set_master_key_raw(key);
762 restarted.set_resume_store(Arc::clone(&store));
763 restarted.blob_upload_streaming(&hash, &file).await.unwrap();
764
765 assert!(
766 put_bodies(&kit).await.is_empty(),
767 "a resume holding every part must not re-send one"
768 );
769 // The redundant session `start` opened is released rather than left to the reaper.
770 assert_eq!(kit.hits(ABORT_PATH).await, 1);
771
772 // What it did instead: named the parts the first run uploaded, with the
773 // ETags that run was given.
774 let complete = kit.body(COMPLETE_PATH).await;
775 let named = complete["parts"].as_array().unwrap();
776 assert_eq!(named.len(), part_count, "complete must name every part");
777 for (i, part) in named.iter().enumerate() {
778 assert_eq!(part["part_number"].as_u64(), Some(i as u64 + 1));
779 assert_eq!(
780 part["etag"].as_str(),
781 Some(format!("\"etag-{}\"", i + 1)).as_deref(),
782 "part {} lost the ETag the first run was given",
783 i + 1
784 );
785 }
786 assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");
787
788 // Guard against a vacuous pass: the bytes the first run sent really were the
789 // whole blob, so assembling them is the right thing to have done.
790 let assembled: Vec<u8> = sent.into_iter().flatten().collect();
791 assert_eq!(assembled.len(), cipher_len);
792 assert_eq!(
793 synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
794 plaintext
795 );
796
797 // Assembled means the record describes nothing.
798 assert!(store.load(&hash).unwrap().is_none());
799
800 std::fs::remove_file(&file).ok();
801 }
802
803 // ── Part-boundary arithmetic ──
804 //
805 // The part plan is arithmetic on the ciphertext length, and the cases that break
806 // it are the exact multiples and their two neighbours: a ciphertext of exactly N
807 // parts, one byte short of N parts, and one byte over. Every fixture above sits
808 // mid-part, so none of them separates `div_ceil` from a truncating divide, nor a
809 // final part sized `cipher_len - part_size * (n - 1)` from one sized `part_size`.
810
811 /// A plaintext length whose v3 ciphertext is exactly `n * part_size`, plus that
812 /// part size. `approx` grows by at most `n - 1` bytes to reach divisibility,
813 /// which cannot change the chunk count for a length that is not itself on a
814 /// chunk boundary.
815 fn exact_part_multiple(n: usize, approx: usize) -> (usize, usize) {
816 let cipher = synckit_client::crypto::blob_encrypted_len(approx);
817 let plaintext_len = approx + (n - cipher % n) % n;
818 let cipher = synckit_client::crypto::blob_encrypted_len(plaintext_len);
819 assert_eq!(cipher % n, 0, "the fixture must land on a part boundary");
820 (plaintext_len, cipher / n)
821 }
822
823 /// Upload a blob whose ciphertext is `n * part_size + delta` bytes and check the
824 /// whole plan: how many parts were requested, how long each PUT was, and that
825 /// the parts concatenate back into a blob that opens.
826 async fn boundary_upload(n: usize, approx: usize, delta: isize) {
827 let (exact_len, part_size) = exact_part_multiple(n, approx);
828 let plaintext_len = exact_len.checked_add_signed(delta).unwrap();
829
830 let plaintext: Vec<u8> = (0..plaintext_len).map(|i| i as u8).collect();
831 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
832 let file = temp_blob("boundary.bin", &plaintext);
833 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext_len);
834
835 // The three cases differ in exactly the way the arithmetic has to notice:
836 // a byte over spills into an extra part carrying a single byte.
837 let expected_parts = if delta > 0 { n + 1 } else { n };
838 assert_eq!(
839 cipher_len.div_ceil(part_size),
840 expected_parts,
841 "fixture geometry: n={n} delta={delta}"
842 );
843
844 let kit = MockKit::start().await;
845 let (client, key) = kit.keyed();
846 let planned = mount_session(&kit, cipher_len, part_size).await;
847 assert_eq!(planned as usize, expected_parts);
848
849 client.blob_upload_streaming(&hash, &file).await.unwrap();
850
851 let puts = kit.requests_to(PART_PUT_PATH).await;
852 assert_eq!(
853 puts.len(),
854 expected_parts,
855 "one PUT per planned part: n={n} delta={delta}"
856 );
857 for (i, put) in puts.iter().enumerate() {
858 let expected = if i + 1 == expected_parts {
859 cipher_len - part_size * (expected_parts - 1)
860 } else {
861 part_size
862 };
863 assert_eq!(
864 put.body.len(),
865 expected,
866 "n={n} delta={delta} part {} length",
867 i + 1
868 );
869 }
870
871 let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
872 assert_eq!(assembled.len(), cipher_len, "n={n} delta={delta}");
873 assert_eq!(
874 synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
875 plaintext,
876 "n={n} delta={delta}: the parts must reassemble into the original"
877 );
878
879 std::fs::remove_file(&file).ok();
880 }
881
882 #[tokio::test]
883 async fn two_part_boundaries_are_planned_and_sent_exactly() {
884 // One sealed chunk cut into two parts.
885 for delta in [-1, 0, 1] {
886 boundary_upload(2, 600_000, delta).await;
887 }
888 }
889
890 #[tokio::test]
891 async fn three_part_boundaries_are_planned_and_sent_exactly() {
892 // Three sealed chunks cut into three parts, so chunk and part boundaries
893 // are near each other without coinciding.
894 for delta in [-1, 0, 1] {
895 boundary_upload(
896 3,
897 synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 500_000,
898 delta,
899 )
900 .await;
901 }
902 }
903
904 // ── Hostile part plans ──
905 //
906 // `part_size` and `part_count` come from the server and drive both the
907 // allocation and the cut points of the sealed stream, so every bound on them is
908 // arithmetic the client cannot get wrong quietly. Each case below is written as
909 // a pair: the value that must be accepted by a gate and the adjacent one that
910 // must not, told apart by which message came back. A test that only checked
911 // "is_err" would pass with any gate firing, including the wrong one.
912
913 /// Mount a start response carrying an arbitrary plan, plus the abort the client
914 /// makes on its way out. No part-URL route is mounted: every case here must be
915 /// refused before a part is requested.
916 async fn mount_hostile_plan(kit: &MockKit, part_size: u64, part_count: u32) {
917 kit.post(START_PATH)
918 .json(json!({
919 "upload_id": "hostile-upload-id",
920 "part_size": part_size,
921 "part_count": part_count,
922 "already_exists": false,
923 }))
924 .await;
925 kit.post(ABORT_PATH).code(204).empty().await;
926 }
927
928 /// Run a streaming upload of a 5000-byte blob against `(part_size, part_count)`
929 /// and return the internal-error message it was refused with.
930 async fn plan_rejection(part_size: u64, part_count: u32) -> String {
931 let kit = MockKit::start().await;
932 let (client, _key) = kit.keyed();
933 let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
934 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
935 let file = temp_blob("hostile.bin", &plaintext);
936 mount_hostile_plan(&kit, part_size, part_count).await;
937
938 let err = client
939 .blob_upload_streaming(&hash, &file)
940 .await
941 .unwrap_err();
942 std::fs::remove_file(&file).ok();
943 assert_eq!(
944 kit.hits(PARTS_PATH).await,
945 0,
946 "a plan refused up front must not mint a single part URL"
947 );
948 match err {
949 SyncKitError::Internal(message) => message,
950 other => panic!("expected an Internal rejection, got {other:?}"),
951 }
952 }
953
954 #[tokio::test]
955 async fn a_plan_with_no_bytes_per_part_or_no_parts_is_refused_as_empty() {
956 // part_size 0 is also the divisor of the tiling check below the guard, so a
957 // guard that let it through would divide by zero rather than mis-upload.
958 assert!(
959 plan_rejection(0, 3).await.contains("empty multipart plan"),
960 "part_size 0 must be refused as an empty plan"
961 );
962 // part_count 0 is the other half of the same `||`: with an `&&` in its
963 // place, a plan that is empty in only one of the two ways gets through.
964 assert!(
965 plan_rejection(1024 * 1024, 0)
966 .await
967 .contains("empty multipart plan"),
968 "part_count 0 must be refused as an empty plan"
969 );
970 }
971
972 #[tokio::test]
973 async fn the_part_size_ceiling_admits_exactly_one_gibibyte_and_refuses_one_byte_more() {
974 // At the ceiling the plan is legal geometry and is judged on whether it
975 // tiles the blob (it does not: 5000 bytes is one part, not two). One byte
976 // over is refused by the ceiling itself. The two messages name which gate
977 // fired, which is the only thing that separates `>` from `>=` and `==`.
978 let at = plan_rejection(1 << 30, 2).await;
979 assert!(
980 at.contains("does not match"),
981 "a part_size of exactly 1 GiB is under the ceiling: {at}"
982 );
983 let over = plan_rejection((1 << 30) + 1, 2).await;
984 assert!(
985 over.contains("exceeds the 1073741824-byte ceiling"),
986 "one byte over the ceiling must be refused by it: {over}"
987 );
988 }
989
990 #[tokio::test]
991 async fn the_part_count_ceiling_admits_exactly_ten_thousand_and_refuses_one_more() {
992 // S3's own hard limit, so 10_000 parts is a legal plan and must reach the
993 // tiling check; 10_001 is not.
994 let at = plan_rejection(1024 * 1024, 10_000).await;
995 assert!(
996 at.contains("does not match"),
997 "a part_count of exactly 10000 is under the ceiling: {at}"
998 );
999 let over = plan_rejection(1024 * 1024, 10_001).await;
1000 assert!(
1001 over.contains("exceeds the 10000-part ceiling"),
1002 "one part over the ceiling must be refused by it: {over}"
1003 );
1004 }
1005
1006 /// Mints one part URL per request with a caller-chosen `part_number` and
1007 /// `content_length`, so a test can make the server's signed geometry disagree
1008 /// with the bytes the client holds.
1009 struct LyingPartsResponder {
1010 part_number: i64,
1011 content_length: u64,
1012 base: String,
1013 }
1014
1015 impl wiremock::Respond for LyingPartsResponder {
1016 fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
1017 ResponseTemplate::new(200).set_body_json(json!({
1018 "parts": [{
1019 "part_number": self.part_number,
1020 "content_length": self.content_length,
1021 "url": format!("{}{PART_PUT_PATH}?partNumber={}", self.base, self.part_number),
1022 }],
1023 }))
1024 }
1025 }
1026
1027 /// A single-part session whose minted URL carries the given geometry. Returns
1028 /// the error the upload was refused with, or `None` if it went through.
1029 async fn minted_part_rejection(part_number: i64, content_length_delta: i64) -> Option<String> {
1030 let kit = MockKit::start().await;
1031 let (client, _key) = kit.keyed();
1032 let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
1033 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
1034 let file = temp_blob("mismatched-part.bin", &plaintext);
1035 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
1036
1037 // One part holding the whole blob, so the plan itself is beyond reproach and
1038 // only the minted URL disagrees.
1039 kit.post(START_PATH)
1040 .json(json!({
1041 "upload_id": "mismatch-upload-id",
1042 "part_size": cipher_len,
1043 "part_count": 1,
1044 "already_exists": false,
1045 }))
1046 .await;
1047 kit.post(PARTS_PATH)
1048 .responder(LyingPartsResponder {
1049 part_number,
1050 content_length: (cipher_len as i64 + content_length_delta) as u64,
1051 base: kit.uri(),
1052 })
1053 .await;
1054 kit.put(PART_PUT_PATH)
1055 .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
1056 .await;
1057 kit.post(COMPLETE_PATH).code(204).empty().await;
1058 kit.post(ABORT_PATH).code(204).empty().await;
1059
1060 let outcome = client.blob_upload_streaming(&hash, &file).await;
1061 std::fs::remove_file(&file).ok();
1062 match outcome {
1063 Ok(_) => {
1064 assert_eq!(kit.hits(PART_PUT_PATH).await, 1, "an accepted plan is PUT");
1065 None
1066 }
1067 Err(SyncKitError::Internal(message)) => {
1068 assert_eq!(
1069 kit.hits(PART_PUT_PATH).await,
1070 0,
1071 "a part whose geometry is disputed must not be sent anyway"
1072 );
1073 Some(message)
1074 }
1075 Err(other) => panic!("expected an Internal rejection, got {other:?}"),
1076 }
1077 }
1078
1079 #[tokio::test]
1080 async fn a_minted_part_url_that_disagrees_with_the_bytes_in_hand_is_refused() {
1081 // The agreeing case first, so the two disagreements below are known to be
1082 // the only difference: part 1, exactly the bytes the client sealed.
1083 assert!(
1084 minted_part_rejection(1, 0).await.is_none(),
1085 "a URL signed for the part the client actually holds must be used"
1086 );
1087
1088 // Signed for a different part: PUTting anyway would store the bytes at the
1089 // wrong index and assemble a scrambled object.
1090 let wrong_number = minted_part_rejection(2, 0)
1091 .await
1092 .expect("a URL signed for part 2 must not be used for part 1");
1093 assert!(
1094 wrong_number.contains("part geometry mismatch"),
1095 "wrong part_number: {wrong_number}"
1096 );
1097
1098 // Signed for a different length: Content-Length is a signed header, so this
1099 // fails SigV4 at S3 with a far less legible error if it is sent.
1100 let wrong_length = minted_part_rejection(1, -1)
1101 .await
1102 .expect("a URL signed for one byte less must not be used");
1103 assert!(
1104 wrong_length.contains("part geometry mismatch"),
1105 "wrong content_length: {wrong_length}"
1106 );
1107 }
1108
1109 #[tokio::test]
1110 async fn a_resume_that_lands_exactly_on_a_chunk_boundary_seals_the_chunk_afresh() {
1111 // The other resume test picks part boundaries that can never coincide with
1112 // a chunk boundary, which is the common case but only one side of the
1113 // question. Here the first part is exactly the header plus chunk 0, so the
1114 // resume restarts with `within == 0`: nothing of the boundary chunk is at
1115 // S3, and it must therefore be sealed from scratch. Demanding a recorded
1116 // nonce here would fail the upload outright, since no nonce was ever
1117 // recorded for a chunk that was never sent.
1118 let kit = MockKit::start().await;
1119 let key = synckit_client::crypto::generate_master_key();
1120 let store = resume_store("aligned");
1121
1122 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 4_321))
1123 .map(|i| i as u8)
1124 .collect();
1125 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
1126 let file = temp_blob("aligned-resume.bin", &plaintext);
1127 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
1128 // Header plus one whole sealed chunk: the one part size that puts the
1129 // second part's first byte on chunk 1's first byte.
1130 let part_size = synckit_client::crypto::blob_header_bytes(plaintext.len()).len()
1131 + synckit_client::crypto::sealed_blob_chunk_len(plaintext.len(), 0);
1132 let part_count = cipher_len.div_ceil(part_size);
1133 assert_eq!(part_count, 3, "three parts, resuming at the second");
1134
1135 // ── First attempt: one part lands, then the transfer dies ──
1136 let client = kit.authed();
1137 client.set_master_key_raw(key);
1138 client.set_resume_store(Arc::clone(&store));
1139
1140 mount_session_without_put(&kit, cipher_len, part_size).await;
1141 kit.put(PART_PUT_PATH)
1142 .responder(DiesAfter {
1143 ok: 1,
1144 seen: std::sync::atomic::AtomicUsize::new(0),
1145 })
1146 .await;
1147 kit.post(ABORT_PATH).code(204).empty().await;
1148
1149 let err = client
1150 .blob_upload_streaming(&hash, &file)
1151 .await
1152 .unwrap_err();
1153 assert!(
1154 matches!(err, SyncKitError::Server { status: 403, .. }),
1155 "got {err:?}"
1156 );
1157 let first: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(1).collect();
1158 assert_eq!(
1159 first[0].len(),
1160 part_size,
1161 "the first part is the header and chunk 0 exactly"
1162 );
1163
1164 let record = store.load(&hash).unwrap().expect("a session was recorded");
1165 assert_eq!(record.usable_parts().len(), 1);
1166 assert!(
1167 record.chunk(1).is_none(),
1168 "chunk 1 was never sent, so no nonce for it can have been recorded"
1169 );
1170
1171 // ── Second attempt ──
1172 kit.reset().await;
1173 mount_session(&kit, cipher_len, part_size).await;
1174 kit.post(ABORT_PATH).code(204).empty().await;
1175
1176 let restarted = kit.authed();
1177 restarted.set_master_key_raw(key);
1178 restarted.set_resume_store(Arc::clone(&store));
1179 restarted.blob_upload_streaming(&hash, &file).await.unwrap();
1180
1181 let resumed = put_bodies(&kit).await;
1182 assert_eq!(
1183 resumed.len(),
1184 part_count - 1,
1185 "only the missing parts go up"
1186 );
1187
1188 let assembled: Vec<u8> = first
1189 .iter()
1190 .chain(resumed.iter())
1191 .flat_map(Clone::clone)
1192 .collect();
1193 assert_eq!(assembled.len(), cipher_len);
1194 assert_eq!(
1195 synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
1196 plaintext,
1197 "a resume aligned to a chunk boundary must still assemble"
1198 );
1199
1200 std::fs::remove_file(&file).ok();
1201 }
1202