Skip to main content

max / synckit

19.4 KB · 531 lines History Blame Raw
1 //! Tests for [`super`].
2
3 use super::*;
4 use crate::types::*;
5
6 mod resume {
7 use super::super::*;
8 use std::sync::Mutex;
9
10 /// A store that answers with whatever the test put in it.
11 #[derive(Default)]
12 struct Fake {
13 record: Mutex<Option<ResumeRecord>>,
14 cleared: Mutex<bool>,
15 }
16 impl BlobResumeStore for Fake {
17 fn load(&self, _hash: &str) -> Result<Option<ResumeRecord>> {
18 Ok(self.record.lock().unwrap().clone())
19 }
20 fn begin(&self, _hash: &str, _session: &ResumeSession) -> Result<()> {
21 Ok(())
22 }
23 fn record_part(
24 &self,
25 _hash: &str,
26 _part: &ResumePart,
27 _chunks: &[ResumeChunk],
28 ) -> Result<()> {
29 Ok(())
30 }
31 fn clear(&self, _hash: &str) -> Result<()> {
32 *self.cleared.lock().unwrap() = true;
33 Ok(())
34 }
35 }
36
37 /// A plausible session: 3 parts of 8 bytes over a 24-byte ciphertext,
38 /// with the first part done.
39 fn fake(age_secs: i64) -> Fake {
40 Fake {
41 record: Mutex::new(Some(ResumeRecord {
42 session: ResumeSession {
43 upload_id: "u".into(),
44 part_size: 8,
45 part_count: 3,
46 size_bytes: 24,
47 },
48 age_secs,
49 parts: vec![ResumePart {
50 part_number: 1,
51 etag: "e".into(),
52 }],
53 chunks: vec![],
54 })),
55 cleared: Mutex::new(false),
56 }
57 }
58
59 #[test]
60 fn a_fresh_matching_record_is_taken() {
61 let store = fake(60);
62 assert!(SyncKitClient::load_resume(&store, "h", 24).is_some());
63 assert!(!*store.cleared.lock().unwrap());
64 }
65
66 #[test]
67 fn a_session_past_the_reaper_window_is_dropped() {
68 // The server aborts abandoned sessions at 24h, so an older record
69 // names an upload_id that no longer exists. Resuming into it would
70 // cost a doomed transfer before failing.
71 let store = fake(RESUME_MAX_AGE_SECS + 1);
72 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
73 assert!(
74 *store.cleared.lock().unwrap(),
75 "a dead record must be forgotten, not re-read next pass"
76 );
77 }
78
79 #[test]
80 fn a_store_failure_is_reported_and_swallowed() {
81 // `best_effort` is the whole of the rule that nothing about the
82 // resume store may fail an upload: it takes the error, says so, and
83 // returns. Both halves matter and neither is a return value, so a
84 // body replaced by `()` would behave identically to any caller. The
85 // log is where the difference lives.
86 let noisy = crate::test_support::events_from(|| {
87 best_effort(
88 "record_part",
89 Err(SyncKitError::Internal("disk full".into())),
90 );
91 });
92 let line = noisy
93 .iter()
94 .find(|e| {
95 e.message
96 .as_deref()
97 .is_some_and(|m| m.contains("record_part") && m.contains("disk full"))
98 })
99 .expect("a store failure must name the operation and the cause");
100 assert!(
101 line.message
102 .as_deref()
103 .is_some_and(|m| m.contains("will not resume")),
104 "the line must say what the failure costs, which is a restart from zero"
105 );
106
107 let quiet = crate::test_support::events_from(|| {
108 best_effort("record_part", Ok(()));
109 });
110 assert!(
111 quiet.is_empty(),
112 "a store that worked has nothing to report"
113 );
114 }
115
116 /// Half the server's 24h orphan-reaper window, in seconds, written out
117 /// rather than read from [`RESUME_MAX_AGE_SECS`]. The point of the two
118 /// tests below is to pin that constant's value as well as the
119 /// comparison against it, and taking the number from the code under
120 /// test would make them agree with whatever it happened to hold.
121 const TWELVE_HOURS: i64 = 43_200;
122
123 #[test]
124 fn a_record_on_the_twelve_hour_boundary_is_still_usable() {
125 // The comparison is `>`, not `>=`: the window is chosen to leave a
126 // slow transfer room to finish inside it, and a record that has just
127 // reached the boundary still names a session the server holds.
128 let store = fake(TWELVE_HOURS);
129 assert!(
130 SyncKitClient::load_resume(&store, "h", 24).is_some(),
131 "a record exactly at the limit is inside the window, not past it"
132 );
133 assert!(!*store.cleared.lock().unwrap());
134 }
135
136 #[test]
137 fn a_record_one_second_past_twelve_hours_is_dropped() {
138 let store = fake(TWELVE_HOURS + 1);
139 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
140 assert!(*store.cleared.lock().unwrap());
141 }
142
143 /// The line `load_resume` writes when it throws a record away.
144 const DISCARD_LINE: &str = "discarding an unusable blob resume record";
145
146 #[test]
147 fn only_a_faulty_record_is_reported_as_discarded() {
148 // The three ways a record is dropped are not one event. A stale
149 // session and a plan that does not tile the blob are faults, and an
150 // operator wondering why an upload restarted wants to see them. A
151 // record with no completed parts is the ordinary case of a run that
152 // died before its first part landed; logging that would put a line
153 // in front of somebody on every such retry, and it says nothing.
154 //
155 // The guard that draws that distinction returns nothing and changes
156 // nothing, so the log is the only place it is observable at all.
157
158 let empty = fake(60);
159 empty.record.lock().unwrap().as_mut().unwrap().parts.clear();
160 let quiet = crate::test_support::events_from(|| {
161 assert!(SyncKitClient::load_resume(&empty, "h", 24).is_none());
162 });
163 assert!(
164 quiet
165 .iter()
166 .all(|e| e.message.as_deref() != Some(DISCARD_LINE)),
167 "a record that simply has nothing to save is not a fault to report"
168 );
169
170 let stale = fake(TWELVE_HOURS + 1);
171 let logged = crate::test_support::events_from(|| {
172 assert!(SyncKitClient::load_resume(&stale, "h", 24).is_none());
173 });
174 let line = logged
175 .iter()
176 .find(|e| e.message.as_deref() == Some(DISCARD_LINE))
177 .expect("a stale session is a fault and must be reported");
178 assert_eq!(line.field("stale"), Some("true"));
179 assert_eq!(line.field("fits"), Some("true"), "it fits, it is just dead");
180
181 let misfit = fake(60);
182 let logged = crate::test_support::events_from(|| {
183 assert!(SyncKitClient::load_resume(&misfit, "h", 999).is_none());
184 });
185 let line = logged
186 .iter()
187 .find(|e| e.message.as_deref() == Some(DISCARD_LINE))
188 .expect("a record that cannot describe this upload must be reported");
189 assert_eq!(line.field("stale"), Some("false"));
190 assert_eq!(line.field("fits"), Some("false"));
191 }
192
193 #[test]
194 fn a_record_for_a_different_length_is_dropped() {
195 // Same content hash, different ciphertext length is a contradiction:
196 // whatever it describes, it is not this upload.
197 let store = fake(60);
198 assert!(SyncKitClient::load_resume(&store, "h", 999).is_none());
199 assert!(*store.cleared.lock().unwrap());
200 }
201
202 #[test]
203 fn a_plan_that_does_not_tile_the_blob_is_dropped() {
204 let store = fake(60);
205 store
206 .record
207 .lock()
208 .unwrap()
209 .as_mut()
210 .unwrap()
211 .session
212 .part_count = 7;
213 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
214 }
215
216 #[test]
217 fn a_plan_with_a_zero_part_size_is_dropped() {
218 // part_size 0 is the hostile case the `> 0` guard exists for: it is
219 // also the divisor of the tiling check below it, so a guard that let
220 // it through would divide by zero rather than merely mis-resume.
221 let store = fake(60);
222 store
223 .record
224 .lock()
225 .unwrap()
226 .as_mut()
227 .unwrap()
228 .session
229 .part_size = 0;
230 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
231 assert!(*store.cleared.lock().unwrap());
232 }
233
234 #[test]
235 fn a_plan_with_no_parts_is_dropped_even_where_the_tiling_check_would_agree() {
236 // part_count 0 over a 0-byte session: 0.div_ceil(8) == 0, so the
237 // tiling check is satisfied and the `part_count > 0` guard is the
238 // only thing rejecting it. A record naming a completed part in a
239 // zero-part plan describes nothing.
240 let store = fake(60);
241 {
242 let mut held = store.record.lock().unwrap();
243 let session = &mut held.as_mut().unwrap().session;
244 session.part_count = 0;
245 session.size_bytes = 0;
246 }
247 assert!(SyncKitClient::load_resume(&store, "h", 0).is_none());
248 }
249
250 #[test]
251 fn a_record_with_no_completed_parts_saves_nothing() {
252 // Not an error: the upload starts at part 1 either way. Dropping it
253 // means the session recorded is the one actually being used.
254 let store = fake(60);
255 store.record.lock().unwrap().as_mut().unwrap().parts.clear();
256 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
257 }
258
259 #[test]
260 fn a_store_that_errors_costs_a_restart_and_nothing_else() {
261 struct Broken;
262 impl BlobResumeStore for Broken {
263 fn load(&self, _: &str) -> Result<Option<ResumeRecord>> {
264 Err(SyncKitError::Internal("disk gone".into()))
265 }
266 fn begin(&self, _: &str, _: &ResumeSession) -> Result<()> {
267 Ok(())
268 }
269 fn record_part(&self, _: &str, _: &ResumePart, _: &[ResumeChunk]) -> Result<()> {
270 Ok(())
271 }
272 fn clear(&self, _: &str) -> Result<()> {
273 Ok(())
274 }
275 }
276 assert!(SyncKitClient::load_resume(&Broken, "h", 24).is_none());
277 }
278
279 #[test]
280 fn only_a_recurring_failure_gives_up_the_session() {
281 assert!(is_resumable_failure(&SyncKitError::Server {
282 status: 503,
283 message: String::new(),
284 retry_after_secs: None,
285 }));
286 assert!(!is_resumable_failure(&SyncKitError::IntegrityFailed {
287 expected: "a".into(),
288 actual: "b".into(),
289 }));
290 assert!(!is_resumable_failure(&SyncKitError::Internal(
291 "geometry".into()
292 )));
293 }
294
295 /// The header plus every sealed chunk, which is what the boundary
296 /// arithmetic walks.
297 fn header_len() -> usize {
298 crypto::blob_header_bytes(0).len()
299 }
300
301 #[test]
302 fn a_fresh_upload_starts_at_the_top() {
303 assert_eq!(resume_boundary(4096, header_len(), 0), (0, 0));
304 }
305
306 #[test]
307 fn a_boundary_inside_the_first_chunk_reports_its_offset() {
308 let h = header_len();
309 // 1000 bytes into chunk 0's sealed bytes.
310 assert_eq!(resume_boundary(4 << 20, h, h + 1000), (0, 1000));
311 }
312
313 #[test]
314 fn a_boundary_past_a_whole_chunk_lands_in_the_next() {
315 let h = header_len();
316 let c0 = crypto::sealed_blob_chunk_len(4 << 20, 0);
317 assert_eq!(resume_boundary(4 << 20, h, h + c0), (1, 0));
318 assert_eq!(resume_boundary(4 << 20, h, h + c0 + 5), (1, 5));
319 }
320
321 #[test]
322 fn a_boundary_past_the_last_chunk_means_nothing_is_left_to_send() {
323 let len = 4 << 20;
324 let cipher = crypto::blob_encrypted_len(len);
325 assert_eq!(
326 resume_boundary(len, header_len(), cipher),
327 (crypto::blob_chunk_count_for(len), 0)
328 );
329 }
330
331 #[test]
332 fn every_boundary_of_a_real_blob_maps_back_to_the_bytes_it_names() {
333 // The invariant the resume depends on: skipping to a part boundary
334 // and re-emitting from `within` into the boundary chunk reproduces
335 // the ciphertext tail exactly. Checked against a real sealed blob.
336 let key = [7u8; 32];
337 let plaintext: Vec<u8> = (0..(crypto::BLOB_CHUNK_SIZE * 2 + 511))
338 .map(|i| i as u8)
339 .collect();
340 let hash = "a".repeat(64);
341 let whole = crypto::encrypt_blob_chunked(&plaintext, &key, &hash).unwrap();
342 let h = crypto::blob_header_bytes(plaintext.len()).len();
343 let count = crypto::blob_chunk_count_for(plaintext.len());
344
345 for skip in [h + 1, h + 700 * 1024, h + 1_400_000, whole.len() - 3] {
346 let (index, within) = resume_boundary(plaintext.len(), h, skip);
347 assert!(index < count, "skip {skip} fell off the end");
348 // Where that chunk starts in the ciphertext.
349 let start: usize = h
350 + (0..index)
351 .map(|i| crypto::sealed_blob_chunk_len(plaintext.len(), i))
352 .sum::<usize>();
353 assert_eq!(start + within, skip, "boundary {skip} must be exact");
354 // And re-sealing it under its own nonce reproduces those bytes.
355 let sealed =
356 &whole[start..start + crypto::sealed_blob_chunk_len(plaintext.len(), index)];
357 let from = index as usize * crypto::BLOB_CHUNK_SIZE;
358 let to = (from + crypto::BLOB_CHUNK_SIZE).min(plaintext.len());
359 let again = crypto::reseal_blob_chunk(
360 &plaintext[from..to],
361 &key,
362 &hash,
363 index,
364 count,
365 &crypto::blob_chunk_nonce(sealed).unwrap(),
366 )
367 .unwrap();
368 assert_eq!(again, sealed, "chunk {index} must re-seal byte-identically");
369 }
370 }
371 }
372
373 #[test]
374 fn blob_upload_url_response_deserialization() {
375 let json = r#"{"upload_url": "https://s3.example.com/upload", "already_exists": false}"#;
376 let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
377 assert_eq!(resp.upload_url, "https://s3.example.com/upload");
378 assert!(!resp.already_exists);
379
380 let json = r#"{"upload_url": "", "already_exists": true}"#;
381 let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
382 assert!(resp.already_exists);
383 }
384
385 #[test]
386 fn blob_upload_url_request_serialization() {
387 let req = BlobUploadUrlRequest {
388 hash: "sha256-abc123".to_string(),
389 size_bytes: 1024,
390 };
391
392 let json = serde_json::to_string(&req).unwrap();
393 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
394 assert_eq!(parsed["hash"], "sha256-abc123");
395 assert_eq!(parsed["size_bytes"], 1024);
396 }
397
398 #[test]
399 fn blob_content_hash_format_matches_consumer() {
400 use sha2::{Digest, Sha256};
401 // The integrity check in blob_download compares against this exact form:
402 // lowercase hex of SHA-256, the same string consumers store as the blob
403 // hash. If this drifts, every verified download would falsely reject.
404 let h = hex::encode(Sha256::digest(b"hello blob"));
405 assert_eq!(h.len(), 64);
406 assert!(
407 h.chars()
408 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
409 );
410 }
411
412 #[test]
413 fn blob_confirm_request_serialization() {
414 let req = BlobConfirmRequest {
415 hash: "sha256-def456".to_string(),
416 size_bytes: 2048,
417 };
418
419 let json = serde_json::to_string(&req).unwrap();
420 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
421 assert_eq!(parsed["hash"], "sha256-def456");
422 assert_eq!(parsed["size_bytes"], 2048);
423 }
424
425 // ── The in-memory and streaming size caps ──
426
427 #[test]
428 fn the_blob_cap_is_four_gibibytes_exactly() {
429 // Pinned as a literal rather than as the same arithmetic the constant
430 // uses, because that arithmetic is what can be wrong. The two readings a
431 // single wrong operator produces here are 1_077_936_128 and 4_195_328:
432 // both look like plausible caps, and either would refuse legitimate
433 // media the SDK documents itself as carrying. Nothing else in the suite
434 // can see the difference, since no fixture is anywhere near any of the
435 // three values.
436 assert_eq!(MAX_BLOB_BYTES, 4_294_967_296, "4 GiB");
437 }
438
439 /// A client holding a key but no session: every blob path gets past the
440 /// key check and stops at `require_token`, which is what makes
441 /// `NotAuthenticated` mean "the size check let this through".
442 fn keyed_but_unauthenticated() -> SyncKitClient {
443 let client = SyncKitClient::new(crate::SyncKitConfig {
444 server_url: "https://example.invalid".to_string(),
445 api_key: "test-api-key".to_string(),
446 });
447 client.set_master_key_raw([9u8; 32]);
448 client
449 }
450
451 /// A sparse file of `len` bytes: `set_len` allocates nothing, so the
452 /// multi-gigabyte sizes the cap is written in terms of cost no disk. The
453 /// cap is read off `metadata`, which is all these tests reach.
454 fn sparse_file(len: u64) -> std::path::PathBuf {
455 use std::sync::atomic::{AtomicU64, Ordering};
456 static N: AtomicU64 = AtomicU64::new(0);
457 let mut p = std::env::temp_dir();
458 p.push(format!(
459 "synckit_cap_{}_{}",
460 std::process::id(),
461 N.fetch_add(1, Ordering::Relaxed)
462 ));
463 let f = std::fs::File::create(&p).unwrap();
464 f.set_len(len).unwrap();
465 p
466 }
467
468 #[tokio::test]
469 async fn the_streaming_cap_accepts_a_blob_of_exactly_the_cap_and_refuses_one_byte_more() {
470 // Both sides of the bound. `>` differs from `>=` and from `==` only at
471 // the cap itself, so a test that only uploads something small cannot
472 // see any of them: at every reachable size all three agree.
473 let client = keyed_but_unauthenticated();
474 let hash = "b".repeat(64);
475
476 let at_cap = sparse_file(MAX_BLOB_BYTES as u64);
477 let err = client
478 .blob_upload_streaming(&hash, &at_cap)
479 .await
480 .unwrap_err();
481 assert!(
482 matches!(err, SyncKitError::NotAuthenticated),
483 "a blob of exactly {MAX_BLOB_BYTES} bytes is under the cap and must reach the session check, got {err:?}"
484 );
485
486 let over = sparse_file(MAX_BLOB_BYTES as u64 + 1);
487 let err = client
488 .blob_upload_streaming(&hash, &over)
489 .await
490 .unwrap_err();
491 match err {
492 SyncKitError::InvalidArgument(m) => {
493 assert!(m.contains("client cap"), "wrong rejection: {m}");
494 }
495 other => panic!("one byte over the cap must be refused, got {other:?}"),
496 }
497
498 // And a small file is not refused by a cap that has been inverted.
499 let small = sparse_file(1_000);
500 let err = client
501 .blob_upload_streaming(&hash, &small)
502 .await
503 .unwrap_err();
504 assert!(
505 matches!(err, SyncKitError::NotAuthenticated),
506 "a 1000-byte blob must reach the session check, got {err:?}"
507 );
508
509 for p in [at_cap, over, small] {
510 let _ = std::fs::remove_file(p);
511 }
512 }
513
514 #[tokio::test]
515 async fn the_in_memory_cap_passes_an_ordinary_blob_through_to_the_put() {
516 // The reachable half of the same bound: a blob far under the cap must
517 // not be rejected by it, so the call fails at the transport instead.
518 // (The unreachable half is the cap itself, which would need a 4 GiB
519 // allocation to reach.) A relative URL is a reqwest builder error,
520 // classified as permanent, so no network attempt is made.
521 let client = keyed_but_unauthenticated();
522 let err = client
523 .blob_upload(&"c".repeat(64), "not-a-url", vec![7u8; 5_000])
524 .await
525 .unwrap_err();
526 assert!(
527 matches!(err, SyncKitError::Http(_)),
528 "a 5000-byte blob is under the cap and must reach the PUT, got {err:?}"
529 );
530 }
531