Skip to main content

max / synckit

45.3 KB · 1415 lines History Blame Raw
1 //! Tests for [`super`].
2
3 use super::*;
4 use crate::ids::DeviceId;
5 use crate::types::ChangeOp;
6 use base64::Engine;
7 use chrono::Utc;
8 use std::time::Duration;
9 use uuid::Uuid;
10
11 use super::super::TOKEN_EXPIRY_BUFFER_SECS;
12
13 fn test_config() -> super::super::SyncKitConfig {
14 super::super::SyncKitConfig {
15 server_url: "https://example.com".to_string(),
16 api_key: "test-api-key-123".to_string(),
17 }
18 }
19
20 /// Build a fake JWT with the given `exp` claim (no real signature).
21 fn fake_jwt(exp: i64) -> String {
22 let header =
23 base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256","typ":"JWT"}"#);
24 let payload_json = serde_json::json!({
25 "sub": "550e8400-e29b-41d4-a716-446655440000",
26 "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
27 "exp": exp,
28 "iat": exp - 3600,
29 });
30 let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD
31 .encode(payload_json.to_string().as_bytes());
32 let signature = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
33 format!("{header}.{payload}.{signature}")
34 }
35
36 // ── wire-version envelope dispatch ──
37
38 #[test]
39 fn split_envelope_dispatches_on_explicit_version() {
40 let node = DeviceId::new(Uuid::from_u128(1));
41 let hlc = Hlc {
42 wall_ms: 5,
43 counter: 2,
44 node,
45 };
46
47 // v2 envelope: explicit __skver, parsed by version.
48 let v2 = serde_json::json!({ "__skver": 2, "__skhlc": hlc, "data": {"k": "v"} });
49 let (got, data, _) = SyncKitClient::split_hlc_envelope(v2, node, 0).unwrap();
50 assert_eq!(got, hlc);
51 assert_eq!(data, Some(serde_json::json!({"k": "v"})));
52
53 // gen-1 envelope: __skhlc present, no version tag.
54 let gen1 = serde_json::json!({ "__skhlc": hlc, "data": null });
55 let (got, data, _) = SyncKitClient::split_hlc_envelope(gen1, node, 0).unwrap();
56 assert_eq!(got, hlc);
57 assert_eq!(data, None);
58
59 // Bare legacy row: HLC synthesized from node + timestamp.
60 let bare = serde_json::json!({ "title": "buy milk" });
61 let (got, data, _) = SyncKitClient::split_hlc_envelope(bare.clone(), node, 1234).unwrap();
62 assert_eq!(got, Hlc::from_legacy(1234, node));
63 assert_eq!(data, Some(bare));
64 }
65
66 #[test]
67 fn the_storage_stamp_rides_inside_the_sealed_envelope_and_survives_a_round_trip() {
68 let node = DeviceId::new(Uuid::from_u128(1));
69 let hlc = Hlc {
70 wall_ms: 5,
71 counter: 2,
72 node,
73 };
74
75 let stamped = SyncKitClient::hlc_envelope(&hlc, Some(&serde_json::json!({"k": "v"})), Some(4));
76 assert_eq!(stamped["__sksv"], serde_json::json!(4));
77 let (got, data, version) = SyncKitClient::split_hlc_envelope(stamped, node, 0).unwrap();
78 assert_eq!(got, hlc);
79 assert_eq!(data, Some(serde_json::json!({"k": "v"})));
80 assert_eq!(version, Some(4));
81
82 // A Delete carries no row payload and still carries the stamp, which is
83 // what lets the gate see a peer whose only pending change is a delete.
84 let delete = SyncKitClient::hlc_envelope(&hlc, None, Some(4));
85 let (_, data, version) = SyncKitClient::split_hlc_envelope(delete, node, 0).unwrap();
86 assert_eq!(data, None);
87 assert_eq!(version, Some(4));
88 }
89
90 #[test]
91 fn an_undeclared_version_leaves_the_sealed_bytes_exactly_as_they_were() {
92 let node = DeviceId::new(Uuid::from_u128(1));
93 let hlc = Hlc {
94 wall_ms: 5,
95 counter: 2,
96 node,
97 };
98 let plain = SyncKitClient::hlc_envelope(&hlc, None, None);
99 assert!(plain.get("__sksv").is_none());
100 assert_eq!(
101 plain,
102 serde_json::json!({ "__skver": 2, "__skhlc": hlc, "data": null }),
103 "an app that declares no version must push byte-identical envelopes"
104 );
105 }
106
107 /// The stamp is added *within* v2 rather than as a new `__skver`, so a reader
108 /// that predates it addresses `__skhlc` and `data` by name and is unaffected.
109 #[test]
110 fn a_stamped_envelope_still_reads_as_an_ordinary_v2_envelope() {
111 let node = DeviceId::new(Uuid::from_u128(1));
112 let hlc = Hlc {
113 wall_ms: 9,
114 counter: 1,
115 node,
116 };
117 let stamped = SyncKitClient::hlc_envelope(&hlc, Some(&serde_json::json!(7)), Some(12));
118 assert_eq!(stamped["__skver"], serde_json::json!(2));
119 assert_eq!(
120 serde_json::from_value::<Hlc>(stamped["__skhlc"].clone()).unwrap(),
121 hlc
122 );
123 assert_eq!(stamped["data"], serde_json::json!(7));
124 }
125
126 /// Optional by construction: a malformed stamp must not fail a change that is
127 /// otherwise fine.
128 #[test]
129 fn a_malformed_stamp_reads_as_no_stamp() {
130 let node = DeviceId::new(Uuid::from_u128(1));
131 let hlc = Hlc {
132 wall_ms: 1,
133 counter: 0,
134 node,
135 };
136 for bad in [
137 serde_json::json!("four"),
138 serde_json::json!(-1),
139 serde_json::json!(null),
140 serde_json::json!(u64::from(u32::MAX) + 1),
141 ] {
142 let env = serde_json::json!({
143 "__skver": 2, "__skhlc": hlc, "data": null, "__sksv": bad
144 });
145 let (_, _, version) = SyncKitClient::split_hlc_envelope(env, node, 0).unwrap();
146 assert_eq!(version, None, "bad stamp {bad} should read as absent");
147 }
148 }
149
150 #[test]
151 fn split_envelope_rejects_unknown_version_loudly() {
152 // The X2 hazard: a future envelope version must error, not silently
153 // fall back to a bare-row read (which would corrupt the clock).
154 let node = DeviceId::new(Uuid::from_u128(1));
155 let hlc = Hlc {
156 wall_ms: 5,
157 counter: 0,
158 node,
159 };
160 let future = serde_json::json!({ "__skver": 3, "__skhlc": hlc, "data": null });
161 let err = SyncKitClient::split_hlc_envelope(future, node, 0).unwrap_err();
162 assert!(
163 matches!(err, SyncKitError::Crypto(ref m) if m.contains("envelope version 3")),
164 "unexpected error: {err:?}"
165 );
166 }
167
168 // ── encrypt_change / decrypt_change ──
169
170 #[test]
171 fn delete_seals_hlc_envelope_and_roundtrips() {
172 // A Delete carries no row payload, but its HLC must still travel, so it is
173 // now encrypted into an envelope (data = Some), and decrypt restores the
174 // op, a None payload, and the exact HLC.
175 let client = SyncKitClient::new(test_config());
176 let key = crypto::generate_master_key();
177 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
178
179 let device = DeviceId::new(Uuid::new_v4());
180 let hlc = Hlc {
181 wall_ms: 12_345,
182 counter: 7,
183 node: device,
184 };
185 let entry = ChangeEntry {
186 table: "tasks".to_string(),
187 op: ChangeOp::Delete,
188 row_id: "row-1".to_string(),
189 timestamp: Utc::now(),
190 hlc,
191 data: None,
192 extra: serde_json::Map::default(),
193 };
194
195 let wire = client.encrypt_change(entry).unwrap();
196 assert_eq!(wire.op, ChangeOp::Delete);
197 assert!(
198 wire.data.is_some(),
199 "delete now seals an encrypted HLC envelope"
200 );
201
202 let pull_entry = PullChangeEntry {
203 seq: 1,
204 device_id: device,
205 table: wire.table,
206 op: wire.op,
207 row_id: wire.row_id,
208 timestamp: wire.timestamp,
209 data: wire.data,
210 key_id: None,
211 gck_version: None,
212 };
213 let decrypted = client.decrypt_change(pull_entry).unwrap();
214 assert_eq!(decrypted.op, ChangeOp::Delete);
215 assert_eq!(decrypted.row_id, "row-1");
216 assert!(
217 decrypted.data.is_none(),
218 "payload is still None after the envelope unwraps"
219 );
220 assert_eq!(decrypted.hlc, hlc, "HLC survives the round trip");
221 }
222
223 #[test]
224 fn encrypt_change_fails_without_master_key() {
225 let client = SyncKitClient::new(test_config());
226 let entry = ChangeEntry {
227 table: "tasks".to_string(),
228 op: ChangeOp::Insert,
229 row_id: "row-1".to_string(),
230 timestamp: Utc::now(),
231 hlc: Hlc::zero(DeviceId::nil()),
232 data: Some(serde_json::json!({"title": "test"})),
233 extra: serde_json::Map::default(),
234 };
235
236 let err = client.encrypt_change(entry).unwrap_err();
237 assert!(matches!(err, SyncKitError::NoMasterKey));
238 }
239
240 #[test]
241 fn encrypt_change_produces_encrypted_data() {
242 let client = SyncKitClient::new(test_config());
243 let key = crypto::generate_master_key();
244 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
245
246 let original_data = serde_json::json!({"title": "Buy milk", "priority": 3});
247 let entry = ChangeEntry {
248 table: "tasks".to_string(),
249 op: ChangeOp::Insert,
250 row_id: "row-1".to_string(),
251 timestamp: Utc::now(),
252 hlc: Hlc::zero(DeviceId::nil()),
253 data: Some(original_data.clone()),
254 extra: serde_json::Map::default(),
255 };
256
257 let wire = client.encrypt_change(entry).unwrap();
258 assert!(wire.data.is_some());
259 let encrypted = wire.data.unwrap();
260 assert!(encrypted.is_string());
261 assert_ne!(encrypted, original_data);
262 }
263
264 #[test]
265 fn encrypt_decrypt_roundtrip() {
266 let client = SyncKitClient::new(test_config());
267 let key = crypto::generate_master_key();
268 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
269
270 let original_data = serde_json::json!({
271 "title": "Buy milk",
272 "tags": ["groceries", "urgent"],
273 "count": 42
274 });
275 let ts = Utc::now();
276 let entry = ChangeEntry {
277 table: "tasks".to_string(),
278 op: ChangeOp::Update,
279 row_id: "row-abc".to_string(),
280 timestamp: ts,
281 hlc: Hlc::zero(DeviceId::nil()),
282 data: Some(original_data.clone()),
283 extra: serde_json::Map::default(),
284 };
285
286 let wire = client.encrypt_change(entry).unwrap();
287 let pull_entry = PullChangeEntry {
288 seq: 1,
289 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
290 table: wire.table,
291 op: wire.op,
292 row_id: wire.row_id,
293 timestamp: wire.timestamp,
294 data: wire.data,
295 key_id: None,
296 gck_version: None,
297 };
298
299 let decrypted = client.decrypt_change(pull_entry).unwrap();
300 assert_eq!(decrypted.table, "tasks");
301 assert_eq!(decrypted.op, ChangeOp::Update);
302 assert_eq!(decrypted.row_id, "row-abc");
303 assert_eq!(decrypted.data.unwrap(), original_data);
304 }
305
306 #[test]
307 fn decrypt_change_with_no_data() {
308 let client = SyncKitClient::new(test_config());
309 let pull_entry = PullChangeEntry {
310 seq: 5,
311 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
312 table: "events".to_string(),
313 op: ChangeOp::Delete,
314 row_id: "evt-1".to_string(),
315 timestamp: Utc::now(),
316 data: None,
317 key_id: None,
318 gck_version: None,
319 };
320
321 let decrypted = client.decrypt_change(pull_entry).unwrap();
322 assert_eq!(decrypted.table, "events");
323 assert_eq!(decrypted.op, ChangeOp::Delete);
324 assert!(decrypted.data.is_none());
325 }
326
327 #[test]
328 fn decrypt_change_fails_without_master_key() {
329 let client = SyncKitClient::new(test_config());
330 let pull_entry = PullChangeEntry {
331 seq: 1,
332 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
333 table: "tasks".to_string(),
334 op: ChangeOp::Insert,
335 row_id: "row-1".to_string(),
336 timestamp: Utc::now(),
337 data: Some(serde_json::json!("some-encrypted-string")),
338 key_id: None,
339 gck_version: None,
340 };
341
342 let err = client.decrypt_change(pull_entry).unwrap_err();
343 assert!(matches!(err, SyncKitError::NoMasterKey));
344 }
345
346 // ── is_transient error classification ──
347
348 #[test]
349 fn is_transient_server_5xx() {
350 let err = SyncKitError::Server {
351 status: 500,
352 message: "Internal Server Error".to_string(),
353 retry_after_secs: None,
354 };
355 assert!(is_transient(&err));
356 let err = SyncKitError::Server {
357 status: 502,
358 message: "Bad Gateway".to_string(),
359 retry_after_secs: None,
360 };
361 assert!(is_transient(&err));
362 let err = SyncKitError::Server {
363 status: 503,
364 message: "Service Unavailable".to_string(),
365 retry_after_secs: None,
366 };
367 assert!(is_transient(&err));
368 let err = SyncKitError::Server {
369 status: 504,
370 message: "Gateway Timeout".to_string(),
371 retry_after_secs: None,
372 };
373 assert!(is_transient(&err));
374 }
375
376 #[test]
377 fn is_transient_rate_limited_429() {
378 let err = SyncKitError::Server {
379 status: 429,
380 message: "Too Many Requests".to_string(),
381 retry_after_secs: None,
382 };
383 assert!(is_transient(&err));
384 }
385
386 #[test]
387 fn is_not_transient_client_4xx() {
388 let err = SyncKitError::Server {
389 status: 400,
390 message: "Bad Request".to_string(),
391 retry_after_secs: None,
392 };
393 assert!(!is_transient(&err));
394 let err = SyncKitError::Server {
395 status: 401,
396 message: "Unauthorized".to_string(),
397 retry_after_secs: None,
398 };
399 assert!(!is_transient(&err));
400 let err = SyncKitError::Server {
401 status: 403,
402 message: "Forbidden".to_string(),
403 retry_after_secs: None,
404 };
405 assert!(!is_transient(&err));
406 let err = SyncKitError::Server {
407 status: 404,
408 message: "Not Found".to_string(),
409 retry_after_secs: None,
410 };
411 assert!(!is_transient(&err));
412 let err = SyncKitError::Server {
413 status: 409,
414 message: "Conflict".to_string(),
415 retry_after_secs: None,
416 };
417 assert!(!is_transient(&err));
418 let err = SyncKitError::Server {
419 status: 422,
420 message: "Unprocessable Entity".to_string(),
421 retry_after_secs: None,
422 };
423 assert!(!is_transient(&err));
424 }
425
426 #[test]
427 fn is_not_transient_not_authenticated() {
428 assert!(!is_transient(&SyncKitError::NotAuthenticated));
429 }
430
431 #[test]
432 fn is_not_transient_no_master_key() {
433 assert!(!is_transient(&SyncKitError::NoMasterKey));
434 }
435
436 #[test]
437 fn is_not_transient_decryption_failed() {
438 assert!(!is_transient(&SyncKitError::DecryptionFailed));
439 }
440
441 #[test]
442 fn is_not_transient_invalid_envelope() {
443 assert!(!is_transient(&SyncKitError::InvalidEnvelope(
444 "bad version".to_string()
445 )));
446 }
447
448 #[test]
449 fn is_not_transient_crypto() {
450 assert!(!is_transient(&SyncKitError::Crypto(
451 "encrypt failed".to_string()
452 )));
453 }
454
455 #[test]
456 fn is_not_transient_json() {
457 let err: SyncKitError = serde_json::from_str::<serde_json::Value>("not json")
458 .unwrap_err()
459 .into();
460 assert!(!is_transient(&err));
461 }
462
463 #[test]
464 fn is_not_transient_base64() {
465 let err: SyncKitError = base64::engine::general_purpose::STANDARD
466 .decode("!!!invalid!!!")
467 .unwrap_err()
468 .into();
469 assert!(!is_transient(&err));
470 }
471
472 #[test]
473 fn is_not_transient_token_expired() {
474 assert!(!is_transient(&SyncKitError::TokenExpired));
475 }
476
477 #[test]
478 fn is_not_transient_internal() {
479 assert!(!is_transient(&SyncKitError::Internal(
480 "lock poisoned".to_string()
481 )));
482 }
483
484 // ── Retry constants ──
485
486 #[test]
487 fn retry_constants_are_sensible() {
488 assert_eq!(MAX_RETRIES, 3);
489 assert_eq!(BASE_DELAY, Duration::from_secs(1));
490 }
491
492 #[test]
493 fn backoff_delays_are_exponential() {
494 let delay_0 = BASE_DELAY * 2u32.pow(0);
495 let delay_1 = BASE_DELAY * 2u32.pow(1);
496 let delay_2 = BASE_DELAY * 2u32.pow(2);
497
498 assert_eq!(delay_0, Duration::from_secs(1));
499 assert_eq!(delay_1, Duration::from_secs(2));
500 assert_eq!(delay_2, Duration::from_secs(4));
501 }
502
503 // ── is_transient boundary: 429 vs 428, 499 vs 500 ──
504
505 #[test]
506 fn is_transient_boundary_values() {
507 assert!(!is_transient(&SyncKitError::Server {
508 status: 428,
509 message: String::new(),
510 retry_after_secs: None
511 }));
512 assert!(is_transient(&SyncKitError::Server {
513 status: 429,
514 message: String::new(),
515 retry_after_secs: None
516 }));
517 assert!(!is_transient(&SyncKitError::Server {
518 status: 430,
519 message: String::new(),
520 retry_after_secs: None
521 }));
522 assert!(!is_transient(&SyncKitError::Server {
523 status: 499,
524 message: String::new(),
525 retry_after_secs: None
526 }));
527 assert!(is_transient(&SyncKitError::Server {
528 status: 500,
529 message: String::new(),
530 retry_after_secs: None
531 }));
532 }
533
534 // ── Token expiry detection ──
535
536 #[test]
537 fn jwt_exp_extracts_expiry() {
538 let exp = Utc::now().timestamp() + 3600;
539 let token = fake_jwt(exp);
540 assert_eq!(jwt_exp(&token), Some(exp));
541 }
542
543 #[test]
544 fn jwt_exp_returns_none_for_garbage() {
545 assert_eq!(jwt_exp("not-a-jwt"), None);
546 assert_eq!(jwt_exp("a.b.c"), None);
547 assert_eq!(jwt_exp(""), None);
548 }
549
550 #[test]
551 fn token_is_expired_for_past_exp() {
552 let token = fake_jwt(Utc::now().timestamp() - 3600);
553 assert!(token_is_expired(&token));
554 }
555
556 #[test]
557 fn token_is_expired_within_buffer() {
558 let token = fake_jwt(Utc::now().timestamp() + 10);
559 assert!(token_is_expired(&token));
560 }
561
562 #[test]
563 fn token_is_not_expired_when_fresh() {
564 let token = fake_jwt(Utc::now().timestamp() + 3600);
565 assert!(!token_is_expired(&token));
566 }
567
568 #[test]
569 fn token_is_not_expired_for_garbage() {
570 assert!(!token_is_expired("garbage"));
571 }
572
573 #[test]
574 fn token_expires_exactly_at_buffer_boundary() {
575 let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS);
576 assert!(token_is_expired(&token));
577 }
578
579 #[test]
580 fn token_expires_just_past_buffer() {
581 let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS + 1);
582 assert!(!token_is_expired(&token));
583 }
584
585 // ── encrypt_change preserves metadata ──
586
587 #[test]
588 fn encrypt_change_preserves_all_metadata() {
589 let client = SyncKitClient::new(test_config());
590 let key = crypto::generate_master_key();
591 client.set_master_key_raw(key);
592
593 let ts = Utc::now();
594 let entry = ChangeEntry {
595 table: "contacts".to_string(),
596 op: ChangeOp::Update,
597 row_id: "unique-row-id".to_string(),
598 timestamp: ts,
599 hlc: Hlc::zero(DeviceId::nil()),
600 data: Some(serde_json::json!({"name": "Alice"})),
601 extra: serde_json::Map::default(),
602 };
603
604 let wire = client.encrypt_change(entry).unwrap();
605 assert_eq!(wire.table, "contacts");
606 assert_eq!(wire.op, ChangeOp::Update);
607 assert_eq!(wire.row_id, "unique-row-id");
608 assert_eq!(wire.timestamp, ts);
609 }
610
611 // ── Multiple entries encrypt/decrypt ──
612
613 #[test]
614 fn multiple_entries_encrypt_decrypt_roundtrip() {
615 let client = SyncKitClient::new(test_config());
616 let key = crypto::generate_master_key();
617 client.set_master_key_raw(key);
618
619 let entries = [
620 ChangeEntry {
621 table: "tasks".to_string(),
622 op: ChangeOp::Insert,
623 row_id: "r1".to_string(),
624 timestamp: Utc::now(),
625 hlc: Hlc::zero(DeviceId::nil()),
626 data: Some(serde_json::json!({"title": "Task 1"})),
627 extra: serde_json::Map::default(),
628 },
629 ChangeEntry {
630 table: "tasks".to_string(),
631 op: ChangeOp::Update,
632 row_id: "r2".to_string(),
633 timestamp: Utc::now(),
634 hlc: Hlc::zero(DeviceId::nil()),
635 data: Some(serde_json::json!({"title": "Task 2", "done": true})),
636 extra: serde_json::Map::default(),
637 },
638 ChangeEntry {
639 table: "events".to_string(),
640 op: ChangeOp::Delete,
641 row_id: "r3".to_string(),
642 timestamp: Utc::now(),
643 hlc: Hlc::zero(DeviceId::nil()),
644 data: None,
645 extra: serde_json::Map::default(),
646 },
647 ];
648
649 let wire_entries: Vec<_> = entries
650 .iter()
651 .cloned()
652 .map(|e| client.encrypt_change(e).unwrap())
653 .collect();
654
655 assert_eq!(wire_entries.len(), 3);
656 assert!(wire_entries[0].data.is_some());
657 assert!(wire_entries[1].data.is_some());
658 // The Delete now also seals an encrypted HLC envelope (was None before HLC).
659 assert!(wire_entries[2].data.is_some());
660
661 for (i, wire) in wire_entries.into_iter().enumerate() {
662 let pull = PullChangeEntry {
663 seq: i as i64,
664 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
665 table: wire.table,
666 op: wire.op,
667 row_id: wire.row_id,
668 timestamp: wire.timestamp,
669 data: wire.data,
670 key_id: None,
671 gck_version: None,
672 };
673 let decrypted = client.decrypt_change(pull).unwrap();
674 assert_eq!(decrypted.table, entries[i].table);
675 assert_eq!(decrypted.op, entries[i].op);
676 assert_eq!(decrypted.data, entries[i].data);
677 }
678 }
679
680 // ── Unicode and edge-case roundtrips ──
681
682 #[test]
683 fn encrypt_decrypt_roundtrip_unicode_table() {
684 let client = SyncKitClient::new(test_config());
685 let key = crypto::generate_master_key();
686 client.set_master_key_raw(key);
687
688 let entry = ChangeEntry {
689 table: "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}".into(),
690 op: ChangeOp::Insert,
691 row_id: "row-1".into(),
692 timestamp: Utc::now(),
693 hlc: Hlc::zero(DeviceId::nil()),
694 data: Some(serde_json::json!({"name": "\u{30C6}\u{30B9}\u{30C8}"})),
695 extra: serde_json::Map::default(),
696 };
697
698 let wire = client.encrypt_change(entry).unwrap();
699 let pull = PullChangeEntry {
700 seq: 1,
701 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
702 table: wire.table,
703 op: wire.op,
704 row_id: wire.row_id,
705 timestamp: wire.timestamp,
706 data: wire.data,
707 key_id: None,
708 gck_version: None,
709 };
710 let decrypted = client.decrypt_change(pull).unwrap();
711 assert_eq!(
712 decrypted.table,
713 "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}"
714 );
715 }
716
717 #[test]
718 fn encrypt_decrypt_roundtrip_empty_row_id() {
719 let client = SyncKitClient::new(test_config());
720 let key = crypto::generate_master_key();
721 client.set_master_key_raw(key);
722
723 let entry = ChangeEntry {
724 table: "t".into(),
725 op: ChangeOp::Insert,
726 row_id: String::new(),
727 timestamp: Utc::now(),
728 hlc: Hlc::zero(DeviceId::nil()),
729 data: Some(serde_json::json!(42)),
730 extra: serde_json::Map::default(),
731 };
732
733 let wire = client.encrypt_change(entry).unwrap();
734 let pull = PullChangeEntry {
735 seq: 1,
736 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
737 table: wire.table,
738 op: wire.op,
739 row_id: wire.row_id,
740 timestamp: wire.timestamp,
741 data: wire.data,
742 key_id: None,
743 gck_version: None,
744 };
745 let decrypted = client.decrypt_change(pull).unwrap();
746 assert_eq!(decrypted.row_id, "");
747 assert_eq!(decrypted.data.unwrap(), serde_json::json!(42));
748 }
749
750 // ── decrypt_change_multi_key key selection ──
751 //
752 // Pins the key-id boundary logic that picks between primary and pending
753 // master keys during a rotation window. The mutations targeted here:
754 // * `effective_key_id == pending_key_id` (== ↔ !=)
755 // * `effective_key_id == primary_key_id || effective_key_id <= 1`
756 // (`||` ↔ `&&`, `<= 1` ↔ `< 1`/`<= 0`)
757 // * `entry.key_id.unwrap_or(1)` default
758 // * the Err-fallthrough that tries pending if primary fails
759
760 fn encrypt_with(key: &[u8; 32], value: &serde_json::Value) -> serde_json::Value {
761 crypto::encrypt_json(value, key).unwrap()
762 }
763
764 fn pull_entry_with(encrypted: serde_json::Value, key_id: Option<i32>) -> PullChangeEntry {
765 PullChangeEntry {
766 seq: 1,
767 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
768 table: "tasks".to_string(),
769 op: ChangeOp::Insert,
770 row_id: "row-multikey".to_string(),
771 timestamp: Utc::now(),
772 data: Some(encrypted),
773 key_id,
774 gck_version: None,
775 }
776 }
777
778 #[test]
779 fn multi_key_picks_pending_when_key_id_matches() {
780 let primary = crypto::generate_master_key();
781 let pending = crypto::generate_master_key();
782 let plaintext = serde_json::json!({"v": "pending-payload"});
783 let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(7));
784
785 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
786 entry,
787 &primary,
788 1,
789 &pending,
790 7,
791 &SyncKitClient::decrypt_change_with_key,
792 )
793 .unwrap();
794 assert_eq!(decrypted.data.unwrap(), plaintext);
795 }
796
797 #[test]
798 fn multi_key_picks_primary_when_key_id_matches_primary() {
799 let primary = crypto::generate_master_key();
800 let pending = crypto::generate_master_key();
801 let plaintext = serde_json::json!({"v": "primary-payload"});
802 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(3));
803
804 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
805 entry,
806 &primary,
807 3,
808 &pending,
809 9,
810 &SyncKitClient::decrypt_change_with_key,
811 )
812 .unwrap();
813 assert_eq!(decrypted.data.unwrap(), plaintext);
814 }
815
816 #[test]
817 fn multi_key_treats_missing_key_id_as_primary() {
818 // `entry.key_id.unwrap_or(1)` defaults to 1; `effective_key_id <= 1`
819 // arm routes to primary. A mutation changing `unwrap_or(1)` to
820 // `unwrap_or(99)` would route to "unknown" path and try primary anyway
821 // via the fallback, but a mutation to `<= 1` → `< 1` would skip the
822 // direct-primary branch and fall into the fallback path.
823 let primary = crypto::generate_master_key();
824 let pending = crypto::generate_master_key();
825 let plaintext = serde_json::json!({"v": "legacy-no-key-id"});
826 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), None);
827
828 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
829 entry,
830 &primary,
831 5,
832 &pending,
833 6,
834 &SyncKitClient::decrypt_change_with_key,
835 )
836 .unwrap();
837 assert_eq!(decrypted.data.unwrap(), plaintext);
838 }
839
840 #[test]
841 fn multi_key_unknown_key_id_falls_back_to_primary_first() {
842 // effective_key_id (42) matches neither primary (5) nor pending (6),
843 // and is > 1. The fallback first tries primary; here the data WAS
844 // encrypted with primary, so the fallback succeeds.
845 let primary = crypto::generate_master_key();
846 let pending = crypto::generate_master_key();
847 let plaintext = serde_json::json!({"v": "via-fallback-primary"});
848 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(42));
849
850 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
851 entry,
852 &primary,
853 5,
854 &pending,
855 6,
856 &SyncKitClient::decrypt_change_with_key,
857 )
858 .unwrap();
859 assert_eq!(decrypted.data.unwrap(), plaintext);
860 }
861
862 #[test]
863 fn multi_key_unknown_key_id_falls_back_to_pending_when_primary_fails() {
864 // Unknown key_id and the data was encrypted with pending, fallback
865 // must try primary first (fails), then pending (succeeds).
866 let primary = crypto::generate_master_key();
867 let pending = crypto::generate_master_key();
868 let plaintext = serde_json::json!({"v": "via-fallback-pending"});
869 let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(42));
870
871 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
872 entry,
873 &primary,
874 5,
875 &pending,
876 6,
877 &SyncKitClient::decrypt_change_with_key,
878 )
879 .unwrap();
880 assert_eq!(decrypted.data.unwrap(), plaintext);
881 }
882
883 // ── Backoff, jitter and Retry-After ──
884
885 fn ensure_crypto_provider() {
886 static PROVIDER: std::sync::Once = std::sync::Once::new();
887 PROVIDER.call_once(|| {
888 let _ = rustls::crypto::ring::default_provider().install_default();
889 });
890 }
891
892 fn server_err(status: u16, retry_after_secs: Option<u64>) -> SyncKitError {
893 SyncKitError::Server {
894 status,
895 message: String::new(),
896 retry_after_secs,
897 }
898 }
899
900 #[test]
901 fn retry_delay_uses_retry_after_capped_at_sixty_seconds() {
902 assert_eq!(
903 retry_delay(&server_err(429, Some(1)), 0),
904 Duration::from_secs(1)
905 );
906 assert_eq!(
907 retry_delay(&server_err(429, Some(7)), 3),
908 Duration::from_secs(7)
909 );
910 assert_eq!(
911 retry_delay(&server_err(429, Some(59)), 0),
912 Duration::from_secs(59)
913 );
914 assert_eq!(
915 retry_delay(&server_err(429, Some(60)), 0),
916 Duration::from_mins(1)
917 );
918 assert_eq!(
919 retry_delay(&server_err(429, Some(61)), 0),
920 Duration::from_mins(1)
921 );
922 assert_eq!(
923 retry_delay(&server_err(503, Some(86_400)), 2),
924 Duration::from_mins(1)
925 );
926 }
927
928 #[test]
929 fn retry_delay_without_retry_after_is_jittered_exponential() {
930 let errors = [
931 server_err(503, None),
932 server_err(429, None),
933 SyncKitError::Internal("no retry-after on this variant".to_string()),
934 ];
935 for attempt in 0..4u32 {
936 let base = BASE_DELAY * 2u32.pow(attempt);
937 let low = base.mul_f64(0.8);
938 let high = base.mul_f64(1.2);
939 for err in &errors {
940 for _ in 0..32 {
941 let delay = retry_delay(err, attempt);
942 assert!(
943 delay >= low && delay <= high,
944 "attempt {attempt}: {delay:?} outside [{low:?}, {high:?}]"
945 );
946 }
947 }
948 }
949 }
950
951 #[test]
952 fn retry_delay_grows_with_the_attempt() {
953 // The windows for consecutive attempts do not overlap, so the ordering
954 // holds for every draw rather than on average.
955 let err = server_err(503, None);
956 for attempt in 0..3u32 {
957 let here = retry_delay(&err, attempt);
958 let next = retry_delay(&err, attempt + 1);
959 assert!(
960 next > here,
961 "attempt {attempt}: {next:?} not above {here:?}"
962 );
963 }
964 }
965
966 #[test]
967 fn jittered_stays_within_twenty_percent_of_base() {
968 for base_ms in [5u64, 10, 1_000, 4_000, 60_000] {
969 let base = Duration::from_millis(base_ms);
970 let span = base_ms / 5;
971 let low = Duration::from_millis(base_ms - span);
972 let high = Duration::from_millis(base_ms + span);
973 for _ in 0..128 {
974 let got = jittered(base);
975 assert!(
976 got >= low && got <= high,
977 "{got:?} outside [{low:?}, {high:?}]"
978 );
979 }
980 }
981 }
982
983 #[test]
984 fn jittered_actually_spreads() {
985 let base = Duration::from_secs(1);
986 let first = jittered(base);
987 assert!(
988 (0..256).any(|_| jittered(base) != first),
989 "256 draws all returned {first:?}, so nothing is being spread"
990 );
991 }
992
993 #[test]
994 fn jittered_returns_base_when_the_span_rounds_to_zero() {
995 // Under 5ms the 20% span truncates to 0 and the base is returned as-is.
996 for base_ms in 0..5u64 {
997 let base = Duration::from_millis(base_ms);
998 assert_eq!(jittered(base), base);
999 }
1000 assert_eq!(jittered(Duration::ZERO), Duration::ZERO);
1001 }
1002
1003 /// Fetch a response carrying the given `Retry-After` header value (or none).
1004 /// The server is returned alongside so it outlives the response.
1005 async fn response_with_retry_after(
1006 header: Option<&str>,
1007 ) -> (wiremock::MockServer, reqwest::Response) {
1008 ensure_crypto_provider();
1009 let server = wiremock::MockServer::start().await;
1010 let mut template = wiremock::ResponseTemplate::new(429);
1011 if let Some(value) = header {
1012 template = template.insert_header("retry-after", value);
1013 }
1014 wiremock::Mock::given(wiremock::matchers::any())
1015 .respond_with(template)
1016 .mount(&server)
1017 .await;
1018 let resp = reqwest::get(server.uri()).await.unwrap();
1019 (server, resp)
1020 }
1021
1022 #[tokio::test]
1023 async fn parse_retry_after_reads_a_numeric_header() {
1024 let (_server, resp) = response_with_retry_after(Some("5")).await;
1025 assert_eq!(parse_retry_after(&resp), Some(5));
1026 }
1027
1028 #[tokio::test]
1029 async fn parse_retry_after_does_not_cap() {
1030 // The 60s cap belongs to retry_delay, not to parsing.
1031 let (_server, resp) = response_with_retry_after(Some("3600")).await;
1032 assert_eq!(parse_retry_after(&resp), Some(3600));
1033 }
1034
1035 #[tokio::test]
1036 async fn parse_retry_after_is_none_when_absent() {
1037 let (_server, resp) = response_with_retry_after(None).await;
1038 assert_eq!(parse_retry_after(&resp), None);
1039 }
1040
1041 #[tokio::test]
1042 async fn parse_retry_after_is_none_for_a_non_numeric_header() {
1043 for value in [
1044 "",
1045 "soon",
1046 "5s",
1047 "-3",
1048 "1.5",
1049 "Wed, 21 Oct 2015 07:28:00 GMT",
1050 ] {
1051 let (_server, resp) = response_with_retry_after(Some(value)).await;
1052 assert_eq!(parse_retry_after(&resp), None, "header {value:?}");
1053 }
1054 }
1055
1056 #[tokio::test]
1057 async fn parse_retry_after_is_none_for_zero() {
1058 let (_server, resp) = response_with_retry_after(Some("0")).await;
1059 assert_eq!(parse_retry_after(&resp), None);
1060 }
1061
1062 #[tokio::test]
1063 async fn check_response_carries_retry_after_onto_the_error() {
1064 let (_server, resp) = response_with_retry_after(Some("12")).await;
1065 let err = check_response(resp).await.unwrap_err();
1066 match err {
1067 SyncKitError::Server {
1068 status,
1069 retry_after_secs,
1070 ..
1071 } => {
1072 assert_eq!(status, 429);
1073 assert_eq!(retry_after_secs, Some(12));
1074 }
1075 other => panic!("expected a Server error, got {other:?}"),
1076 }
1077 // And the retry loop then uses it verbatim.
1078 assert_eq!(
1079 retry_delay(&server_err(429, Some(12)), 0),
1080 Duration::from_secs(12)
1081 );
1082 }
1083
1084 // ── is_transient over the Http arm ──
1085
1086 #[tokio::test]
1087 async fn is_transient_over_reqwest_error_kinds() {
1088 ensure_crypto_provider();
1089
1090 // A builder error is a programming mistake: an invalid header value.
1091 let builder = reqwest::Client::new()
1092 .get("http://example.invalid/")
1093 .header("x-bad", "line\nbreak")
1094 .build()
1095 .unwrap_err();
1096 assert!(builder.is_builder());
1097 assert!(!is_transient(&SyncKitError::Http(builder)));
1098
1099 // A redirect loop is permanent.
1100 let server = wiremock::MockServer::start().await;
1101 let uri = server.uri();
1102 wiremock::Mock::given(wiremock::matchers::any())
1103 .respond_with(wiremock::ResponseTemplate::new(302).insert_header("location", uri.as_str()))
1104 .mount(&server)
1105 .await;
1106 let redirect = reqwest::get(server.uri()).await.unwrap_err();
1107 assert!(redirect.is_redirect());
1108 assert!(!is_transient(&SyncKitError::Http(redirect)));
1109
1110 // A body that will not decode is permanent. The uncapped reader is
1111 // banned in production code; a decode error is what this test needs and
1112 // the body is eight bytes long.
1113 let json_server = wiremock::MockServer::start().await;
1114 wiremock::Mock::given(wiremock::matchers::any())
1115 .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
1116 .mount(&json_server)
1117 .await;
1118 #[allow(clippy::disallowed_methods)]
1119 let decode = {
1120 let resp = reqwest::get(json_server.uri()).await.unwrap();
1121 resp.json::<u32>().await.unwrap_err()
1122 };
1123 assert!(decode.is_decode());
1124 assert!(!is_transient(&SyncKitError::Http(decode)));
1125
1126 // A connect failure is the transient case the arm exists for.
1127 let connect = reqwest::get("http://127.0.0.1:1/").await.unwrap_err();
1128 assert!(connect.is_connect());
1129 assert!(!connect.is_builder() && !connect.is_redirect() && !connect.is_decode());
1130 assert!(is_transient(&SyncKitError::Http(connect)));
1131 }
1132
1133 // ── read_body_capped ──
1134
1135 async fn body_of_len(len: usize) -> (wiremock::MockServer, reqwest::Response) {
1136 ensure_crypto_provider();
1137 let server = wiremock::MockServer::start().await;
1138 wiremock::Mock::given(wiremock::matchers::any())
1139 .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(vec![b'x'; len]))
1140 .mount(&server)
1141 .await;
1142 let resp = reqwest::get(server.uri()).await.unwrap();
1143 (server, resp)
1144 }
1145
1146 #[tokio::test]
1147 async fn read_body_capped_returns_a_body_under_the_cap() {
1148 let (_server, resp) = body_of_len(100).await;
1149 let bytes = read_body_capped(resp, 1024).await.unwrap();
1150 assert_eq!(bytes.len(), 100);
1151 assert!(bytes.iter().all(|&b| b == b'x'));
1152 }
1153
1154 #[tokio::test]
1155 async fn read_body_capped_allows_a_body_exactly_at_the_cap() {
1156 // The check is `>`, so a body the size of the cap is legal.
1157 let (_server, resp) = body_of_len(256).await;
1158 let bytes = read_body_capped(resp, 256).await.unwrap();
1159 assert_eq!(bytes.len(), 256);
1160 }
1161
1162 #[tokio::test]
1163 async fn read_body_capped_rejects_an_oversized_content_length() {
1164 let (_server, resp) = body_of_len(2048).await;
1165 assert_eq!(resp.content_length(), Some(2048));
1166 let err = read_body_capped(resp, 1024).await.unwrap_err();
1167 match err {
1168 SyncKitError::Internal(msg) => {
1169 assert!(
1170 msg.contains("2048"),
1171 "message did not name the length: {msg}"
1172 );
1173 assert!(msg.contains("1024"), "message did not name the cap: {msg}");
1174 }
1175 other => panic!("expected Internal, got {other:?}"),
1176 }
1177 }
1178
1179 #[tokio::test]
1180 async fn read_text_capped_falls_back_to_empty_on_an_oversized_body() {
1181 let (_server, resp) = body_of_len(2048).await;
1182 assert_eq!(read_text_capped(resp, 16).await, String::new());
1183 }
1184
1185 #[tokio::test]
1186 async fn read_json_capped_propagates_the_cap_error() {
1187 let (_server, resp) = body_of_len(2048).await;
1188 let err = read_json_capped::<serde_json::Value>(resp, 16)
1189 .await
1190 .unwrap_err();
1191 assert!(matches!(err, SyncKitError::Internal(_)));
1192 }
1193
1194 // ── the retry loop's sleep, observed on a frozen clock ──
1195 //
1196 // Every other retry test counts requests, and a request count cannot see
1197 // `if attempt < max_attempts`: that guard gates only `tokio::time::sleep`,
1198 // so flipping it to `>` (never sleep) or `<=` (sleep once more, after the
1199 // final attempt) leaves the number of attempts untouched. These tests stamp
1200 // the paused clock inside the operation itself, which makes each gap
1201 // between attempts exactly the delay that was slept.
1202
1203 /// The gaps between consecutive clock stamps.
1204 fn gaps(stamps: &[tokio::time::Instant]) -> Vec<Duration> {
1205 stamps.windows(2).map(|w| w[1] - w[0]).collect()
1206 }
1207
1208 /// Assert a slept gap lands inside `jittered`'s +/-20% window for `base_ms`.
1209 /// The windows for 1s, 2s and 4s do not overlap, so a gap identifies which
1210 /// backoff step produced it.
1211 #[track_caller]
1212 fn assert_slept(got: Duration, base_ms: u64) {
1213 let low = Duration::from_millis(base_ms - base_ms / 5);
1214 let high = Duration::from_millis(base_ms + base_ms / 5);
1215 assert!(
1216 got >= low && got <= high,
1217 "slept {got:?}, outside the +/-20% window [{low:?}, {high:?}] around {base_ms}ms"
1218 );
1219 }
1220
1221 #[tokio::test(start_paused = true)]
1222 async fn retry_request_sleeps_before_each_replay_and_not_after_the_last() {
1223 let client = SyncKitClient::new(test_config());
1224 let stamps = std::sync::Mutex::new(Vec::new());
1225 let start = tokio::time::Instant::now();
1226
1227 let err = client
1228 .retry_request(Idempotency::ReadOnly, || {
1229 stamps.lock().unwrap().push(tokio::time::Instant::now());
1230 async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1231 })
1232 .await
1233 .unwrap_err();
1234 let end = tokio::time::Instant::now();
1235
1236 assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
1237 let stamps = stamps.into_inner().unwrap();
1238 assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
1239
1240 // Nothing is slept before the first try.
1241 assert_eq!(stamps[0] - start, Duration::ZERO);
1242 let gaps = gaps(&stamps);
1243 assert_slept(gaps[0], 1_000);
1244 assert_slept(gaps[1], 2_000);
1245 assert_slept(gaps[2], 4_000);
1246 // And nothing after the last: the loop gives up the instant the final
1247 // attempt fails. A guard that fired on `attempt == max_attempts` would
1248 // burn an 8s backoff here for no replay.
1249 assert_eq!(
1250 end - stamps[3],
1251 Duration::ZERO,
1252 "slept after the final attempt, which buys nothing"
1253 );
1254 }
1255
1256 #[tokio::test(start_paused = true)]
1257 async fn retry_request_neither_replays_nor_sleeps_an_unsafe_operation() {
1258 let client = SyncKitClient::new(test_config());
1259 let calls = std::sync::Mutex::new(0u32);
1260 let start = tokio::time::Instant::now();
1261
1262 let err = client
1263 .retry_request(Idempotency::Unsafe, || {
1264 *calls.lock().unwrap() += 1;
1265 async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1266 })
1267 .await
1268 .unwrap_err();
1269
1270 assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
1271 assert_eq!(*calls.lock().unwrap(), 1, "Unsafe gets exactly one attempt");
1272 // max_attempts is 0 here, so the sleep guard must not fire even once:
1273 // the caller is told the request failed with no delay bought for a
1274 // replay that is never going to happen.
1275 assert_eq!(tokio::time::Instant::now() - start, Duration::ZERO);
1276 }
1277
1278 #[tokio::test(start_paused = true)]
1279 async fn retry_request_json_sleeps_before_each_replay_and_not_after_the_last() {
1280 let client = SyncKitClient::new(test_config());
1281 let stamps = std::sync::Mutex::new(Vec::new());
1282 let start = tokio::time::Instant::now();
1283
1284 let err = client
1285 .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
1286 stamps.lock().unwrap().push(tokio::time::Instant::now());
1287 async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1288 })
1289 .await
1290 .unwrap_err();
1291 let end = tokio::time::Instant::now();
1292
1293 assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
1294 let stamps = stamps.into_inner().unwrap();
1295 assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
1296 assert_eq!(stamps[0] - start, Duration::ZERO);
1297 let gaps = gaps(&stamps);
1298 assert_slept(gaps[0], 1_000);
1299 assert_slept(gaps[1], 2_000);
1300 assert_slept(gaps[2], 4_000);
1301 assert_eq!(
1302 end - stamps[3],
1303 Duration::ZERO,
1304 "slept after the final attempt, which buys nothing"
1305 );
1306 }
1307
1308 #[tokio::test]
1309 async fn retry_request_json_sleeps_between_body_read_failures() {
1310 // The body-read arm has its own copy of the guard, reached only when
1311 // the request succeeds and the parse does not.
1312 ensure_crypto_provider();
1313 let server = wiremock::MockServer::start().await;
1314 wiremock::Mock::given(wiremock::matchers::any())
1315 .respond_with(wiremock::ResponseTemplate::new(200))
1316 .mount(&server)
1317 .await;
1318
1319 // Fetch every response BEFORE freezing the clock. An empty body is
1320 // Content-Length 0, so hyper ends the stream without another socket
1321 // read: inside the loop there is no I/O left, and the only thing that
1322 // can advance a paused clock is the loop's own sleep.
1323 let mut responses = std::collections::VecDeque::new();
1324 for _ in 0..4 {
1325 responses.push_back(reqwest::get(server.uri()).await.unwrap());
1326 }
1327 let responses = std::sync::Mutex::new(responses);
1328 let stamps = std::sync::Mutex::new(Vec::new());
1329
1330 let client = SyncKitClient::new(test_config());
1331 tokio::time::pause();
1332 let start = tokio::time::Instant::now();
1333 let err = client
1334 .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
1335 stamps.lock().unwrap().push(tokio::time::Instant::now());
1336 let resp = responses
1337 .lock()
1338 .unwrap()
1339 .pop_front()
1340 .expect("the loop asked for a fifth attempt");
1341 async move { Result::<reqwest::Response>::Ok(resp) }
1342 })
1343 .await
1344 .unwrap_err();
1345 let end = tokio::time::Instant::now();
1346
1347 // An empty body is not JSON, so every attempt fails in the parse.
1348 assert!(matches!(err, SyncKitError::Json(_)), "got {err:?}");
1349 let stamps = stamps.into_inner().unwrap();
1350 assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
1351 assert_eq!(stamps[0] - start, Duration::ZERO);
1352 let gaps = gaps(&stamps);
1353 assert_slept(gaps[0], 1_000);
1354 assert_slept(gaps[1], 2_000);
1355 assert_slept(gaps[2], 4_000);
1356 assert_eq!(
1357 end - stamps[3],
1358 Duration::ZERO,
1359 "slept after the final attempt, which buys nothing"
1360 );
1361 }
1362
1363 /// Serve one chunked HTTP/1.1 response carrying `len` bytes and no
1364 /// `Content-Length`. The whole response goes out in a single write so the
1365 /// body reaches `bytes_stream` as one item.
1366 async fn chunked_body_of_len(len: usize) -> reqwest::Response {
1367 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1368 ensure_crypto_provider();
1369 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1370 let addr = listener.local_addr().unwrap();
1371 tokio::spawn(async move {
1372 let (mut sock, _) = listener.accept().await.unwrap();
1373 // Drain the request head before replying. A single `read` is not
1374 // enough: TCP may split the request across segments, and answering
1375 // a half-read request races the client's own write.
1376 let mut request = Vec::new();
1377 let mut scratch = [0u8; 256];
1378 while !request.windows(4).any(|w| w == b"\r\n\r\n") {
1379 let n = sock.read(&mut scratch).await.unwrap();
1380 assert_ne!(n, 0, "the client closed before sending a request head");
1381 request.extend_from_slice(&scratch[..n]);
1382 }
1383 let mut out = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec();
1384 out.extend_from_slice(format!("{len:x}\r\n").as_bytes());
1385 out.extend_from_slice(&vec![b'x'; len]);
1386 out.extend_from_slice(b"\r\n0\r\n\r\n");
1387 sock.write_all(&out).await.unwrap();
1388 sock.flush().await.unwrap();
1389 });
1390 reqwest::get(format!("http://{addr}/")).await.unwrap()
1391 }
1392
1393 #[tokio::test]
1394 async fn read_body_capped_rejects_an_oversized_body_that_declares_no_length() {
1395 // Every other oversized case is caught by the Content-Length fast path,
1396 // which leaves the running `buf.len() + chunk.len()` total unobserved.
1397 // A chunked body has no declared length, so that accumulation is the
1398 // only thing standing between the cap and an unbounded read: with the
1399 // buffer still empty it is the incoming chunk's own size that has to
1400 // trip the limit.
1401 let resp = chunked_body_of_len(100).await;
1402 assert_eq!(
1403 resp.content_length(),
1404 None,
1405 "the fast path must not be what rejects this"
1406 );
1407 let err = read_body_capped(resp, 50).await.unwrap_err();
1408 match err {
1409 SyncKitError::Internal(msg) => {
1410 assert!(msg.contains("50"), "message did not name the cap: {msg}");
1411 }
1412 other => panic!("expected Internal, got {other:?}"),
1413 }
1414 }
1415