Skip to main content

max / synckit

117.6 KB · 3660 lines History Blame Raw
1 //! Integration tests using wiremock to simulate the MNW SyncKit server.
2 //!
3 //! These tests verify the full HTTP round-trip including retry behavior,
4 //! encryption/decryption, and error classification.
5
6 use std::sync::Arc;
7
8 use base64::Engine;
9 use chrono::Utc;
10 use serde_json::json;
11 use sha2::Digest;
12 use uuid::Uuid;
13 use wiremock::matchers::{method, path};
14 use wiremock::{Mock, MockServer, ResponseTemplate};
15
16 use std::time::Duration;
17 use synckit_client::{
18 AppId, ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
19 };
20
21 // ── Helpers ──
22
23 fn fake_jwt(exp: i64) -> String {
24 let header =
25 base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256","typ":"JWT"}"#);
26 let payload = json!({
27 "sub": "550e8400-e29b-41d4-a716-446655440000",
28 "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
29 "exp": exp,
30 "iat": exp - 3600,
31 });
32 let payload_b64 =
33 base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.to_string().as_bytes());
34 let sig = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
35 format!("{header}.{payload_b64}.{sig}")
36 }
37
38 fn fresh_token() -> String {
39 fake_jwt(Utc::now().timestamp() + 3600)
40 }
41
42 fn test_ids() -> (UserId, AppId) {
43 (
44 UserId::new(Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap()),
45 AppId::new(Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap()),
46 )
47 }
48
49 /// Install the rustls crypto provider once. reqwest is built `rustls-no-provider`,
50 /// so a real consumer app installs one at startup (audiofiles installs ring); these
51 /// tests have no such app, so they install ring themselves before building a client.
52 fn ensure_crypto_provider() {
53 static PROVIDER: std::sync::Once = std::sync::Once::new();
54 PROVIDER.call_once(|| {
55 let _ = rustls::crypto::ring::default_provider().install_default();
56 });
57 }
58
59 fn client_for(server: &MockServer) -> SyncKitClient {
60 ensure_crypto_provider();
61 SyncKitClient::new(SyncKitConfig {
62 server_url: server.uri(),
63 api_key: "test-api-key".to_string(),
64 })
65 }
66
67 fn authed_client(server: &MockServer) -> SyncKitClient {
68 let client = client_for(server);
69 let (user_id, app_id) = test_ids();
70 client.restore_session(&fresh_token(), user_id, app_id);
71 client
72 }
73
74 fn auth_response_json() -> serde_json::Value {
75 let (user_id, app_id) = test_ids();
76 json!({
77 "token": fresh_token(),
78 "user_id": user_id,
79 "app_id": app_id,
80 })
81 }
82
83 fn device_json() -> serde_json::Value {
84 let (user_id, app_id) = test_ids();
85 json!({
86 "id": Uuid::new_v4(),
87 "app_id": app_id,
88 "user_id": user_id,
89 "device_name": "Test Device",
90 "platform": "test",
91 "last_seen_at": "2025-01-01T00:00:00Z",
92 "created_at": "2025-01-01T00:00:00Z",
93 })
94 }
95
96 // ── Response body cap (DoS) ──
97
98 #[tokio::test]
99 async fn oversized_response_body_is_capped_not_buffered() {
100 // A hostile/buggy server streams a control-plane body far larger than the
101 // 8 MiB cap. The client must reject it (the cap fast-rejects on the honest
102 // Content-Length) instead of buffering it into memory and OOMing.
103 let server = MockServer::start().await;
104 let huge = vec![b'x'; 9 * 1024 * 1024];
105 Mock::given(method("GET"))
106 .and(path("/api/v1/sync/status"))
107 .respond_with(ResponseTemplate::new(200).set_body_bytes(huge))
108 .mount(&server)
109 .await;
110
111 let client = authed_client(&server);
112 let err = client.status().await.unwrap_err();
113 assert!(
114 matches!(err, SyncKitError::Internal(ref m) if m.contains("cap")),
115 "expected the body cap to reject the oversized response, got: {err:?}"
116 );
117 }
118
119 // ── Auth flow ──
120
121 #[tokio::test]
122 async fn authenticate_success_stores_session() {
123 let server = MockServer::start().await;
124
125 Mock::given(method("POST"))
126 .and(path("/api/v1/sync/auth"))
127 .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
128 .mount(&server)
129 .await;
130
131 let client = client_for(&server);
132 let (user_id, app_id) = client
133 .authenticate("user@test.com", "password", "test-key")
134 .await
135 .unwrap();
136
137 let info = client.session_info().expect("session stored");
138 assert_eq!(info.user_id, user_id);
139 assert_eq!(info.app_id, app_id);
140 }
141
142 #[tokio::test]
143 async fn authenticate_wrong_password_no_retry() {
144 let server = MockServer::start().await;
145
146 Mock::given(method("POST"))
147 .and(path("/api/v1/sync/auth"))
148 .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
149 .expect(1) // Must be called exactly once (no retry)
150 .mount(&server)
151 .await;
152
153 let client = client_for(&server);
154 let err = client
155 .authenticate("user@test.com", "wrong", "test-key")
156 .await
157 .unwrap_err();
158
159 assert!(
160 matches!(err, SyncKitError::Server { status: 401, .. }),
161 "Expected 401 error, got: {err:?}"
162 );
163 }
164
165 #[tokio::test]
166 async fn authenticate_retries_on_503() {
167 let server = MockServer::start().await;
168
169 Mock::given(method("POST"))
170 .and(path("/api/v1/sync/auth"))
171 .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
172 .up_to_n_times(1)
173 .mount(&server)
174 .await;
175
176 Mock::given(method("POST"))
177 .and(path("/api/v1/sync/auth"))
178 .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
179 .mount(&server)
180 .await;
181
182 let client = client_for(&server);
183 let result = client
184 .authenticate("user@test.com", "password", "test-key")
185 .await;
186 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
187 }
188
189 #[tokio::test]
190 async fn authenticate_with_code_success() {
191 let server = MockServer::start().await;
192
193 let (user_id, app_id) = test_ids();
194 Mock::given(method("POST"))
195 .and(path("/oauth/token"))
196 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
197 "access_token": fresh_token(),
198 "token_type": "Bearer",
199 "expires_in": 3600,
200 "user_id": user_id,
201 "app_id": app_id,
202 })))
203 .mount(&server)
204 .await;
205
206 let client = client_for(&server);
207 let (uid, aid) = client
208 .authenticate_with_code("auth-code", "verifier", 8080, "test-key")
209 .await
210 .unwrap();
211
212 assert_eq!(uid, user_id);
213 assert_eq!(aid, app_id);
214 assert!(client.session_info().is_some());
215 }
216
217 // ── Device management ──
218
219 #[tokio::test]
220 async fn register_device_success() {
221 let server = MockServer::start().await;
222
223 Mock::given(method("POST"))
224 .and(path("/api/v1/sync/devices"))
225 .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
226 .mount(&server)
227 .await;
228
229 let client = authed_client(&server);
230 let device = client.register_device("MacBook", "macos").await.unwrap();
231 assert_eq!(device.device_name, "Test Device");
232 }
233
234 #[tokio::test]
235 async fn register_device_retries_on_transient() {
236 let server = MockServer::start().await;
237
238 Mock::given(method("POST"))
239 .and(path("/api/v1/sync/devices"))
240 .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
241 .up_to_n_times(1)
242 .mount(&server)
243 .await;
244
245 Mock::given(method("POST"))
246 .and(path("/api/v1/sync/devices"))
247 .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
248 .mount(&server)
249 .await;
250
251 let client = authed_client(&server);
252 let result = client.register_device("MacBook", "macos").await;
253 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
254 }
255
256 #[tokio::test]
257 async fn list_devices_success() {
258 let server = MockServer::start().await;
259
260 Mock::given(method("GET"))
261 .and(path("/api/v1/sync/devices"))
262 .respond_with(ResponseTemplate::new(200).set_body_json(json!([device_json()])))
263 .mount(&server)
264 .await;
265
266 let client = authed_client(&server);
267 let devices = client.list_devices().await.unwrap();
268 assert_eq!(devices.len(), 1);
269 assert_eq!(devices[0].device_name, "Test Device");
270 }
271
272 // ── Push / Pull with encryption ──
273
274 #[tokio::test]
275 async fn push_encrypts_data() {
276 let server = MockServer::start().await;
277
278 Mock::given(method("POST"))
279 .and(path("/api/v1/sync/push"))
280 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
281 .mount(&server)
282 .await;
283
284 let client = authed_client(&server);
285 let key = synckit_client::crypto::generate_master_key();
286 client.set_master_key_raw(key);
287
288 let device_id = DeviceId::new(Uuid::new_v4());
289 let cursor = client
290 .push(
291 device_id,
292 vec![ChangeEntry {
293 table: "tasks".into(),
294 op: ChangeOp::Insert,
295 row_id: "row-1".into(),
296 timestamp: Utc::now(),
297 hlc: Hlc::zero(DeviceId::nil()),
298 data: Some(json!({"title": "Secret task"})),
299 extra: serde_json::Map::default(),
300 }],
301 )
302 .await
303 .unwrap();
304
305 assert_eq!(cursor, 1);
306
307 // Verify the request body was sent with encrypted data (not plaintext)
308 let requests = server.received_requests().await.unwrap();
309 let push_req = requests
310 .iter()
311 .find(|r| r.url.path() == "/api/v1/sync/push")
312 .unwrap();
313 let body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
314 let wire_data = body["changes"][0]["data"].as_str().unwrap();
315 assert!(
316 !wire_data.contains("Secret task"),
317 "Plaintext should not appear on the wire"
318 );
319 }
320
321 #[tokio::test]
322 async fn pull_decrypts_data() {
323 let server = MockServer::start().await;
324
325 let client = authed_client(&server);
326 let key = synckit_client::crypto::generate_master_key();
327 client.set_master_key_raw(key);
328
329 // Encrypt a value to simulate what the server would return
330 let plaintext = json!({"title": "Decrypted task"});
331 let encrypted = synckit_client::crypto::encrypt_json(&plaintext, &key).unwrap();
332
333 let device_id = DeviceId::new(Uuid::new_v4());
334 Mock::given(method("POST"))
335 .and(path("/api/v1/sync/pull"))
336 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
337 "changes": [{
338 "seq": 1,
339 "device_id": device_id,
340 "table": "tasks",
341 "op": "INSERT",
342 "row_id": "row-1",
343 "timestamp": "2025-06-01T12:00:00Z",
344 "data": encrypted,
345 }],
346 "cursor": 1,
347 "has_more": false,
348 })))
349 .mount(&server)
350 .await;
351
352 let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
353 assert_eq!(changes.len(), 1);
354 assert_eq!(cursor, 1);
355 assert!(!has_more);
356 assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
357 }
358
359 #[tokio::test]
360 async fn push_retries_on_503() {
361 let server = MockServer::start().await;
362
363 Mock::given(method("POST"))
364 .and(path("/api/v1/sync/push"))
365 .respond_with(ResponseTemplate::new(503))
366 .up_to_n_times(1)
367 .mount(&server)
368 .await;
369
370 Mock::given(method("POST"))
371 .and(path("/api/v1/sync/push"))
372 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 5})))
373 .mount(&server)
374 .await;
375
376 let client = authed_client(&server);
377 let key = synckit_client::crypto::generate_master_key();
378 client.set_master_key_raw(key);
379
380 let cursor = client
381 .push(DeviceId::new(Uuid::new_v4()), vec![])
382 .await
383 .unwrap();
384 assert_eq!(cursor, 5);
385 }
386
387 #[tokio::test]
388 async fn push_fails_immediately_on_401() {
389 let server = MockServer::start().await;
390
391 Mock::given(method("POST"))
392 .and(path("/api/v1/sync/push"))
393 .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
394 .expect(1)
395 .mount(&server)
396 .await;
397
398 let client = authed_client(&server);
399 let key = synckit_client::crypto::generate_master_key();
400 client.set_master_key_raw(key);
401
402 let err = client
403 .push(DeviceId::new(Uuid::new_v4()), vec![])
404 .await
405 .unwrap_err();
406 assert!(matches!(err, SyncKitError::Server { status: 401, .. }));
407 }
408
409 #[tokio::test]
410 async fn pull_with_has_more_pagination() {
411 let server = MockServer::start().await;
412
413 let client = authed_client(&server);
414 let key = synckit_client::crypto::generate_master_key();
415 client.set_master_key_raw(key);
416
417 let device_id = DeviceId::new(Uuid::new_v4());
418
419 // First pull: has_more = true
420 Mock::given(method("POST"))
421 .and(path("/api/v1/sync/pull"))
422 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
423 "changes": [],
424 "cursor": 50,
425 "has_more": true,
426 })))
427 .up_to_n_times(1)
428 .mount(&server)
429 .await;
430
431 let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
432 assert!(changes.is_empty());
433 assert_eq!(cursor, 50);
434 assert!(has_more);
435
436 // Second pull from cursor 50: has_more = false
437 Mock::given(method("POST"))
438 .and(path("/api/v1/sync/pull"))
439 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
440 "changes": [],
441 "cursor": 100,
442 "has_more": false,
443 })))
444 .mount(&server)
445 .await;
446
447 let (_, cursor2, has_more2) = client.pull(device_id, 50).await.unwrap();
448 assert_eq!(cursor2, 100);
449 assert!(!has_more2);
450 }
451
452 // ── Blob operations ──
453
454 #[tokio::test]
455 async fn blob_upload_url_success() {
456 let server = MockServer::start().await;
457
458 Mock::given(method("POST"))
459 .and(path("/api/v1/sync/blobs/upload"))
460 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
461 "upload_url": "https://s3.example.com/put",
462 "already_exists": false,
463 })))
464 .mount(&server)
465 .await;
466
467 let client = authed_client(&server);
468 let resp = client.blob_upload_url("sha256-abc", 1024).await.unwrap();
469 assert_eq!(resp.upload_url, "https://s3.example.com/put");
470 assert!(!resp.already_exists);
471 }
472
473 #[tokio::test]
474 async fn blob_upload_url_declares_the_length_the_put_will_carry() {
475 // The server signs the declared size into the presigned URL as
476 // Content-Length, a SignedHeader, so declaring anything other than the
477 // exact ciphertext length makes the PUT fail SigV4. The caller passes the
478 // plaintext size it sees on disk; the SDK converts. This test pins the two
479 // halves together, which is the only place the mismatch would show up:
480 // wiremock does not verify signatures, and the server's own tests use an
481 // in-memory backend that does not sign at all.
482 let server = MockServer::start().await;
483
484 let upload_path = "/s3/sized-upload";
485 Mock::given(method("POST"))
486 .and(path("/api/v1/sync/blobs/upload"))
487 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
488 "upload_url": format!("{}{upload_path}", server.uri()),
489 "already_exists": false,
490 })))
491 .mount(&server)
492 .await;
493 Mock::given(method("PUT"))
494 .and(path(upload_path))
495 .respond_with(ResponseTemplate::new(200))
496 .mount(&server)
497 .await;
498
499 let client = authed_client(&server);
500 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
501
502 // Spans two chunks, so the framing overhead is more than a single chunk's.
503 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE + 500))
504 .map(|i| i as u8)
505 .collect();
506 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
507
508 let resp = client
509 .blob_upload_url(&hash, plaintext.len() as i64)
510 .await
511 .unwrap();
512 client
513 .blob_upload(&hash, &resp.upload_url, plaintext.clone())
514 .await
515 .unwrap();
516
517 let reqs = server.received_requests().await.unwrap();
518 let declared: serde_json::Value = reqs
519 .iter()
520 .find(|r| r.url.path() == "/api/v1/sync/blobs/upload")
521 .unwrap()
522 .body_json()
523 .unwrap();
524 let put_len = reqs
525 .iter()
526 .find(|r| r.url.path() == upload_path)
527 .unwrap()
528 .body
529 .len();
530
531 assert_eq!(
532 declared["size_bytes"].as_u64().unwrap(),
533 put_len as u64,
534 "the declared size must equal the bytes actually PUT, or the signature fails"
535 );
536 assert!(
537 put_len > plaintext.len(),
538 "the PUT carries ciphertext, which is longer than the plaintext"
539 );
540 }
541
542 #[tokio::test]
543 async fn blob_upload_encrypts_data() {
544 let server = MockServer::start().await;
545
546 let upload_path = "/s3/upload";
547 Mock::given(method("PUT"))
548 .and(path(upload_path))
549 .respond_with(ResponseTemplate::new(200))
550 .mount(&server)
551 .await;
552
553 let client = authed_client(&server);
554 let key = synckit_client::crypto::generate_master_key();
555 client.set_master_key_raw(key);
556
557 let plaintext = b"hello blob data";
558 let presigned = format!("{}{}", server.uri(), upload_path);
559 client
560 .blob_upload("sha256-test", &presigned, plaintext.to_vec())
561 .await
562 .unwrap();
563
564 // Verify uploaded body is encrypted (not plaintext)
565 let requests = server.received_requests().await.unwrap();
566 let upload_req = requests
567 .iter()
568 .find(|r| r.url.path() == upload_path)
569 .unwrap();
570 assert!(
571 !upload_req
572 .body
573 .windows(plaintext.len())
574 .any(|w| w == plaintext),
575 "Plaintext should not appear in uploaded body"
576 );
577 // Encrypted blob should be larger due to nonce + tag overhead
578 assert!(upload_req.body.len() > plaintext.len());
579 }
580
581 #[tokio::test]
582 async fn blob_download_decrypts_data() {
583 let server = MockServer::start().await;
584
585 let client = authed_client(&server);
586 let key = synckit_client::crypto::generate_master_key();
587 client.set_master_key_raw(key);
588
589 // Encrypt data to simulate what S3 would return. A legacy (untagged) blob
590 // still decrypts through the AAD-aware reader and must pass the hash check.
591 let plaintext = b"decrypted blob content";
592 let hash = hex::encode(sha2::Sha256::digest(plaintext));
593 let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();
594
595 let download_path = "/s3/download";
596 Mock::given(method("GET"))
597 .and(path(download_path))
598 .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
599 .mount(&server)
600 .await;
601
602 let presigned = format!("{}{}", server.uri(), download_path);
603 let result = client.blob_download(&hash, &presigned).await.unwrap();
604 assert_eq!(result, plaintext);
605 }
606
607 #[tokio::test]
608 async fn blob_upload_retries_on_503() {
609 let server = MockServer::start().await;
610
611 let upload_path = "/s3/retry-upload";
612 Mock::given(method("PUT"))
613 .and(path(upload_path))
614 .respond_with(ResponseTemplate::new(503))
615 .up_to_n_times(1)
616 .mount(&server)
617 .await;
618
619 Mock::given(method("PUT"))
620 .and(path(upload_path))
621 .respond_with(ResponseTemplate::new(200))
622 .mount(&server)
623 .await;
624
625 let client = authed_client(&server);
626 let key = synckit_client::crypto::generate_master_key();
627 client.set_master_key_raw(key);
628
629 let presigned = format!("{}{}", server.uri(), upload_path);
630 let result = client
631 .blob_upload("sha256-x", &presigned, b"data".to_vec())
632 .await;
633 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
634 }
635
636 // ── Token handling ──
637
638 #[tokio::test]
639 async fn expired_jwt_returns_token_expired() {
640 let server = MockServer::start().await;
641 let client = client_for(&server);
642 let (user_id, app_id) = test_ids();
643
644 let expired = fake_jwt(Utc::now().timestamp() - 3600);
645 client.restore_session(&expired, user_id, app_id);
646
647 let err = client.status().await.unwrap_err();
648 assert!(
649 matches!(err, SyncKitError::TokenExpired),
650 "Expected TokenExpired, got: {err:?}"
651 );
652 }
653
654 #[tokio::test]
655 async fn near_expiry_jwt_returns_token_expired() {
656 let server = MockServer::start().await;
657 let client = client_for(&server);
658 let (user_id, app_id) = test_ids();
659
660 // Token expires in 10 seconds (within 30-second buffer)
661 let near_expiry = fake_jwt(Utc::now().timestamp() + 10);
662 client.restore_session(&near_expiry, user_id, app_id);
663
664 let err = client.status().await.unwrap_err();
665 assert!(
666 matches!(err, SyncKitError::TokenExpired),
667 "Expected TokenExpired, got: {err:?}"
668 );
669 }
670
671 // ── Session management ──
672
673 #[tokio::test]
674 async fn restore_then_clear_session() {
675 let server = MockServer::start().await;
676
677 Mock::given(method("GET"))
678 .and(path("/api/v1/sync/status"))
679 .respond_with(
680 ResponseTemplate::new(200)
681 .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
682 )
683 .mount(&server)
684 .await;
685
686 let client = authed_client(&server);
687
688 // Should work while authenticated
689 let status = client.status().await.unwrap();
690 assert_eq!(status.total_changes, 0);
691
692 // Clear session
693 client.clear_session();
694
695 // Now should fail
696 let err = client.status().await.unwrap_err();
697 assert!(matches!(err, SyncKitError::NotAuthenticated));
698 }
699
700 #[tokio::test]
701 async fn status_without_auth_returns_not_authenticated() {
702 let server = MockServer::start().await;
703 let client = client_for(&server);
704
705 let err = client.status().await.unwrap_err();
706 assert!(matches!(err, SyncKitError::NotAuthenticated));
707 }
708
709 #[tokio::test]
710 async fn push_without_auth_returns_not_authenticated() {
711 let server = MockServer::start().await;
712 let client = client_for(&server);
713
714 let err = client
715 .push(DeviceId::new(Uuid::new_v4()), vec![])
716 .await
717 .unwrap_err();
718 assert!(matches!(err, SyncKitError::NotAuthenticated));
719 }
720
721 // ── Key management ──
722
723 #[tokio::test]
724 async fn has_server_key_true_on_200() {
725 let server = MockServer::start().await;
726
727 Mock::given(method("GET"))
728 .and(path("/api/v1/sync/keys"))
729 .respond_with(
730 ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope-data"})),
731 )
732 .mount(&server)
733 .await;
734
735 let client = authed_client(&server);
736 assert!(client.has_server_key().await.unwrap());
737 }
738
739 #[tokio::test]
740 async fn has_server_key_false_on_404() {
741 let server = MockServer::start().await;
742
743 Mock::given(method("GET"))
744 .and(path("/api/v1/sync/keys"))
745 .respond_with(ResponseTemplate::new(404))
746 .mount(&server)
747 .await;
748
749 let client = authed_client(&server);
750 assert!(!client.has_server_key().await.unwrap());
751 }
752
753 #[tokio::test]
754 async fn has_server_key_retries_on_500() {
755 let server = MockServer::start().await;
756
757 Mock::given(method("GET"))
758 .and(path("/api/v1/sync/keys"))
759 .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
760 .up_to_n_times(1)
761 .mount(&server)
762 .await;
763
764 Mock::given(method("GET"))
765 .and(path("/api/v1/sync/keys"))
766 .respond_with(
767 ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope"})),
768 )
769 .mount(&server)
770 .await;
771
772 let client = authed_client(&server);
773 assert!(client.has_server_key().await.unwrap());
774 }
775
776 // ── Concurrent access ──
777
778 #[tokio::test]
779 async fn concurrent_push_pull_no_panics() {
780 let server = MockServer::start().await;
781
782 Mock::given(method("POST"))
783 .and(path("/api/v1/sync/push"))
784 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
785 .mount(&server)
786 .await;
787
788 let device_id = DeviceId::new(Uuid::new_v4());
789 Mock::given(method("POST"))
790 .and(path("/api/v1/sync/pull"))
791 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
792 "changes": [],
793 "cursor": 0,
794 "has_more": false,
795 })))
796 .mount(&server)
797 .await;
798
799 let client = Arc::new(authed_client(&server));
800 let key = synckit_client::crypto::generate_master_key();
801 client.set_master_key_raw(key);
802
803 let mut handles = Vec::new();
804 for _ in 0..4 {
805 let c = Arc::clone(&client);
806 let did = device_id;
807 handles.push(tokio::spawn(async move {
808 let _ = c.push(did, vec![]).await;
809 let _ = c.pull(did, 0).await;
810 }));
811 }
812
813 for h in handles {
814 h.await.unwrap(); // No panics
815 }
816 }
817
818 // ── Error classification ──
819
820 #[tokio::test]
821 async fn error_429_is_retried() {
822 let server = MockServer::start().await;
823
824 Mock::given(method("GET"))
825 .and(path("/api/v1/sync/status"))
826 .respond_with(ResponseTemplate::new(429).set_body_string("Too Many Requests"))
827 .up_to_n_times(1)
828 .mount(&server)
829 .await;
830
831 Mock::given(method("GET"))
832 .and(path("/api/v1/sync/status"))
833 .respond_with(
834 ResponseTemplate::new(200)
835 .set_body_json(json!({"total_changes": 10, "latest_cursor": 5})),
836 )
837 .mount(&server)
838 .await;
839
840 let client = authed_client(&server);
841 let status = client.status().await.unwrap();
842 assert_eq!(status.total_changes, 10);
843 }
844
845 #[tokio::test]
846 async fn error_400_not_retried() {
847 let server = MockServer::start().await;
848
849 Mock::given(method("POST"))
850 .and(path("/api/v1/sync/devices"))
851 .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
852 .expect(1)
853 .mount(&server)
854 .await;
855
856 let client = authed_client(&server);
857 let err = client.register_device("Device", "test").await.unwrap_err();
858 assert!(matches!(err, SyncKitError::Server { status: 400, .. }));
859 }
860
861 // ── Status endpoint ──
862
863 #[tokio::test]
864 async fn status_success() {
865 let server = MockServer::start().await;
866
867 Mock::given(method("GET"))
868 .and(path("/api/v1/sync/status"))
869 .respond_with(
870 ResponseTemplate::new(200)
871 .set_body_json(json!({"total_changes": 42, "latest_cursor": 100})),
872 )
873 .mount(&server)
874 .await;
875
876 let client = authed_client(&server);
877 let status = client.status().await.unwrap();
878 assert_eq!(status.total_changes, 42);
879 assert_eq!(status.latest_cursor, Some(100));
880 }
881
882 #[tokio::test]
883 async fn status_retries_on_transient() {
884 let server = MockServer::start().await;
885
886 Mock::given(method("GET"))
887 .and(path("/api/v1/sync/status"))
888 .respond_with(ResponseTemplate::new(504).set_body_string("Gateway Timeout"))
889 .up_to_n_times(1)
890 .mount(&server)
891 .await;
892
893 Mock::given(method("GET"))
894 .and(path("/api/v1/sync/status"))
895 .respond_with(
896 ResponseTemplate::new(200)
897 .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
898 )
899 .mount(&server)
900 .await;
901
902 let client = authed_client(&server);
903 let status = client.status().await.unwrap();
904 assert_eq!(status.total_changes, 0);
905 }
906
907 // ── Blob confirm ──
908
909 #[tokio::test]
910 async fn blob_confirm_success() {
911 let server = MockServer::start().await;
912
913 Mock::given(method("POST"))
914 .and(path("/api/v1/sync/blobs/confirm"))
915 .respond_with(ResponseTemplate::new(200))
916 .mount(&server)
917 .await;
918
919 let client = authed_client(&server);
920 client.blob_confirm("sha256-abc", 1024).await.unwrap();
921 }
922
923 // ── Blob download URL ──
924
925 #[tokio::test]
926 async fn blob_download_url_success() {
927 let server = MockServer::start().await;
928
929 Mock::given(method("POST"))
930 .and(path("/api/v1/sync/blobs/download"))
931 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
932 "download_url": "https://s3.example.com/get",
933 })))
934 .mount(&server)
935 .await;
936
937 let client = authed_client(&server);
938 let url = client.blob_download_url("sha256-abc").await.unwrap();
939 assert_eq!(url, "https://s3.example.com/get");
940 }
941
942 // ── change_password: CRITICAL bug fix tests ──
943
944 /// Helper: set up a server that serves a wrapped master key envelope.
945 /// Returns (client, master_key, envelope_json).
946 fn setup_change_password_test(
947 server: &MockServer,
948 password: &str,
949 ) -> (SyncKitClient, [u8; 32], String) {
950 let client = authed_client(server);
951 let master_key = synckit_client::crypto::generate_master_key();
952 let envelope = synckit_client::crypto::wrap_master_key(&master_key, password).unwrap();
953
954 // Cache the master key in the client (simulating normal logged-in state)
955 client.set_master_key_raw(master_key);
956
957 (client, master_key, envelope)
958 }
959
960 #[tokio::test]
961 async fn change_password_wrong_old_password_with_cached_key_fails() {
962 let server = MockServer::start().await;
963 let (client, _master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");
964
965 // Server returns the envelope on GET
966 Mock::given(method("GET"))
967 .and(path("/api/v1/sync/keys"))
968 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
969 .mount(&server)
970 .await;
971
972 // Attempt to change password with wrong old password.
973 // The key IS cached, but the old password must still be validated.
974 let result = client.change_password("wrong-old-pass", "new-pass").await;
975
976 assert!(
977 result.is_err(),
978 "change_password must fail when old_password is wrong, even with cached key"
979 );
980 assert!(
981 matches!(result.unwrap_err(), SyncKitError::DecryptionFailed),
982 "Should get DecryptionFailed for wrong old password"
983 );
984 }
985
986 #[tokio::test]
987 async fn change_password_correct_old_password_with_cached_key_succeeds() {
988 let server = MockServer::start().await;
989 let (client, master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");
990
991 // Server returns the envelope on GET
992 Mock::given(method("GET"))
993 .and(path("/api/v1/sync/keys"))
994 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
995 .mount(&server)
996 .await;
997
998 // Server accepts the new envelope on PUT
999 Mock::given(method("PUT"))
1000 .and(path("/api/v1/sync/keys"))
1001 .respond_with(ResponseTemplate::new(200))
1002 .mount(&server)
1003 .await;
1004
1005 let result = client.change_password("correct-old-pass", "new-pass").await;
1006 assert!(
1007 result.is_ok(),
1008 "change_password should succeed with correct old password"
1009 );
1010
1011 // Verify the PUT request was made (new envelope was uploaded)
1012 let requests = server.received_requests().await.unwrap();
1013 let put_requests: Vec<_> = requests
1014 .iter()
1015 .filter(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
1016 .collect();
1017 assert_eq!(put_requests.len(), 1, "Should have sent exactly one PUT");
1018
1019 // Verify the new envelope can be unwrapped with the new password
1020 let put_body: serde_json::Value = serde_json::from_slice(&put_requests[0].body).unwrap();
1021 let new_envelope = put_body["encrypted_key"].as_str().unwrap();
1022 let recovered = synckit_client::crypto::unwrap_master_key(new_envelope, "new-pass").unwrap();
1023 assert_eq!(
1024 recovered, master_key,
1025 "New envelope should unwrap to the same master key"
1026 );
1027 }
1028
1029 #[tokio::test]
1030 async fn change_password_wrong_old_password_without_cached_key_fails() {
1031 let server = MockServer::start().await;
1032 let master_key = synckit_client::crypto::generate_master_key();
1033 let envelope =
1034 synckit_client::crypto::wrap_master_key(&master_key, "correct-old-pass").unwrap();
1035
1036 let client = authed_client(&server);
1037 // Deliberately NOT setting master key: no cached key
1038
1039 Mock::given(method("GET"))
1040 .and(path("/api/v1/sync/keys"))
1041 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1042 .mount(&server)
1043 .await;
1044
1045 let result = client.change_password("wrong-old-pass", "new-pass").await;
1046
1047 assert!(
1048 result.is_err(),
1049 "change_password must fail with wrong old password even without cached key"
1050 );
1051 assert!(matches!(
1052 result.unwrap_err(),
1053 SyncKitError::DecryptionFailed
1054 ));
1055 }
1056
1057 #[tokio::test]
1058 async fn change_password_old_envelope_invalid_with_new_password() {
1059 let server = MockServer::start().await;
1060 let (client, _master_key, envelope) = setup_change_password_test(&server, "old-pass");
1061
1062 Mock::given(method("GET"))
1063 .and(path("/api/v1/sync/keys"))
1064 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1065 .mount(&server)
1066 .await;
1067
1068 Mock::given(method("PUT"))
1069 .and(path("/api/v1/sync/keys"))
1070 .respond_with(ResponseTemplate::new(200))
1071 .mount(&server)
1072 .await;
1073
1074 client
1075 .change_password("old-pass", "new-pass")
1076 .await
1077 .unwrap();
1078
1079 // Old password should NOT work on the new envelope
1080 let requests = server.received_requests().await.unwrap();
1081 let put_req = requests
1082 .iter()
1083 .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
1084 .unwrap();
1085 let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
1086 let new_envelope = body["encrypted_key"].as_str().unwrap();
1087
1088 let result = synckit_client::crypto::unwrap_master_key(new_envelope, "old-pass");
1089 assert!(
1090 result.is_err(),
1091 "Old password must not work on the new envelope"
1092 );
1093 }
1094
1095 // ── Empty changelog push ──
1096
1097 #[tokio::test]
1098 async fn push_empty_changes_succeeds() {
1099 let server = MockServer::start().await;
1100
1101 Mock::given(method("POST"))
1102 .and(path("/api/v1/sync/push"))
1103 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 0})))
1104 .mount(&server)
1105 .await;
1106
1107 let client = authed_client(&server);
1108 let key = synckit_client::crypto::generate_master_key();
1109 client.set_master_key_raw(key);
1110
1111 let cursor = client
1112 .push(DeviceId::new(Uuid::new_v4()), vec![])
1113 .await
1114 .unwrap();
1115 assert_eq!(cursor, 0);
1116 }
1117
1118 // ── Malformed server responses ──
1119
1120 #[tokio::test]
1121 async fn push_malformed_json_response_handled() {
1122 let server = MockServer::start().await;
1123
1124 Mock::given(method("POST"))
1125 .and(path("/api/v1/sync/push"))
1126 .respond_with(ResponseTemplate::new(200).set_body_string("not valid json at all"))
1127 .mount(&server)
1128 .await;
1129
1130 let client = authed_client(&server);
1131 let key = synckit_client::crypto::generate_master_key();
1132 client.set_master_key_raw(key);
1133
1134 let result = client.push(DeviceId::new(Uuid::new_v4()), vec![]).await;
1135 assert!(result.is_err(), "Malformed JSON should produce an error");
1136 // Should be a JSON parse error, not a panic
1137 let err = result.unwrap_err();
1138 assert!(
1139 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
1140 "Expected Http or Json error for malformed response, got: {err:?}"
1141 );
1142 }
1143
1144 #[tokio::test]
1145 async fn pull_malformed_json_response_handled() {
1146 let server = MockServer::start().await;
1147
1148 Mock::given(method("POST"))
1149 .and(path("/api/v1/sync/pull"))
1150 .respond_with(ResponseTemplate::new(200).set_body_string("{invalid json}"))
1151 .mount(&server)
1152 .await;
1153
1154 let client = authed_client(&server);
1155 let key = synckit_client::crypto::generate_master_key();
1156 client.set_master_key_raw(key);
1157
1158 let result = client.pull(DeviceId::new(Uuid::new_v4()), 0).await;
1159 assert!(result.is_err(), "Malformed JSON should produce an error");
1160 }
1161
1162 #[tokio::test]
1163 async fn status_malformed_json_response_handled() {
1164 let server = MockServer::start().await;
1165
1166 Mock::given(method("GET"))
1167 .and(path("/api/v1/sync/status"))
1168 .respond_with(ResponseTemplate::new(200).set_body_string("this is not json"))
1169 .mount(&server)
1170 .await;
1171
1172 let client = authed_client(&server);
1173 let result = client.status().await;
1174 assert!(result.is_err());
1175 }
1176
1177 // ── Server error messages preserved ──
1178
1179 #[tokio::test]
1180 async fn server_error_message_preserved() {
1181 let server = MockServer::start().await;
1182
1183 Mock::given(method("GET"))
1184 .and(path("/api/v1/sync/status"))
1185 .respond_with(
1186 ResponseTemplate::new(422).set_body_string("Validation failed: missing field"),
1187 )
1188 .mount(&server)
1189 .await;
1190
1191 let client = authed_client(&server);
1192 let err = client.status().await.unwrap_err();
1193 match err {
1194 SyncKitError::Server {
1195 status, message, ..
1196 } => {
1197 assert_eq!(status, 422);
1198 assert!(
1199 message.contains("Validation failed"),
1200 "Error message should be preserved: {message}"
1201 );
1202 }
1203 other => panic!("Expected Server error, got: {other:?}"),
1204 }
1205 }
1206
1207 // ── Concurrent operations ──
1208
1209 #[tokio::test]
1210 async fn concurrent_push_operations_no_data_corruption() {
1211 let server = MockServer::start().await;
1212
1213 Mock::given(method("POST"))
1214 .and(path("/api/v1/sync/push"))
1215 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
1216 .mount(&server)
1217 .await;
1218
1219 let client = Arc::new(authed_client(&server));
1220 let key = synckit_client::crypto::generate_master_key();
1221 client.set_master_key_raw(key);
1222
1223 let mut handles = Vec::new();
1224 for i in 0..8 {
1225 let c = Arc::clone(&client);
1226 handles.push(tokio::spawn(async move {
1227 let device_id = DeviceId::new(Uuid::new_v4());
1228 let entry = ChangeEntry {
1229 table: format!("table_{i}"),
1230 op: ChangeOp::Insert,
1231 row_id: format!("row_{i}"),
1232 timestamp: Utc::now(),
1233 hlc: Hlc::zero(DeviceId::nil()),
1234 data: Some(json!({"index": i})),
1235 extra: serde_json::Map::default(),
1236 };
1237 c.push(device_id, vec![entry]).await
1238 }));
1239 }
1240
1241 for h in handles {
1242 let result = h.await.unwrap();
1243 assert!(result.is_ok(), "Concurrent push should succeed: {result:?}");
1244 }
1245 }
1246
1247 #[tokio::test]
1248 async fn concurrent_push_and_pull_interleaved() {
1249 let server = MockServer::start().await;
1250 let device_id = DeviceId::new(Uuid::new_v4());
1251
1252 Mock::given(method("POST"))
1253 .and(path("/api/v1/sync/push"))
1254 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 10})))
1255 .mount(&server)
1256 .await;
1257
1258 Mock::given(method("POST"))
1259 .and(path("/api/v1/sync/pull"))
1260 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
1261 "changes": [],
1262 "cursor": 10,
1263 "has_more": false,
1264 })))
1265 .mount(&server)
1266 .await;
1267
1268 let client = Arc::new(authed_client(&server));
1269 let key = synckit_client::crypto::generate_master_key();
1270 client.set_master_key_raw(key);
1271
1272 let mut handles = Vec::new();
1273 for i in 0..4 {
1274 let c = Arc::clone(&client);
1275 let did = device_id;
1276 handles.push(tokio::spawn(async move {
1277 // Alternate push and pull
1278 if i % 2 == 0 {
1279 c.push(did, vec![]).await.map(|_| ())
1280 } else {
1281 c.pull(did, 0).await.map(|_| ())
1282 }
1283 }));
1284 }
1285
1286 for h in handles {
1287 let result = h.await.unwrap();
1288 assert!(
1289 result.is_ok(),
1290 "Interleaved push/pull should succeed: {result:?}"
1291 );
1292 }
1293 }
1294
1295 // ── Large payload handling ──
1296
1297 #[tokio::test]
1298 async fn push_many_changes_succeeds() {
1299 let server = MockServer::start().await;
1300
1301 Mock::given(method("POST"))
1302 .and(path("/api/v1/sync/push"))
1303 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1000})))
1304 .mount(&server)
1305 .await;
1306
1307 let client = authed_client(&server);
1308 let key = synckit_client::crypto::generate_master_key();
1309 client.set_master_key_raw(key);
1310
1311 // Create 1000+ change entries
1312 let changes: Vec<ChangeEntry> = (0..1100)
1313 .map(|i| ChangeEntry {
1314 table: "bulk_table".into(),
1315 op: ChangeOp::Insert,
1316 row_id: format!("row-{i}"),
1317 timestamp: Utc::now(),
1318 hlc: Hlc::zero(DeviceId::nil()),
1319 data: Some(json!({"index": i, "value": format!("data-{i}")})),
1320 extra: serde_json::Map::default(),
1321 })
1322 .collect();
1323
1324 let cursor = client
1325 .push(DeviceId::new(Uuid::new_v4()), changes)
1326 .await
1327 .unwrap();
1328 assert_eq!(cursor, 1000);
1329 }
1330
1331 // ── Session expiry handling ──
1332
1333 #[tokio::test]
1334 async fn expired_token_detected_before_push() {
1335 let server = MockServer::start().await;
1336 let client = client_for(&server);
1337 let (user_id, app_id) = test_ids();
1338
1339 let expired = fake_jwt(Utc::now().timestamp() - 100);
1340 client.restore_session(&expired, user_id, app_id);
1341 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
1342
1343 let err = client
1344 .push(DeviceId::new(Uuid::new_v4()), vec![])
1345 .await
1346 .unwrap_err();
1347 assert!(
1348 matches!(err, SyncKitError::TokenExpired),
1349 "Expired token should be detected pre-flight, got: {err:?}"
1350 );
1351 }
1352
1353 #[tokio::test]
1354 async fn expired_token_detected_before_pull() {
1355 let server = MockServer::start().await;
1356 let client = client_for(&server);
1357 let (user_id, app_id) = test_ids();
1358
1359 let expired = fake_jwt(Utc::now().timestamp() - 100);
1360 client.restore_session(&expired, user_id, app_id);
1361 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
1362
1363 let err = client
1364 .pull(DeviceId::new(Uuid::new_v4()), 0)
1365 .await
1366 .unwrap_err();
1367 assert!(matches!(err, SyncKitError::TokenExpired));
1368 }
1369
1370 #[tokio::test]
1371 async fn expired_token_detected_before_register_device() {
1372 let server = MockServer::start().await;
1373 let client = client_for(&server);
1374 let (user_id, app_id) = test_ids();
1375
1376 let expired = fake_jwt(Utc::now().timestamp() - 100);
1377 client.restore_session(&expired, user_id, app_id);
1378
1379 let err = client.register_device("Test", "test").await.unwrap_err();
1380 assert!(matches!(err, SyncKitError::TokenExpired));
1381 }
1382
1383 #[tokio::test]
1384 async fn expired_token_detected_before_list_devices() {
1385 let server = MockServer::start().await;
1386 let client = client_for(&server);
1387 let (user_id, app_id) = test_ids();
1388
1389 let expired = fake_jwt(Utc::now().timestamp() - 100);
1390 client.restore_session(&expired, user_id, app_id);
1391
1392 let err = client.list_devices().await.unwrap_err();
1393 assert!(matches!(err, SyncKitError::TokenExpired));
1394 }
1395
1396 // ── Blob edge cases ──
1397
1398 #[tokio::test]
1399 async fn blob_upload_zero_byte_data() {
1400 let server = MockServer::start().await;
1401
1402 let upload_path = "/s3/zero-byte";
1403 Mock::given(method("PUT"))
1404 .and(path(upload_path))
1405 .respond_with(ResponseTemplate::new(200))
1406 .mount(&server)
1407 .await;
1408
1409 let client = authed_client(&server);
1410 let key = synckit_client::crypto::generate_master_key();
1411 client.set_master_key_raw(key);
1412
1413 let presigned = format!("{}{}", server.uri(), upload_path);
1414 let result = client.blob_upload("sha256-empty", &presigned, vec![]).await;
1415 assert!(result.is_ok(), "Zero-byte blob upload should succeed");
1416
1417 // Verify the uploaded data is the v3 chunked framing over an empty blob.
1418 let requests = server.received_requests().await.unwrap();
1419 let req = requests
1420 .iter()
1421 .find(|r| r.url.path() == upload_path)
1422 .unwrap();
1423 assert_eq!(
1424 req.body.len(),
1425 synckit_client::crypto::chunked_blob_overhead(0),
1426 "Empty plaintext should produce exactly the chunked overhead bytes"
1427 );
1428 }
1429
1430 #[tokio::test]
1431 async fn blob_upload_download_roundtrip() {
1432 let server = MockServer::start().await;
1433
1434 let client = authed_client(&server);
1435 let key = synckit_client::crypto::generate_master_key();
1436 client.set_master_key_raw(key);
1437
1438 let plaintext = b"roundtrip blob data with special bytes \x00\xFF\x01";
1439 let hash = hex::encode(sha2::Sha256::digest(plaintext));
1440
1441 // Upload
1442 let upload_path = "/s3/roundtrip-upload";
1443 Mock::given(method("PUT"))
1444 .and(path(upload_path))
1445 .respond_with(ResponseTemplate::new(200))
1446 .mount(&server)
1447 .await;
1448
1449 client
1450 .blob_upload(
1451 &hash,
1452 &format!("{}{}", server.uri(), upload_path),
1453 plaintext.to_vec(),
1454 )
1455 .await
1456 .unwrap();
1457
1458 // Capture what was uploaded
1459 let requests = server.received_requests().await.unwrap();
1460 let uploaded_body = &requests
1461 .iter()
1462 .find(|r| r.url.path() == upload_path)
1463 .unwrap()
1464 .body;
1465
1466 // Serve that exact encrypted data back for download
1467 let download_path = "/s3/roundtrip-download";
1468 Mock::given(method("GET"))
1469 .and(path(download_path))
1470 .respond_with(ResponseTemplate::new(200).set_body_bytes(uploaded_body.clone()))
1471 .mount(&server)
1472 .await;
1473
1474 let downloaded = client
1475 .blob_download(&hash, &format!("{}{}", server.uri(), download_path))
1476 .await
1477 .unwrap();
1478
1479 assert_eq!(downloaded, plaintext, "Blob roundtrip must preserve data");
1480 }
1481
1482 // ── Push without master key ──
1483
1484 #[tokio::test]
1485 async fn push_with_data_fails_without_master_key() {
1486 let server = MockServer::start().await;
1487 let client = authed_client(&server);
1488 // No master key set
1489
1490 let changes = vec![ChangeEntry {
1491 table: "tasks".into(),
1492 op: ChangeOp::Insert,
1493 row_id: "r1".into(),
1494 timestamp: Utc::now(),
1495 hlc: Hlc::zero(DeviceId::nil()),
1496 data: Some(json!({"title": "test"})),
1497 extra: serde_json::Map::default(),
1498 }];
1499
1500 let err = client
1501 .push(DeviceId::new(Uuid::new_v4()), changes)
1502 .await
1503 .unwrap_err();
1504 assert!(
1505 matches!(err, SyncKitError::NoMasterKey),
1506 "Push with data should fail without master key: {err:?}"
1507 );
1508 }
1509
1510 #[tokio::test]
1511 async fn push_delete_requires_master_key() {
1512 // Deletes used to push without a key (no payload to encrypt). With HLC, a
1513 // Delete now seals its clock into an encrypted envelope, so the master key is
1514 // required for every push, Deletes included.
1515 let server = MockServer::start().await;
1516
1517 Mock::given(method("POST"))
1518 .and(path("/api/v1/sync/push"))
1519 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
1520 .mount(&server)
1521 .await;
1522
1523 let client = authed_client(&server);
1524 // No master key loaded.
1525
1526 let changes = vec![ChangeEntry {
1527 table: "tasks".into(),
1528 op: ChangeOp::Delete,
1529 row_id: "r1".into(),
1530 timestamp: Utc::now(),
1531 hlc: Hlc::zero(DeviceId::nil()),
1532 data: None,
1533 extra: serde_json::Map::default(),
1534 }];
1535
1536 let err = client
1537 .push(DeviceId::new(Uuid::new_v4()), changes)
1538 .await
1539 .unwrap_err();
1540 assert!(
1541 matches!(err, SyncKitError::NoMasterKey),
1542 "Delete now seals an HLC envelope and needs the key: {err:?}"
1543 );
1544 }
1545
1546 // ── Blob operations require auth ──
1547
1548 #[tokio::test]
1549 async fn blob_upload_url_without_auth_fails() {
1550 let server = MockServer::start().await;
1551 let client = client_for(&server);
1552
1553 let result = client.blob_upload_url("hash", 100).await;
1554 match result {
1555 Err(SyncKitError::NotAuthenticated) => {} // expected
1556 Err(other) => panic!("Expected NotAuthenticated, got: {other:?}"),
1557 Ok(_) => panic!("Expected NotAuthenticated error, got Ok"),
1558 }
1559 }
1560
1561 #[tokio::test]
1562 async fn blob_confirm_without_auth_fails() {
1563 let server = MockServer::start().await;
1564 let client = client_for(&server);
1565
1566 let err = client.blob_confirm("hash", 100).await.unwrap_err();
1567 assert!(matches!(err, SyncKitError::NotAuthenticated));
1568 }
1569
1570 #[tokio::test]
1571 async fn blob_download_url_without_auth_fails() {
1572 let server = MockServer::start().await;
1573 let client = client_for(&server);
1574
1575 let err = client.blob_download_url("hash").await.unwrap_err();
1576 assert!(matches!(err, SyncKitError::NotAuthenticated));
1577 }
1578
1579 // ── Double-push same data ──
1580
1581 #[tokio::test]
1582 async fn double_push_same_data_both_succeed() {
1583 let server = MockServer::start().await;
1584
1585 // Server returns incrementing cursors
1586 Mock::given(method("POST"))
1587 .and(path("/api/v1/sync/push"))
1588 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
1589 .up_to_n_times(1)
1590 .mount(&server)
1591 .await;
1592
1593 Mock::given(method("POST"))
1594 .and(path("/api/v1/sync/push"))
1595 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 2})))
1596 .mount(&server)
1597 .await;
1598
1599 let client = authed_client(&server);
1600 let key = synckit_client::crypto::generate_master_key();
1601 client.set_master_key_raw(key);
1602
1603 let entry = ChangeEntry {
1604 table: "tasks".into(),
1605 op: ChangeOp::Insert,
1606 row_id: "same-row".into(),
1607 timestamp: Utc::now(),
1608 hlc: Hlc::zero(DeviceId::nil()),
1609 data: Some(json!({"title": "duplicate push test"})),
1610 extra: serde_json::Map::default(),
1611 };
1612
1613 let cursor1 = client
1614 .push(DeviceId::new(Uuid::new_v4()), vec![entry.clone()])
1615 .await
1616 .unwrap();
1617 let cursor2 = client
1618 .push(DeviceId::new(Uuid::new_v4()), vec![entry])
1619 .await
1620 .unwrap();
1621
1622 assert_eq!(cursor1, 1);
1623 assert_eq!(cursor2, 2);
1624 }
1625
1626 // ── Encryption setup ──
1627
1628 #[tokio::test]
1629 async fn setup_encryption_new_stores_key_and_uploads_envelope() {
1630 let server = MockServer::start().await;
1631
1632 Mock::given(method("PUT"))
1633 .and(path("/api/v1/sync/keys"))
1634 .respond_with(ResponseTemplate::new(200))
1635 .expect(1)
1636 .mount(&server)
1637 .await;
1638
1639 let client = authed_client(&server);
1640 assert!(!client.has_master_key());
1641
1642 client.setup_encryption_new("test-password").await.unwrap();
1643
1644 // Master key should now be in memory
1645 assert!(client.has_master_key());
1646
1647 // Verify the PUT body contains a valid envelope unwrappable with the same password
1648 let requests = server.received_requests().await.unwrap();
1649 let put_req = requests
1650 .iter()
1651 .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
1652 .unwrap();
1653 let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
1654 let envelope_str = body["encrypted_key"].as_str().unwrap();
1655 let recovered =
1656 synckit_client::crypto::unwrap_master_key(envelope_str, "test-password").unwrap();
1657 assert_eq!(recovered.len(), 32);
1658 }
1659
1660 #[tokio::test]
1661 async fn setup_encryption_new_without_auth_fails() {
1662 let server = MockServer::start().await;
1663 let client = client_for(&server);
1664
1665 let err = client.setup_encryption_new("password").await.unwrap_err();
1666 assert!(matches!(err, SyncKitError::NotAuthenticated));
1667 }
1668
1669 #[tokio::test]
1670 async fn setup_encryption_new_retries_on_server_error() {
1671 let server = MockServer::start().await;
1672
1673 Mock::given(method("PUT"))
1674 .and(path("/api/v1/sync/keys"))
1675 .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
1676 .up_to_n_times(1)
1677 .mount(&server)
1678 .await;
1679
1680 Mock::given(method("PUT"))
1681 .and(path("/api/v1/sync/keys"))
1682 .respond_with(ResponseTemplate::new(200))
1683 .mount(&server)
1684 .await;
1685
1686 let client = authed_client(&server);
1687 let result = client.setup_encryption_new("password").await;
1688 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
1689 assert!(client.has_master_key());
1690 }
1691
1692 #[tokio::test]
1693 async fn setup_encryption_existing_recovers_key() {
1694 let server = MockServer::start().await;
1695
1696 let master_key = synckit_client::crypto::generate_master_key();
1697 let envelope = synckit_client::crypto::wrap_master_key(&master_key, "my-password").unwrap();
1698
1699 Mock::given(method("GET"))
1700 .and(path("/api/v1/sync/keys"))
1701 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1702 .mount(&server)
1703 .await;
1704
1705 let client = authed_client(&server);
1706 assert!(!client.has_master_key());
1707
1708 client
1709 .setup_encryption_existing("my-password")
1710 .await
1711 .unwrap();
1712
1713 assert!(client.has_master_key());
1714 }
1715
1716 #[tokio::test]
1717 async fn setup_encryption_existing_wrong_password_fails() {
1718 let server = MockServer::start().await;
1719
1720 let master_key = synckit_client::crypto::generate_master_key();
1721 let envelope =
1722 synckit_client::crypto::wrap_master_key(&master_key, "correct-password").unwrap();
1723
1724 Mock::given(method("GET"))
1725 .and(path("/api/v1/sync/keys"))
1726 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1727 .mount(&server)
1728 .await;
1729
1730 let client = authed_client(&server);
1731 let err = client
1732 .setup_encryption_existing("wrong-password")
1733 .await
1734 .unwrap_err();
1735 assert!(
1736 matches!(err, SyncKitError::DecryptionFailed),
1737 "Wrong password should produce DecryptionFailed: {err:?}"
1738 );
1739 assert!(!client.has_master_key());
1740 }
1741
1742 #[tokio::test]
1743 async fn setup_encryption_existing_without_auth_fails() {
1744 let server = MockServer::start().await;
1745 let client = client_for(&server);
1746
1747 let err = client
1748 .setup_encryption_existing("password")
1749 .await
1750 .unwrap_err();
1751 assert!(matches!(err, SyncKitError::NotAuthenticated));
1752 }
1753
1754 #[tokio::test]
1755 async fn setup_encryption_existing_retries_on_server_error() {
1756 let server = MockServer::start().await;
1757
1758 let master_key = synckit_client::crypto::generate_master_key();
1759 let envelope = synckit_client::crypto::wrap_master_key(&master_key, "password").unwrap();
1760
1761 Mock::given(method("GET"))
1762 .and(path("/api/v1/sync/keys"))
1763 .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
1764 .up_to_n_times(1)
1765 .mount(&server)
1766 .await;
1767
1768 Mock::given(method("GET"))
1769 .and(path("/api/v1/sync/keys"))
1770 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1771 .mount(&server)
1772 .await;
1773
1774 let client = authed_client(&server);
1775 let result = client.setup_encryption_existing("password").await;
1776 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
1777 assert!(client.has_master_key());
1778 }
1779
1780 #[tokio::test]
1781 async fn setup_encryption_existing_no_server_key_returns_error() {
1782 let server = MockServer::start().await;
1783
1784 Mock::given(method("GET"))
1785 .and(path("/api/v1/sync/keys"))
1786 .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
1787 .mount(&server)
1788 .await;
1789
1790 let client = authed_client(&server);
1791 let err = client
1792 .setup_encryption_existing("password")
1793 .await
1794 .unwrap_err();
1795 assert!(
1796 matches!(err, SyncKitError::Server { status: 404, .. }),
1797 "Missing server key should produce 404 error: {err:?}"
1798 );
1799 }
1800
1801 /// Two-device roundtrip: device 1 generates key via setup_encryption_new,
1802 /// device 2 recovers it via setup_encryption_existing. Data encrypted by
1803 /// device 1 must be decryptable by device 2.
1804 #[tokio::test]
1805 async fn encryption_setup_cross_device_roundtrip() {
1806 let server = MockServer::start().await;
1807
1808 // Device 1: setup_encryption_new
1809 Mock::given(method("PUT"))
1810 .and(path("/api/v1/sync/keys"))
1811 .respond_with(ResponseTemplate::new(200))
1812 .mount(&server)
1813 .await;
1814
1815 Mock::given(method("POST"))
1816 .and(path("/api/v1/sync/push"))
1817 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
1818 .mount(&server)
1819 .await;
1820
1821 let client1 = authed_client(&server);
1822 client1
1823 .setup_encryption_new("shared-password")
1824 .await
1825 .unwrap();
1826
1827 // Push encrypted data from device 1
1828 let device_id = DeviceId::new(Uuid::new_v4());
1829 let original_data = json!({"title": "cross-device test", "secret": true});
1830 client1
1831 .push(
1832 device_id,
1833 vec![ChangeEntry {
1834 table: "tasks".into(),
1835 op: ChangeOp::Insert,
1836 row_id: "cross-r1".into(),
1837 timestamp: Utc::now(),
1838 hlc: Hlc::zero(DeviceId::nil()),
1839 data: Some(original_data.clone()),
1840 extra: serde_json::Map::default(),
1841 }],
1842 )
1843 .await
1844 .unwrap();
1845
1846 // Capture the envelope and encrypted data
1847 let requests = server.received_requests().await.unwrap();
1848 let put_req = requests
1849 .iter()
1850 .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
1851 .unwrap();
1852 let put_body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
1853 let envelope = put_body["encrypted_key"].as_str().unwrap().to_string();
1854
1855 let push_req = requests
1856 .iter()
1857 .find(|r| r.url.path() == "/api/v1/sync/push")
1858 .unwrap();
1859 let push_body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
1860 let encrypted_data = push_body["changes"][0]["data"].clone();
1861
1862 // Device 2: setup_encryption_existing with same password
1863 server.reset().await;
1864
1865 Mock::given(method("GET"))
1866 .and(path("/api/v1/sync/keys"))
1867 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
1868 .mount(&server)
1869 .await;
1870
1871 Mock::given(method("POST"))
1872 .and(path("/api/v1/sync/pull"))
1873 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
1874 "changes": [{
1875 "seq": 1,
1876 "device_id": device_id,
1877 "table": "tasks",
1878 "op": "INSERT",
1879 "row_id": "cross-r1",
1880 "timestamp": "2025-06-01T12:00:00Z",
1881 "data": encrypted_data,
1882 }],
1883 "cursor": 1,
1884 "has_more": false,
1885 })))
1886 .mount(&server)
1887 .await;
1888
1889 let client2 = authed_client(&server);
1890 client2
1891 .setup_encryption_existing("shared-password")
1892 .await
1893 .unwrap();
1894
1895 // Pull and decrypt with device 2's recovered key
1896 let (changes, _, _) = client2.pull(device_id, 0).await.unwrap();
1897 assert_eq!(changes.len(), 1);
1898 assert_eq!(
1899 changes[0].data.as_ref().unwrap(),
1900 &original_data,
1901 "Data encrypted by device 1 must be decryptable by device 2"
1902 );
1903 }
1904
1905 // ── Config persistence (SyncKitConfig serialization) ──
1906
1907 #[tokio::test]
1908 async fn config_serialization_roundtrip() {
1909 let config = SyncKitConfig {
1910 server_url: "https://makenot.work".to_string(),
1911 api_key: "ak_test_12345".to_string(),
1912 };
1913
1914 // SyncKitConfig derives Clone and Debug; verify round trip through Debug
1915 let debug = format!("{config:?}");
1916 assert!(debug.contains("makenot.work"));
1917 assert!(debug.contains("ak_test_12345"));
1918
1919 let cloned = config.clone();
1920 assert_eq!(cloned.server_url, config.server_url);
1921 assert_eq!(cloned.api_key, config.api_key);
1922 }
1923
1924 // ── API error mapping ──
1925
1926 #[tokio::test]
1927 async fn all_4xx_error_codes_mapped() {
1928 let server = MockServer::start().await;
1929
1930 for status_code in [400, 401, 403, 404, 409, 422] {
1931 // Reset mocks
1932 server.reset().await;
1933
1934 Mock::given(method("GET"))
1935 .and(path("/api/v1/sync/status"))
1936 .respond_with(
1937 ResponseTemplate::new(status_code).set_body_string(format!("Error {status_code}")),
1938 )
1939 .mount(&server)
1940 .await;
1941
1942 let client = authed_client(&server);
1943 let err = client.status().await.unwrap_err();
1944
1945 match err {
1946 SyncKitError::Server {
1947 status, message, ..
1948 } => {
1949 assert_eq!(status, status_code);
1950 assert!(message.contains(&format!("Error {status_code}")));
1951 }
1952 other => panic!("Status {status_code} should map to Server error, got: {other:?}"),
1953 }
1954 }
1955 }
1956
1957 #[tokio::test]
1958 async fn all_5xx_error_codes_retried() {
1959 for status_code in [500, 502, 503, 504] {
1960 let server = MockServer::start().await;
1961
1962 // First request fails with 5xx
1963 Mock::given(method("GET"))
1964 .and(path("/api/v1/sync/status"))
1965 .respond_with(ResponseTemplate::new(status_code).set_body_string("Server Error"))
1966 .up_to_n_times(1)
1967 .mount(&server)
1968 .await;
1969
1970 // Second request succeeds
1971 Mock::given(method("GET"))
1972 .and(path("/api/v1/sync/status"))
1973 .respond_with(
1974 ResponseTemplate::new(200)
1975 .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
1976 )
1977 .mount(&server)
1978 .await;
1979
1980 let client = authed_client(&server);
1981 let result = client.status().await;
1982 assert!(
1983 result.is_ok(),
1984 "Status {status_code} should be retried and succeed: {result:?}"
1985 );
1986 }
1987 }
1988
1989 // ── Blob download with wrong key ──
1990
1991 #[tokio::test]
1992 async fn blob_download_with_wrong_key_fails() {
1993 let server = MockServer::start().await;
1994
1995 let key1 = synckit_client::crypto::generate_master_key();
1996 let key2 = synckit_client::crypto::generate_master_key();
1997
1998 // Encrypt with key1
1999 let plaintext = b"encrypted with key1";
2000 let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key1).unwrap();
2001
2002 let download_path = "/s3/wrong-key";
2003 Mock::given(method("GET"))
2004 .and(path(download_path))
2005 .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
2006 .mount(&server)
2007 .await;
2008
2009 // Client has key2 (wrong key)
2010 let client = authed_client(&server);
2011 client.set_master_key_raw(key2);
2012
2013 let result = client
2014 .blob_download("sha256-x", &format!("{}{}", server.uri(), download_path))
2015 .await;
2016
2017 assert!(
2018 result.is_err(),
2019 "Download with wrong key should fail: {result:?}"
2020 );
2021 assert!(matches!(
2022 result.unwrap_err(),
2023 SyncKitError::DecryptionFailed
2024 ));
2025 }
2026
2027 // ── Pull without auth ──
2028
2029 #[tokio::test]
2030 async fn pull_without_auth_returns_not_authenticated() {
2031 let server = MockServer::start().await;
2032 let client = client_for(&server);
2033
2034 let err = client
2035 .pull(DeviceId::new(Uuid::new_v4()), 0)
2036 .await
2037 .unwrap_err();
2038 assert!(matches!(err, SyncKitError::NotAuthenticated));
2039 }
2040
2041 // ── List devices without auth ──
2042
2043 #[tokio::test]
2044 async fn list_devices_without_auth_returns_not_authenticated() {
2045 let server = MockServer::start().await;
2046 let client = client_for(&server);
2047
2048 let err = client.list_devices().await.unwrap_err();
2049 assert!(matches!(err, SyncKitError::NotAuthenticated));
2050 }
2051
2052 // ── has_server_key without auth ──
2053
2054 #[tokio::test]
2055 async fn has_server_key_without_auth_returns_not_authenticated() {
2056 let server = MockServer::start().await;
2057 let client = client_for(&server);
2058
2059 let err = client.has_server_key().await.unwrap_err();
2060 assert!(matches!(err, SyncKitError::NotAuthenticated));
2061 }
2062
2063 // ── Encryption roundtrip through push/pull (end-to-end) ──
2064
2065 #[tokio::test]
2066 async fn end_to_end_push_pull_encryption_roundtrip() {
2067 let server = MockServer::start().await;
2068
2069 let client = authed_client(&server);
2070 let key = synckit_client::crypto::generate_master_key();
2071 client.set_master_key_raw(key);
2072
2073 let device_id = DeviceId::new(Uuid::new_v4());
2074 let original_data = json!({
2075 "title": "End-to-end test",
2076 "tags": ["e2e", "encryption"],
2077 "nested": {"key": "value"},
2078 "count": 42
2079 });
2080
2081 // Capture push request to feed back through pull
2082 Mock::given(method("POST"))
2083 .and(path("/api/v1/sync/push"))
2084 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
2085 .mount(&server)
2086 .await;
2087
2088 client
2089 .push(
2090 device_id,
2091 vec![ChangeEntry {
2092 table: "tasks".into(),
2093 op: ChangeOp::Insert,
2094 row_id: "e2e-row".into(),
2095 timestamp: Utc::now(),
2096 hlc: Hlc::zero(DeviceId::nil()),
2097 data: Some(original_data.clone()),
2098 extra: serde_json::Map::default(),
2099 }],
2100 )
2101 .await
2102 .unwrap();
2103
2104 // Extract the encrypted data that was sent to the server
2105 let requests = server.received_requests().await.unwrap();
2106 let push_body: serde_json::Value = serde_json::from_slice(
2107 &requests
2108 .iter()
2109 .find(|r| r.url.path() == "/api/v1/sync/push")
2110 .unwrap()
2111 .body,
2112 )
2113 .unwrap();
2114
2115 let wire_entry = &push_body["changes"][0];
2116
2117 // Feed encrypted data back through pull
2118 Mock::given(method("POST"))
2119 .and(path("/api/v1/sync/pull"))
2120 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2121 "changes": [{
2122 "seq": 1,
2123 "device_id": device_id,
2124 "table": wire_entry["table"],
2125 "op": wire_entry["op"],
2126 "row_id": wire_entry["row_id"],
2127 "timestamp": wire_entry["timestamp"],
2128 "data": wire_entry["data"],
2129 }],
2130 "cursor": 1,
2131 "has_more": false,
2132 })))
2133 .mount(&server)
2134 .await;
2135
2136 let (changes, _, _) = client.pull(device_id, 0).await.unwrap();
2137 assert_eq!(changes.len(), 1);
2138 assert_eq!(
2139 changes[0].data.as_ref().unwrap(),
2140 &original_data,
2141 "Data must survive push encryption + pull decryption"
2142 );
2143 }
2144
2145 // ── Retry count verification ──
2146
2147 #[tokio::test]
2148 async fn retry_exhausts_all_attempts_on_persistent_503() {
2149 let server = MockServer::start().await;
2150
2151 Mock::given(method("GET"))
2152 .and(path("/api/v1/sync/status"))
2153 .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
2154 .mount(&server)
2155 .await;
2156
2157 let client = authed_client(&server);
2158 let err = client.status().await.unwrap_err();
2159 assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
2160
2161 // Should have made exactly 4 requests (1 initial + 3 retries)
2162 let requests = server.received_requests().await.unwrap();
2163 let status_requests: Vec<_> = requests
2164 .iter()
2165 .filter(|r| r.url.path() == "/api/v1/sync/status")
2166 .collect();
2167 assert_eq!(
2168 status_requests.len(),
2169 4,
2170 "Expected 4 total requests (1 + MAX_RETRIES=3), got {}",
2171 status_requests.len()
2172 );
2173 }
2174
2175 #[tokio::test]
2176 async fn unsafe_op_makes_exactly_one_attempt_on_transient_error() {
2177 // create_subscription_checkout is Idempotency::Unsafe (a retry could mint a
2178 // second Stripe session), so a transient 503 must NOT be retried.
2179 let server = MockServer::start().await;
2180
2181 Mock::given(method("POST"))
2182 .and(path("/api/v1/sync/subscription/checkout"))
2183 .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
2184 .mount(&server)
2185 .await;
2186
2187 let client = authed_client(&server);
2188 let err = client
2189 .create_subscription_checkout(1_000_000_000, synckit_client::BillingInterval::Monthly)
2190 .await
2191 .unwrap_err();
2192 assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
2193
2194 let requests = server.received_requests().await.unwrap();
2195 let checkout_requests = requests
2196 .iter()
2197 .filter(|r| r.url.path() == "/api/v1/sync/subscription/checkout")
2198 .count();
2199 assert_eq!(
2200 checkout_requests, 1,
2201 "Unsafe op must be attempted exactly once, got {checkout_requests}"
2202 );
2203 }
2204
2205 #[tokio::test]
2206 async fn retry_not_attempted_on_404() {
2207 let server = MockServer::start().await;
2208
2209 Mock::given(method("GET"))
2210 .and(path("/api/v1/sync/status"))
2211 .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
2212 .mount(&server)
2213 .await;
2214
2215 let client = authed_client(&server);
2216 let err = client.status().await.unwrap_err();
2217 assert!(matches!(err, SyncKitError::Server { status: 404, .. }));
2218
2219 let requests = server.received_requests().await.unwrap();
2220 let count = requests
2221 .iter()
2222 .filter(|r| r.url.path() == "/api/v1/sync/status")
2223 .count();
2224 assert_eq!(count, 1, "404 should not be retried");
2225 }
2226
2227 #[tokio::test]
2228 async fn retry_succeeds_on_third_attempt() {
2229 let server = MockServer::start().await;
2230
2231 Mock::given(method("GET"))
2232 .and(path("/api/v1/sync/status"))
2233 .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
2234 .up_to_n_times(2)
2235 .mount(&server)
2236 .await;
2237
2238 Mock::given(method("GET"))
2239 .and(path("/api/v1/sync/status"))
2240 .respond_with(
2241 ResponseTemplate::new(200)
2242 .set_body_json(json!({"total_changes": 7, "latest_cursor": 3})),
2243 )
2244 .mount(&server)
2245 .await;
2246
2247 let client = authed_client(&server);
2248 let status = client.status().await.unwrap();
2249 assert_eq!(status.total_changes, 7);
2250
2251 let requests = server.received_requests().await.unwrap();
2252 let count = requests
2253 .iter()
2254 .filter(|r| r.url.path() == "/api/v1/sync/status")
2255 .count();
2256 assert_eq!(count, 3, "Should succeed on 3rd attempt");
2257 }
2258
2259 // ── Malformed / unexpected responses ──
2260
2261 #[tokio::test]
2262 async fn authenticate_html_response_returns_error() {
2263 let server = MockServer::start().await;
2264
2265 Mock::given(method("POST"))
2266 .and(path("/api/v1/sync/auth"))
2267 .respond_with(
2268 ResponseTemplate::new(200)
2269 .insert_header("content-type", "text/html")
2270 .set_body_string("<html><body>Not JSON</body></html>"),
2271 )
2272 .mount(&server)
2273 .await;
2274
2275 let client = client_for(&server);
2276 let err = client
2277 .authenticate("user@test.com", "pass", "test-key")
2278 .await
2279 .unwrap_err();
2280 // reqwest .json() fails when body isn't valid JSON
2281 assert!(
2282 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2283 "HTML response should produce Http or Json error, got: {err:?}"
2284 );
2285 }
2286
2287 #[tokio::test]
2288 async fn push_empty_response_body_returns_error() {
2289 let server = MockServer::start().await;
2290
2291 Mock::given(method("POST"))
2292 .and(path("/api/v1/sync/push"))
2293 .respond_with(ResponseTemplate::new(200).set_body_string(""))
2294 .mount(&server)
2295 .await;
2296
2297 let client = authed_client(&server);
2298 let key = synckit_client::crypto::generate_master_key();
2299 client.set_master_key_raw(key);
2300
2301 let err = client
2302 .push(DeviceId::new(Uuid::new_v4()), vec![])
2303 .await
2304 .unwrap_err();
2305 assert!(
2306 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2307 "Empty body should produce parse error, got: {err:?}"
2308 );
2309 }
2310
2311 #[tokio::test]
2312 async fn pull_response_missing_has_more_returns_error() {
2313 let server = MockServer::start().await;
2314
2315 Mock::given(method("POST"))
2316 .and(path("/api/v1/sync/pull"))
2317 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2318 "changes": [],
2319 "cursor": 0
2320 // missing "has_more"
2321 })))
2322 .mount(&server)
2323 .await;
2324
2325 let client = authed_client(&server);
2326 let key = synckit_client::crypto::generate_master_key();
2327 client.set_master_key_raw(key);
2328
2329 let err = client
2330 .pull(DeviceId::new(Uuid::new_v4()), 0)
2331 .await
2332 .unwrap_err();
2333 assert!(
2334 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2335 "Missing has_more should produce parse error, got: {err:?}"
2336 );
2337 }
2338
2339 #[tokio::test]
2340 async fn status_response_cursor_wrong_type_returns_error() {
2341 let server = MockServer::start().await;
2342
2343 Mock::given(method("GET"))
2344 .and(path("/api/v1/sync/status"))
2345 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2346 "total_changes": 10,
2347 "latest_cursor": "not-a-number"
2348 })))
2349 .mount(&server)
2350 .await;
2351
2352 let client = authed_client(&server);
2353 let err = client.status().await.unwrap_err();
2354 assert!(
2355 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2356 "Wrong type for cursor should produce parse error, got: {err:?}"
2357 );
2358 }
2359
2360 #[tokio::test]
2361 async fn authenticate_response_missing_app_id_returns_error() {
2362 let server = MockServer::start().await;
2363
2364 Mock::given(method("POST"))
2365 .and(path("/api/v1/sync/auth"))
2366 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2367 "token": fresh_token(),
2368 "user_id": "550e8400-e29b-41d4-a716-446655440000"
2369 // missing "app_id"
2370 })))
2371 .mount(&server)
2372 .await;
2373
2374 let client = client_for(&server);
2375 let err = client
2376 .authenticate("user@test.com", "pass", "test-key")
2377 .await
2378 .unwrap_err();
2379 assert!(
2380 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2381 "Missing app_id should produce parse error, got: {err:?}"
2382 );
2383 }
2384
2385 #[tokio::test]
2386 async fn register_device_extra_fields_ignored() {
2387 let server = MockServer::start().await;
2388
2389 let (user_id, app_id) = test_ids();
2390 let response_with_extras = json!({
2391 "id": Uuid::new_v4(),
2392 "app_id": app_id,
2393 "user_id": user_id,
2394 "device_name": "Test Device",
2395 "platform": "test",
2396 "last_seen_at": "2025-01-01T00:00:00Z",
2397 "created_at": "2025-01-01T00:00:00Z",
2398 "extra_field": "should be ignored",
2399 "unknown_number": 42
2400 });
2401
2402 Mock::given(method("POST"))
2403 .and(path("/api/v1/sync/devices"))
2404 .respond_with(ResponseTemplate::new(200).set_body_json(response_with_extras))
2405 .mount(&server)
2406 .await;
2407
2408 let client = authed_client(&server);
2409 let device = client.register_device("Test", "test").await.unwrap();
2410 assert_eq!(device.device_name, "Test Device");
2411 }
2412
2413 #[tokio::test]
2414 async fn blob_upload_url_response_missing_already_exists_returns_error() {
2415 let server = MockServer::start().await;
2416
2417 Mock::given(method("POST"))
2418 .and(path("/api/v1/sync/blobs/upload"))
2419 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2420 "upload_url": "https://s3.example.com/put"
2421 // missing "already_exists"
2422 })))
2423 .mount(&server)
2424 .await;
2425
2426 let client = authed_client(&server);
2427 let result = client.blob_upload_url("hash", 100).await;
2428 match result {
2429 Err(err) => assert!(
2430 matches!(err, SyncKitError::Http(_) | SyncKitError::Json(_)),
2431 "Missing already_exists should produce error, got: {err:?}"
2432 ),
2433 Ok(_) => panic!("Expected error for missing already_exists field"),
2434 }
2435 }
2436
2437 #[tokio::test]
2438 async fn server_returns_413_request_entity_too_large() {
2439 let server = MockServer::start().await;
2440
2441 Mock::given(method("POST"))
2442 .and(path("/api/v1/sync/push"))
2443 .respond_with(ResponseTemplate::new(413).set_body_string("Request entity too large"))
2444 .mount(&server)
2445 .await;
2446
2447 let client = authed_client(&server);
2448 let key = synckit_client::crypto::generate_master_key();
2449 client.set_master_key_raw(key);
2450
2451 let err = client
2452 .push(DeviceId::new(Uuid::new_v4()), vec![])
2453 .await
2454 .unwrap_err();
2455 match err {
2456 SyncKitError::Server {
2457 status, message, ..
2458 } => {
2459 assert_eq!(status, 413);
2460 assert!(message.contains("too large"));
2461 }
2462 other => panic!("Expected Server error, got: {other:?}"),
2463 }
2464 }
2465
2466 // ── Session edge cases ──
2467
2468 #[tokio::test]
2469 async fn double_authenticate_overwrites_session() {
2470 let server = MockServer::start().await;
2471
2472 let (user_id, app_id) = test_ids();
2473 let second_user_id = UserId::new(Uuid::new_v4());
2474
2475 // First auth response
2476 Mock::given(method("POST"))
2477 .and(path("/api/v1/sync/auth"))
2478 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2479 "token": fresh_token(),
2480 "user_id": user_id,
2481 "app_id": app_id,
2482 })))
2483 .up_to_n_times(1)
2484 .mount(&server)
2485 .await;
2486
2487 // Second auth response with different user_id
2488 Mock::given(method("POST"))
2489 .and(path("/api/v1/sync/auth"))
2490 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2491 "token": fresh_token(),
2492 "user_id": second_user_id,
2493 "app_id": app_id,
2494 })))
2495 .mount(&server)
2496 .await;
2497
2498 let client = client_for(&server);
2499 let (uid1, _) = client
2500 .authenticate("user1@test.com", "pass1", "test-key")
2501 .await
2502 .unwrap();
2503 assert_eq!(uid1, user_id);
2504
2505 let (uid2, _) = client
2506 .authenticate("user2@test.com", "pass2", "test-key")
2507 .await
2508 .unwrap();
2509 assert_eq!(uid2, second_user_id);
2510
2511 // Session should now reflect the second auth
2512 let info = client.session_info().unwrap();
2513 assert_eq!(info.user_id, second_user_id);
2514 }
2515
2516 #[tokio::test]
2517 async fn clear_session_then_authenticate_succeeds() {
2518 let server = MockServer::start().await;
2519
2520 Mock::given(method("POST"))
2521 .and(path("/api/v1/sync/auth"))
2522 .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
2523 .mount(&server)
2524 .await;
2525
2526 let client = authed_client(&server);
2527 assert!(client.session_info().is_some());
2528
2529 client.clear_session();
2530 assert!(client.session_info().is_none());
2531
2532 // Re-authenticate should work
2533 let result = client
2534 .authenticate("user@test.com", "pass", "test-key")
2535 .await;
2536 assert!(
2537 result.is_ok(),
2538 "Should be able to re-authenticate after clear: {result:?}"
2539 );
2540 assert!(client.session_info().is_some());
2541 }
2542
2543 #[tokio::test]
2544 async fn restore_session_with_expired_token_then_push_returns_token_expired() {
2545 let server = MockServer::start().await;
2546 let client = client_for(&server);
2547 let (user_id, app_id) = test_ids();
2548
2549 let expired = fake_jwt(Utc::now().timestamp() - 3600);
2550 client.restore_session(&expired, user_id, app_id);
2551 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
2552
2553 let err = client
2554 .push(DeviceId::new(Uuid::new_v4()), vec![])
2555 .await
2556 .unwrap_err();
2557 assert!(
2558 matches!(err, SyncKitError::TokenExpired),
2559 "Restored expired token should return TokenExpired, got: {err:?}"
2560 );
2561 }
2562
2563 // ── Encryption state edge cases ──
2564
2565 #[tokio::test]
2566 async fn setup_encryption_new_twice_overwrites() {
2567 let server = MockServer::start().await;
2568
2569 Mock::given(method("PUT"))
2570 .and(path("/api/v1/sync/keys"))
2571 .respond_with(ResponseTemplate::new(200))
2572 .mount(&server)
2573 .await;
2574
2575 let client = authed_client(&server);
2576
2577 client.setup_encryption_new("pass1").await.unwrap();
2578 assert!(client.has_master_key());
2579
2580 // Second call overwrites
2581 client.setup_encryption_new("pass2").await.unwrap();
2582 assert!(client.has_master_key());
2583
2584 // Verify the second PUT used a different envelope
2585 let requests = server.received_requests().await.unwrap();
2586 let puts: Vec<_> = requests
2587 .iter()
2588 .filter(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
2589 .collect();
2590 assert_eq!(puts.len(), 2);
2591 // Different envelopes (different random keys)
2592 assert_ne!(puts[0].body, puts[1].body);
2593 }
2594
2595 // ── Blob edge cases ──
2596
2597 #[tokio::test]
2598 async fn blob_confirm_retries_on_503() {
2599 let server = MockServer::start().await;
2600
2601 Mock::given(method("POST"))
2602 .and(path("/api/v1/sync/blobs/confirm"))
2603 .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
2604 .up_to_n_times(1)
2605 .mount(&server)
2606 .await;
2607
2608 Mock::given(method("POST"))
2609 .and(path("/api/v1/sync/blobs/confirm"))
2610 .respond_with(ResponseTemplate::new(200))
2611 .mount(&server)
2612 .await;
2613
2614 let client = authed_client(&server);
2615 let result = client.blob_confirm("sha256-retry", 512).await;
2616 assert!(result.is_ok(), "Should succeed after retry: {result:?}");
2617 }
2618
2619 #[tokio::test]
2620 async fn blob_download_retries_on_503() {
2621 let server = MockServer::start().await;
2622
2623 let client = authed_client(&server);
2624 let key = synckit_client::crypto::generate_master_key();
2625 client.set_master_key_raw(key);
2626
2627 let plaintext = b"retry download test";
2628 let hash = hex::encode(sha2::Sha256::digest(plaintext));
2629 let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();
2630
2631 let download_path = "/s3/retry-download";
2632 Mock::given(method("GET"))
2633 .and(path(download_path))
2634 .respond_with(ResponseTemplate::new(503))
2635 .up_to_n_times(1)
2636 .mount(&server)
2637 .await;
2638
2639 Mock::given(method("GET"))
2640 .and(path(download_path))
2641 .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
2642 .mount(&server)
2643 .await;
2644
2645 let presigned = format!("{}{}", server.uri(), download_path);
2646 let result = client.blob_download(&hash, &presigned).await.unwrap();
2647 assert_eq!(result, plaintext);
2648 }
2649
2650 #[tokio::test]
2651 async fn blob_upload_1mb_with_correct_overhead() {
2652 let server = MockServer::start().await;
2653
2654 let upload_path = "/s3/1mb-upload";
2655 Mock::given(method("PUT"))
2656 .and(path(upload_path))
2657 .respond_with(ResponseTemplate::new(200))
2658 .mount(&server)
2659 .await;
2660
2661 let client = authed_client(&server);
2662 let key = synckit_client::crypto::generate_master_key();
2663 client.set_master_key_raw(key);
2664
2665 let plaintext: Vec<u8> = (0..1_048_576u32).map(|i| (i % 256) as u8).collect();
2666 let presigned = format!("{}{}", server.uri(), upload_path);
2667 client
2668 .blob_upload("sha256-1mb", &presigned, plaintext.clone())
2669 .await
2670 .unwrap();
2671
2672 let requests = server.received_requests().await.unwrap();
2673 let upload_req = requests
2674 .iter()
2675 .find(|r| r.url.path() == upload_path)
2676 .unwrap();
2677 assert_eq!(
2678 upload_req.body.len(),
2679 plaintext.len() + synckit_client::crypto::chunked_blob_overhead(plaintext.len()),
2680 "1MB upload should add exactly the v3 chunked overhead"
2681 );
2682 }
2683
2684 // ── Device management edge cases ──
2685
2686 #[tokio::test]
2687 async fn register_device_with_empty_name() {
2688 let server = MockServer::start().await;
2689
2690 Mock::given(method("POST"))
2691 .and(path("/api/v1/sync/devices"))
2692 .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
2693 .mount(&server)
2694 .await;
2695
2696 let client = authed_client(&server);
2697 // Empty name should not panic; server may accept or reject
2698 let result = client.register_device("", "macos").await;
2699 assert!(result.is_ok(), "Empty device name should not panic");
2700 }
2701
2702 #[tokio::test]
2703 async fn register_device_with_unicode_name() {
2704 let server = MockServer::start().await;
2705
2706 let (user_id, app_id) = test_ids();
2707 Mock::given(method("POST"))
2708 .and(path("/api/v1/sync/devices"))
2709 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
2710 "id": Uuid::new_v4(),
2711 "app_id": app_id,
2712 "user_id": user_id,
2713 "device_name": "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}",
2714 "platform": "macos",
2715 "last_seen_at": "2025-01-01T00:00:00Z",
2716 "created_at": "2025-01-01T00:00:00Z",
2717 })))
2718 .mount(&server)
2719 .await;
2720
2721 let client = authed_client(&server);
2722 let device = client
2723 .register_device("\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}", "macos")
2724 .await
2725 .unwrap();
2726 assert_eq!(
2727 device.device_name,
2728 "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}"
2729 );
2730 }
2731
2732 #[tokio::test]
2733 async fn list_devices_empty_array() {
2734 let server = MockServer::start().await;
2735
2736 Mock::given(method("GET"))
2737 .and(path("/api/v1/sync/devices"))
2738 .respond_with(ResponseTemplate::new(200).set_body_json(json!([])))
2739 .mount(&server)
2740 .await;
2741
2742 let client = authed_client(&server);
2743 let devices = client.list_devices().await.unwrap();
2744 assert!(devices.is_empty());
2745 }
2746
2747 // ── Concurrency stress tests ──
2748
2749 #[tokio::test]
2750 async fn concurrent_session_info_reads() {
2751 let server = MockServer::start().await;
2752 let client = Arc::new(authed_client(&server));
2753
2754 let mut handles = Vec::new();
2755 for _ in 0..50 {
2756 let c = Arc::clone(&client);
2757 handles.push(tokio::spawn(async move { c.session_info() }));
2758 }
2759
2760 for h in handles {
2761 let info = h.await.unwrap();
2762 assert!(
2763 info.is_some(),
2764 "All concurrent reads should see the session"
2765 );
2766 }
2767 }
2768
2769 #[tokio::test]
2770 async fn concurrent_has_master_key_reads() {
2771 let server = MockServer::start().await;
2772 let client = Arc::new(authed_client(&server));
2773 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
2774
2775 let mut handles = Vec::new();
2776 for _ in 0..50 {
2777 let c = Arc::clone(&client);
2778 handles.push(tokio::spawn(async move { c.has_master_key() }));
2779 }
2780
2781 for h in handles {
2782 let has_key = h.await.unwrap();
2783 assert!(has_key, "All concurrent reads should see the master key");
2784 }
2785 }
2786
2787 #[tokio::test]
2788 async fn concurrent_status_checks() {
2789 let server = MockServer::start().await;
2790
2791 Mock::given(method("GET"))
2792 .and(path("/api/v1/sync/status"))
2793 .respond_with(
2794 ResponseTemplate::new(200)
2795 .set_body_json(json!({"total_changes": 5, "latest_cursor": 3})),
2796 )
2797 .mount(&server)
2798 .await;
2799
2800 let client = Arc::new(authed_client(&server));
2801
2802 let mut handles = Vec::new();
2803 for _ in 0..20 {
2804 let c = Arc::clone(&client);
2805 handles.push(tokio::spawn(async move { c.status().await }));
2806 }
2807
2808 for h in handles {
2809 let result = h.await.unwrap();
2810 assert!(
2811 result.is_ok(),
2812 "All concurrent status checks should succeed: {result:?}"
2813 );
2814 assert_eq!(result.unwrap().total_changes, 5);
2815 }
2816 }
2817
2818 #[tokio::test]
2819 async fn concurrent_push_100_entries_each() {
2820 let server = MockServer::start().await;
2821
2822 Mock::given(method("POST"))
2823 .and(path("/api/v1/sync/push"))
2824 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
2825 .mount(&server)
2826 .await;
2827
2828 let client = Arc::new(authed_client(&server));
2829 let key = synckit_client::crypto::generate_master_key();
2830 client.set_master_key_raw(key);
2831
2832 let mut handles = Vec::new();
2833 for batch in 0..4 {
2834 let c = Arc::clone(&client);
2835 handles.push(tokio::spawn(async move {
2836 let changes: Vec<ChangeEntry> = (0..100)
2837 .map(|i| ChangeEntry {
2838 table: format!("batch_{batch}"),
2839 op: ChangeOp::Insert,
2840 row_id: format!("row_{i}"),
2841 timestamp: Utc::now(),
2842 hlc: Hlc::zero(DeviceId::nil()),
2843 data: Some(json!({"index": i})),
2844 extra: serde_json::Map::default(),
2845 })
2846 .collect();
2847 c.push(DeviceId::new(Uuid::new_v4()), changes).await
2848 }));
2849 }
2850
2851 for h in handles {
2852 let result = h.await.unwrap();
2853 assert!(
2854 result.is_ok(),
2855 "Concurrent 100-entry push should succeed: {result:?}"
2856 );
2857 }
2858 }
2859
2860 // ── Timeout tests ──
2861
2862 fn short_timeout_client(server: &MockServer) -> SyncKitClient {
2863 ensure_crypto_provider();
2864 let http = reqwest::Client::builder()
2865 .timeout(Duration::from_millis(100))
2866 .connect_timeout(Duration::from_millis(100))
2867 .build()
2868 .unwrap();
2869 SyncKitClient::with_http_client(
2870 SyncKitConfig {
2871 server_url: server.uri(),
2872 api_key: "test-api-key".to_string(),
2873 },
2874 http,
2875 )
2876 }
2877
2878 fn authed_short_timeout_client(server: &MockServer) -> SyncKitClient {
2879 let client = short_timeout_client(server);
2880 let (user_id, app_id) = test_ids();
2881 client.restore_session(&fresh_token(), user_id, app_id);
2882 client
2883 }
2884
2885 #[tokio::test]
2886 async fn status_times_out_on_slow_server() {
2887 let server = MockServer::start().await;
2888
2889 Mock::given(method("GET"))
2890 .and(path("/api/v1/sync/status"))
2891 .respond_with(
2892 ResponseTemplate::new(200)
2893 .set_body_json(json!({"total_changes": 0, "latest_cursor": null}))
2894 .set_delay(Duration::from_secs(5)),
2895 )
2896 .mount(&server)
2897 .await;
2898
2899 let client = authed_short_timeout_client(&server);
2900 let err = client.status().await.unwrap_err();
2901 // Timeout triggers Http error, which is transient, so it retries and eventually exhausts
2902 assert!(
2903 matches!(err, SyncKitError::Http(_)),
2904 "Slow server should produce Http (timeout) error, got: {err:?}"
2905 );
2906 }
2907
2908 #[tokio::test]
2909 async fn push_retries_on_timeout_then_succeeds() {
2910 let server = MockServer::start().await;
2911
2912 // First request: slow (will timeout)
2913 Mock::given(method("POST"))
2914 .and(path("/api/v1/sync/push"))
2915 .respond_with(
2916 ResponseTemplate::new(200)
2917 .set_body_json(json!({"cursor": 1}))
2918 .set_delay(Duration::from_secs(5)),
2919 )
2920 .up_to_n_times(1)
2921 .mount(&server)
2922 .await;
2923
2924 // Second request: fast (succeeds)
2925 Mock::given(method("POST"))
2926 .and(path("/api/v1/sync/push"))
2927 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 42})))
2928 .mount(&server)
2929 .await;
2930
2931 let client = authed_short_timeout_client(&server);
2932 let key = synckit_client::crypto::generate_master_key();
2933 client.set_master_key_raw(key);
2934
2935 let cursor = client
2936 .push(DeviceId::new(Uuid::new_v4()), vec![])
2937 .await
2938 .unwrap();
2939 assert_eq!(cursor, 42);
2940 }
2941
2942 // ── SSE subscribe / reconnect state machine ──
2943 //
2944 // These drive the real `subscribe()` -> `SyncNotifyStream::next_change()` path
2945 // against wiremock so the reconnect state machine (subscribe.rs) is exercised
2946 // end to end: a live notification, a transparent reconnect across a stream
2947 // drop, the fatal auth-rejection exit, and the give-up-after-N-failures cap.
2948
2949 const SUBSCRIBE_PATH: &str = "/api/v1/sync/subscribe";
2950
2951 /// One complete SSE "changed" block, body-terminated so the client sees a full
2952 /// event and then end-of-stream.
2953 fn sse_changed_block() -> &'static str {
2954 "event: changed\n\n"
2955 }
2956
2957 #[tokio::test]
2958 async fn subscribe_yields_changed_event() {
2959 let server = MockServer::start().await;
2960 Mock::given(method("GET"))
2961 .and(path(SUBSCRIBE_PATH))
2962 .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
2963 .mount(&server)
2964 .await;
2965
2966 let client = authed_client(&server);
2967 let mut stream = client.subscribe().await.expect("subscribe should succeed");
2968
2969 // The first block is delivered inside the initial response body, so this
2970 // returns without any reconnect.
2971 assert_eq!(stream.next_change().await, Some(()));
2972 }
2973
2974 #[tokio::test]
2975 async fn subscribe_reconnects_transparently_after_stream_drop() {
2976 let server = MockServer::start().await;
2977 // Every connection serves one block then ends (Content-Length terminates the
2978 // body). Consuming the first event, then reading past it, drops the stream
2979 // and forces a reconnect that must transparently yield the next event.
2980 Mock::given(method("GET"))
2981 .and(path(SUBSCRIBE_PATH))
2982 .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
2983 .mount(&server)
2984 .await;
2985
2986 let client = authed_client(&server);
2987 let mut stream = client.subscribe().await.expect("subscribe should succeed");
2988
2989 // #1 comes from the initial connection; #2 can only arrive after the stream
2990 // drops (body EOF) and reconnect() re-establishes it.
2991 assert_eq!(stream.next_change().await, Some(()));
2992 assert_eq!(stream.next_change().await, Some(()));
2993
2994 // subscribe() opened one connection; the second event required at least one
2995 // reconnect, so the server saw two or more subscribe requests.
2996 let hits = server
2997 .received_requests()
2998 .await
2999 .unwrap()
3000 .into_iter()
3001 .filter(|r| r.url.path() == SUBSCRIBE_PATH)
3002 .count();
3003 assert!(
3004 hits >= 2,
3005 "expected a reconnect (>=2 subscribe requests), got {hits}"
3006 );
3007 }
3008
3009 #[tokio::test]
3010 async fn subscribe_stream_closes_on_auth_rejection() {
3011 let server = MockServer::start().await;
3012 // First connection opens cleanly but carries no event and ends immediately,
3013 // forcing a reconnect. The reconnect is rejected for auth -> fatal, so the
3014 // stream ends with `None` rather than retrying.
3015 Mock::given(method("GET"))
3016 .and(path(SUBSCRIBE_PATH))
3017 .respond_with(ResponseTemplate::new(200).set_body_string(""))
3018 .up_to_n_times(1)
3019 .mount(&server)
3020 .await;
3021 Mock::given(method("GET"))
3022 .and(path(SUBSCRIBE_PATH))
3023 .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
3024 .mount(&server)
3025 .await;
3026
3027 let client = authed_client(&server);
3028 let mut stream = client.subscribe().await.expect("initial subscribe is 200");
3029
3030 // Empty body -> EOF -> reconnect -> 401 -> fatal -> None.
3031 assert_eq!(stream.next_change().await, None);
3032 }
3033
3034 // Note: the "give up after MAX_RECONNECT_ATTEMPTS consecutive failures" path is
3035 // deliberately not covered end-to-end here. Exercising it against a live mock
3036 // server would incur the real exponential backoff (growing to a 60s cap, minutes
3037 // of wall-clock), and `tokio::time::pause()` cannot collapse it: the wiremock
3038 // server runs on its own runtime, so a paused clock auto-advances past the
3039 // cross-thread request to the nearest timer and the reconnect never completes.
3040 // The two pieces of that path are covered separately: the backoff schedule by
3041 // `reconnect_delay_grows_then_caps` (subscribe.rs), and the fatal-exit mechanics
3042 // (returning `None` and clearing state) by the auth-rejection test above.
3043
3044 // ── End-to-end key-rotation orchestration ──
3045 //
3046 // These drive `rotate_key()` through the full server protocol against wiremock:
3047 // fetch key -> begin -> re-encrypt loop -> complete, plus the straggler retry on
3048 // a 409. They are gated `#[cfg(not(feature = "keychain"))]` because `rotate_key`
3049 // finishes by caching the new key with `keystore::store_key`, which hits the OS
3050 // secret-service under the `keychain` feature and is unavailable on a headless
3051 // test host. With keychain off, `store_key` is the no-op stub, so the test
3052 // exercises the orchestration hermetically. Run with:
3053 // cargo test --no-default-features --features store,testing
3054 #[cfg(not(feature = "keychain"))]
3055 mod rotation_e2e {
3056 use super::*;
3057
3058 const KEYS_PATH: &str = "/api/v1/sync/keys";
3059 const ROTATE_PATH: &str = "/api/v1/sync/keys/rotate";
3060 const ENTRIES_PATH: &str = "/api/v1/sync/keys/rotate/entries";
3061 const BATCH_PATH: &str = "/api/v1/sync/keys/rotate/batch";
3062 const COMPLETE_PATH: &str = "/api/v1/sync/keys/rotate/complete";
3063
3064 const ROTATE_PW: &str = "rotate-password";
3065
3066 /// A `GET /keys` body wrapping `old_key` under [`ROTATE_PW`], with no rotation
3067 /// in progress, so `rotate_key` verifies the password and mints a fresh key.
3068 fn get_keys_body(old_key: &[u8; 32]) -> serde_json::Value {
3069 let envelope = synckit_client::crypto::wrap_master_key(old_key, ROTATE_PW).unwrap();
3070 json!({ "encrypted_key": envelope, "key_version": 1, "key_id": 1 })
3071 }
3072
3073 /// One rotation entry: `plaintext` sealed under `old_key` with the same
3074 /// `(table, row_id)` AAD the client rebinds during re-encryption.
3075 fn rotation_entry(old_key: &[u8; 32], table: &str, row_id: &str) -> serde_json::Value {
3076 let ctx = synckit_client::crypto::AeadContext::entry(table, row_id);
3077 let sealed =
3078 synckit_client::crypto::encrypt_json_aad(&json!({"title": "secret"}), old_key, &ctx)
3079 .unwrap();
3080 json!({ "seq": 1, "table": table, "row_id": row_id, "data": sealed })
3081 }
3082
3083 fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
3084 reqs.iter().filter(|r| r.url.path() == p).count()
3085 }
3086
3087 #[tokio::test]
3088 async fn rotate_key_drives_full_orchestration() {
3089 let server = MockServer::start().await;
3090 let old_key = synckit_client::crypto::generate_master_key();
3091
3092 Mock::given(method("GET"))
3093 .and(path(KEYS_PATH))
3094 .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
3095 .mount(&server)
3096 .await;
3097 Mock::given(method("POST"))
3098 .and(path(ROTATE_PATH))
3099 .respond_with(ResponseTemplate::new(200).set_body_json(
3100 json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
3101 ))
3102 .mount(&server)
3103 .await;
3104 // One batch of work, then drained (has_more = false ends the re-encrypt loop).
3105 Mock::given(method("POST"))
3106 .and(path(ENTRIES_PATH))
3107 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3108 "entries": [rotation_entry(&old_key, "tasks", "r1")],
3109 "has_more": false
3110 })))
3111 .mount(&server)
3112 .await;
3113 Mock::given(method("POST"))
3114 .and(path(BATCH_PATH))
3115 .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
3116 .mount(&server)
3117 .await;
3118 Mock::given(method("POST"))
3119 .and(path(COMPLETE_PATH))
3120 .respond_with(ResponseTemplate::new(200))
3121 .mount(&server)
3122 .await;
3123
3124 let client = authed_client(&server);
3125 client
3126 .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
3127 .await
3128 .expect("full rotation should complete");
3129
3130 // Every stage of the protocol was driven, in the right shape.
3131 let reqs = server.received_requests().await.unwrap();
3132 assert_eq!(hits(&reqs, KEYS_PATH), 1, "fetched key state once");
3133 assert_eq!(hits(&reqs, ROTATE_PATH), 1, "began rotation once");
3134 assert!(
3135 hits(&reqs, ENTRIES_PATH) >= 1,
3136 "pulled entries to re-encrypt"
3137 );
3138 assert_eq!(hits(&reqs, BATCH_PATH), 1, "pushed one re-encrypted batch");
3139 assert_eq!(
3140 hits(&reqs, COMPLETE_PATH),
3141 1,
3142 "completed once (no stragglers)"
3143 );
3144 }
3145
3146 #[tokio::test]
3147 async fn rotate_key_retries_reencrypt_on_straggler_conflict() {
3148 let server = MockServer::start().await;
3149 let old_key = synckit_client::crypto::generate_master_key();
3150
3151 Mock::given(method("GET"))
3152 .and(path(KEYS_PATH))
3153 .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
3154 .mount(&server)
3155 .await;
3156 Mock::given(method("POST"))
3157 .and(path(ROTATE_PATH))
3158 .respond_with(ResponseTemplate::new(200).set_body_json(
3159 json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
3160 ))
3161 .mount(&server)
3162 .await;
3163 // First entries pull returns work; every later pull is drained. Mounted in
3164 // this order so the up_to_n_times(1) mock wins the first call, then the
3165 // empty-set fallback serves the straggler round's re-pull.
3166 Mock::given(method("POST"))
3167 .and(path(ENTRIES_PATH))
3168 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3169 "entries": [rotation_entry(&old_key, "tasks", "r1")],
3170 "has_more": false
3171 })))
3172 .up_to_n_times(1)
3173 .mount(&server)
3174 .await;
3175 Mock::given(method("POST"))
3176 .and(path(ENTRIES_PATH))
3177 .respond_with(
3178 ResponseTemplate::new(200)
3179 .set_body_json(json!({ "entries": [], "has_more": false })),
3180 )
3181 .mount(&server)
3182 .await;
3183 Mock::given(method("POST"))
3184 .and(path(BATCH_PATH))
3185 .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
3186 .mount(&server)
3187 .await;
3188 // First completion reports a straggler (409); the retry then succeeds.
3189 Mock::given(method("POST"))
3190 .and(path(COMPLETE_PATH))
3191 .respond_with(
3192 ResponseTemplate::new(409).set_body_json(json!({ "message": "stragglers" })),
3193 )
3194 .up_to_n_times(1)
3195 .mount(&server)
3196 .await;
3197 Mock::given(method("POST"))
3198 .and(path(COMPLETE_PATH))
3199 .respond_with(ResponseTemplate::new(200))
3200 .mount(&server)
3201 .await;
3202
3203 let client = authed_client(&server);
3204 client
3205 .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
3206 .await
3207 .expect("rotation should converge after the straggler retry");
3208
3209 // The 409 forced a second completion attempt, and the straggler round
3210 // re-ran the re-encrypt loop (a second entries pull).
3211 let reqs = server.received_requests().await.unwrap();
3212 assert_eq!(
3213 hits(&reqs, COMPLETE_PATH),
3214 2,
3215 "completed twice: 409 then 200"
3216 );
3217 assert!(
3218 hits(&reqs, ENTRIES_PATH) >= 2,
3219 "straggler round re-pulled entries"
3220 );
3221 }
3222 }
3223
3224 // ── Multipart blob upload (streaming) ──
3225 //
3226 // The transport for blobs above the server's one-shot PUT ceiling. What matters
3227 // here is that the client never buffers the whole ciphertext yet still produces
3228 // a byte-exact v3 blob: it seals 1 MiB chunks as it reads the file and cuts the
3229 // sealed stream at the part boundaries the server signed, which it could only
3230 // pick because `blob_encrypted_len` predicts the ciphertext size up front.
3231 mod blob_multipart {
3232 use super::*;
3233 use std::path::PathBuf;
3234
3235 const START_PATH: &str = "/api/v1/sync/blobs/multipart/start";
3236 const PARTS_PATH: &str = "/api/v1/sync/blobs/multipart/parts";
3237 const COMPLETE_PATH: &str = "/api/v1/sync/blobs/multipart/complete";
3238 const ABORT_PATH: &str = "/api/v1/sync/blobs/multipart/abort";
3239 const PART_PUT_PATH: &str = "/s3/part";
3240
3241 fn temp_blob(name: &str, contents: &[u8]) -> PathBuf {
3242 use std::sync::atomic::{AtomicU64, Ordering};
3243 static N: AtomicU64 = AtomicU64::new(0);
3244 let mut p = std::env::temp_dir();
3245 p.push(format!(
3246 "synckit_mp_{}_{}_{name}",
3247 std::process::id(),
3248 N.fetch_add(1, Ordering::Relaxed)
3249 ));
3250 std::fs::write(&p, contents).unwrap();
3251 p
3252 }
3253
3254 /// Stands in for the server's part-URL minting: answers whatever window the
3255 /// client asked for, rather than a fixed list, since the client requests one
3256 /// part at a time (it can only checksum a part it has already sealed).
3257 struct PartsResponder {
3258 cipher_len: usize,
3259 part_size: usize,
3260 base: String,
3261 }
3262
3263 impl wiremock::Respond for PartsResponder {
3264 fn respond(&self, req: &wiremock::Request) -> ResponseTemplate {
3265 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
3266 let first = body["first_part"].as_u64().unwrap() as usize;
3267 let count = body["count"].as_u64().unwrap() as usize;
3268 let part_count = self.cipher_len.div_ceil(self.part_size);
3269 let last = (first + count - 1).min(part_count);
3270
3271 let parts: Vec<serde_json::Value> = (first..=last)
3272 .map(|n| {
3273 let content_length = if n == part_count {
3274 self.cipher_len - self.part_size * (part_count - 1)
3275 } else {
3276 self.part_size
3277 };
3278 json!({
3279 "part_number": n,
3280 "content_length": content_length,
3281 "url": format!("{}{PART_PUT_PATH}?partNumber={n}", self.base),
3282 })
3283 })
3284 .collect();
3285 ResponseTemplate::new(200).set_body_json(json!({ "parts": parts }))
3286 }
3287 }
3288
3289 /// Mount the whole session: start (with the given plan), part-URL minting,
3290 /// the PUT target, and complete.
3291 async fn mount_session(server: &MockServer, cipher_len: usize, part_size: usize) -> u32 {
3292 let part_count = cipher_len.div_ceil(part_size) as u32;
3293
3294 Mock::given(method("POST"))
3295 .and(path(START_PATH))
3296 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3297 "upload_id": "test-upload-id",
3298 "part_size": part_size,
3299 "part_count": part_count,
3300 "already_exists": false,
3301 })))
3302 .mount(server)
3303 .await;
3304 Mock::given(method("POST"))
3305 .and(path(PARTS_PATH))
3306 .respond_with(PartsResponder {
3307 cipher_len,
3308 part_size,
3309 base: server.uri(),
3310 })
3311 .mount(server)
3312 .await;
3313 Mock::given(method("PUT"))
3314 .and(path(PART_PUT_PATH))
3315 .respond_with(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
3316 .mount(server)
3317 .await;
3318 Mock::given(method("POST"))
3319 .and(path(COMPLETE_PATH))
3320 .respond_with(ResponseTemplate::new(204))
3321 .mount(server)
3322 .await;
3323
3324 part_count
3325 }
3326
3327 fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
3328 reqs.iter().filter(|r| r.url.path() == p).count()
3329 }
3330
3331 #[tokio::test]
3332 async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
3333 let server = MockServer::start().await;
3334 let client = authed_client(&server);
3335 let key = synckit_client::crypto::generate_master_key();
3336 client.set_master_key_raw(key);
3337
3338 // Spans four 1 MiB chunks (three full plus a remainder), so sealed
3339 // chunks straddle part boundaries rather than lining up with them.
3340 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
3341 .map(|i| i as u8)
3342 .collect();
3343 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
3344 let file = temp_blob("big.bin", &plaintext);
3345
3346 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3347 let part_size = 1024 * 1024;
3348 let part_count = mount_session(&server, cipher_len, part_size).await;
3349 assert!(part_count > 1, "the fixture must actually be multipart");
3350
3351 client.blob_upload_streaming(&hash, &file).await.unwrap();
3352
3353 let reqs = server.received_requests().await.unwrap();
3354
3355 // The session was sized in ciphertext, predicted from the plaintext.
3356 let start: serde_json::Value = reqs
3357 .iter()
3358 .find(|r| r.url.path() == START_PATH)
3359 .unwrap()
3360 .body_json()
3361 .unwrap();
3362 assert_eq!(start["size_bytes"].as_u64().unwrap(), cipher_len as u64);
3363 assert_eq!(start["hash"].as_str().unwrap(), hash);
3364
3365 // Every part carried exactly the length the server signed for it.
3366 let puts: Vec<&wiremock::Request> = reqs
3367 .iter()
3368 .filter(|r| r.url.path() == PART_PUT_PATH)
3369 .collect();
3370 assert_eq!(puts.len() as u32, part_count, "one PUT per planned part");
3371 for (i, put) in puts.iter().enumerate() {
3372 let expected = if i as u32 == part_count - 1 {
3373 cipher_len - part_size * (part_count as usize - 1)
3374 } else {
3375 part_size
3376 };
3377 assert_eq!(put.body.len(), expected, "part {} length", i + 1);
3378 }
3379
3380 // Each part was requested with the SHA-256 of exactly the bytes that
3381 // part then carried, which is what S3 rehashes against at write time.
3382 // The pairing is what matters: a checksum bound to the wrong part is
3383 // worse than none, since it would reject a correct upload.
3384 let part_reqs: Vec<&wiremock::Request> =
3385 reqs.iter().filter(|r| r.url.path() == PARTS_PATH).collect();
3386 assert_eq!(
3387 part_reqs.len() as u32,
3388 part_count,
3389 "one URL request per part: a digest exists only once the part is sealed"
3390 );
3391 for (i, req) in part_reqs.iter().enumerate() {
3392 let body: serde_json::Value = req.body_json().unwrap();
3393 assert_eq!(body["first_part"].as_u64().unwrap(), i as u64 + 1);
3394 assert_eq!(body["count"].as_u64().unwrap(), 1);
3395 let declared = body["checksums"][0].as_str().unwrap();
3396 let expected = base64::engine::general_purpose::STANDARD
3397 .encode(sha2::Sha256::digest(&puts[i].body));
3398 assert_eq!(
3399 declared,
3400 expected,
3401 "part {} checksum must match its bytes",
3402 i + 1
3403 );
3404 // And the client must actually send it: it is a signed header, so
3405 // dropping it would fail SigV4 at S3.
3406 assert_eq!(
3407 puts[i]
3408 .headers
3409 .get("x-amz-checksum-sha256")
3410 .expect("the PUT must carry the checksum header")
3411 .to_str()
3412 .unwrap(),
3413 declared
3414 );
3415 }
3416
3417 // The concatenated parts are a valid v3 blob for this content address:
3418 // proof that streaming produced the same wire format as the in-memory
3419 // encrypt, boundaries and all.
3420 let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
3421 assert_eq!(assembled.len(), cipher_len);
3422 let decrypted =
3423 synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap();
3424 assert_eq!(decrypted, plaintext, "streamed blob must round-trip");
3425
3426 // Complete named every part, in order, with the ETag S3 returned.
3427 let complete: serde_json::Value = reqs
3428 .iter()
3429 .find(|r| r.url.path() == COMPLETE_PATH)
3430 .unwrap()
3431 .body_json()
3432 .unwrap();
3433 let named = complete["parts"].as_array().unwrap();
3434 assert_eq!(named.len() as u32, part_count);
3435 for (i, part) in named.iter().enumerate() {
3436 assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1);
3437 assert_eq!(part["etag"].as_str().unwrap(), "\"part-etag\"");
3438 }
3439 assert_eq!(hits(&reqs, ABORT_PATH), 0, "a clean upload must not abort");
3440
3441 std::fs::remove_file(&file).ok();
3442 }
3443
3444 #[tokio::test]
3445 async fn streaming_upload_rejects_a_server_part_plan_that_lies_about_geometry() {
3446 let server = MockServer::start().await;
3447 let client = authed_client(&server);
3448 let key = synckit_client::crypto::generate_master_key();
3449 client.set_master_key_raw(key);
3450
3451 // A blob that genuinely spans several 1 MiB parts.
3452 let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
3453 .map(|i| i as u8)
3454 .collect();
3455 let hash = hex::encode(sha2::Sha256::digest(&plaintext));
3456 let file = temp_blob("liar.bin", &plaintext);
3457
3458 // Hostile server: claims the whole multi-part blob fits in ONE part.
3459 // Trusting it would defeat the one-part-in-memory bound, so the client
3460 // must refuse before minting or PUTting anything.
3461 Mock::given(method("POST"))
3462 .and(path(START_PATH))
3463 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3464 "upload_id": "test-upload-id",
3465 "part_size": 1024 * 1024,
3466 "part_count": 1,
3467 "already_exists": false,
3468 })))
3469 .mount(&server)
3470 .await;
3471
3472 let err = client
3473 .blob_upload_streaming(&hash, &file)
3474 .await
3475 .unwrap_err();
3476 assert!(
3477 matches!(err, SyncKitError::Internal(ref m) if m.contains("does not match")),
3478 "expected a geometry-mismatch rejection, got {err:?}"
3479 );
3480 let reqs = server.received_requests().await.unwrap();
3481 assert_eq!(
3482 hits(&reqs, PART_PUT_PATH),
3483 0,
3484 "no part may be uploaded once the plan is rejected"
3485 );
3486
3487 std::fs::remove_file(&file).ok();
3488 }
3489
3490 #[tokio::test]
3491 async fn streaming_upload_handles_an_empty_file() {
3492 let server = MockServer::start().await;
3493 let client = authed_client(&server);
3494 let key = synckit_client::crypto::generate_master_key();
3495 client.set_master_key_raw(key);
3496
3497 let hash = hex::encode(sha2::Sha256::digest(b""));
3498 let file = temp_blob("empty.bin", b"");
3499 let cipher_len = synckit_client::crypto::blob_encrypted_len(0);
3500 mount_session(&server, cipher_len, 1024 * 1024).await;
3501
3502 client.blob_upload_streaming(&hash, &file).await.unwrap();
3503
3504 let reqs = server.received_requests().await.unwrap();
3505 let put = reqs.iter().find(|r| r.url.path() == PART_PUT_PATH).unwrap();
3506 assert_eq!(
3507 put.body.len(),
3508 cipher_len,
3509 "one part carries the whole blob"
3510 );
3511 assert_eq!(
3512 synckit_client::crypto::decrypt_blob_chunked(&put.body, &key, &hash).unwrap(),
3513 Vec::<u8>::new(),
3514 "an empty blob is still an authenticated single chunk"
3515 );
3516
3517 std::fs::remove_file(&file).ok();
3518 }
3519
3520 #[tokio::test]
3521 async fn streaming_upload_skips_when_the_server_already_has_the_content() {
3522 let server = MockServer::start().await;
3523 let client = authed_client(&server);
3524 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3525
3526 Mock::given(method("POST"))
3527 .and(path(START_PATH))
3528 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3529 "upload_id": "",
3530 "part_size": 0,
3531 "part_count": 0,
3532 "already_exists": true,
3533 })))
3534 .mount(&server)
3535 .await;
3536
3537 let plaintext = b"content the server already holds";
3538 let hash = hex::encode(sha2::Sha256::digest(plaintext));
3539 let file = temp_blob("dedup.bin", plaintext);
3540
3541 client.blob_upload_streaming(&hash, &file).await.unwrap();
3542
3543 // Dedup must cost nothing: no file bytes read out to the wire, no
3544 // session to clean up.
3545 let reqs = server.received_requests().await.unwrap();
3546 assert_eq!(hits(&reqs, PART_PUT_PATH), 0, "dedup must not upload parts");
3547 assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
3548 assert_eq!(hits(&reqs, ABORT_PATH), 0);
3549
3550 std::fs::remove_file(&file).ok();
3551 }
3552
3553 #[tokio::test]
3554 async fn streaming_upload_aborts_when_the_file_no_longer_matches_its_hash() {
3555 // The caller hashed the file in an earlier pass. If it changed since,
3556 // storing it under the stale content address would poison the address:
3557 // every later download would re-hash and reject it. Fail here instead,
3558 // and release the parts.
3559 let server = MockServer::start().await;
3560 let client = authed_client(&server);
3561 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3562
3563 let plaintext = b"the bytes actually on disk";
3564 let stale_hash = hex::encode(sha2::Sha256::digest(b"what the caller hashed earlier"));
3565 let file = temp_blob("changed.bin", plaintext);
3566
3567 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3568 mount_session(&server, cipher_len, 1024 * 1024).await;
3569 Mock::given(method("POST"))
3570 .and(path(ABORT_PATH))
3571 .respond_with(ResponseTemplate::new(204))
3572 .mount(&server)
3573 .await;
3574
3575 let err = client
3576 .blob_upload_streaming(&stale_hash, &file)
3577 .await
3578 .expect_err("a hash mismatch must not be uploaded");
3579 assert!(
3580 matches!(err, SyncKitError::IntegrityFailed { .. }),
3581 "expected IntegrityFailed, got {err:?}"
3582 );
3583
3584 let reqs = server.received_requests().await.unwrap();
3585 assert_eq!(
3586 hits(&reqs, COMPLETE_PATH),
3587 0,
3588 "a mismatched blob must not be assembled"
3589 );
3590 assert_eq!(hits(&reqs, ABORT_PATH), 1, "the session must be released");
3591
3592 std::fs::remove_file(&file).ok();
3593 }
3594
3595 #[tokio::test]
3596 async fn streaming_upload_aborts_when_a_part_upload_fails() {
3597 // Parts already sent are billed until the session is aborted, so any
3598 // failure past `start` has to release it.
3599 let server = MockServer::start().await;
3600 let client = authed_client(&server);
3601 client.set_master_key_raw(synckit_client::crypto::generate_master_key());
3602
3603 let plaintext = b"a blob whose part upload will fail";
3604 let hash = hex::encode(sha2::Sha256::digest(plaintext));
3605 let file = temp_blob("failing.bin", plaintext);
3606 let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
3607
3608 Mock::given(method("POST"))
3609 .and(path(START_PATH))
3610 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3611 "upload_id": "test-upload-id",
3612 "part_size": cipher_len,
3613 "part_count": 1,
3614 "already_exists": false,
3615 })))
3616 .mount(&server)
3617 .await;
3618 Mock::given(method("POST"))
3619 .and(path(PARTS_PATH))
3620 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3621 "parts": [{
3622 "part_number": 1,
3623 "content_length": cipher_len,
3624 "url": format!("{}{PART_PUT_PATH}", server.uri()),
3625 }]
3626 })))
3627 .mount(&server)
3628 .await;
3629 Mock::given(method("PUT"))
3630 .and(path(PART_PUT_PATH))
3631 .respond_with(ResponseTemplate::new(403))
3632 .mount(&server)
3633 .await;
3634 Mock::given(method("POST"))
3635 .and(path(ABORT_PATH))
3636 .respond_with(ResponseTemplate::new(204))
3637 .mount(&server)
3638 .await;
3639
3640 let err = client
3641 .blob_upload_streaming(&hash, &file)
3642 .await
3643 .unwrap_err();
3644 assert!(
3645 matches!(err, SyncKitError::Server { status: 403, .. }),
3646 "got {err:?}"
3647 );
3648
3649 let reqs = server.received_requests().await.unwrap();
3650 assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
3651 assert_eq!(
3652 hits(&reqs, ABORT_PATH),
3653 1,
3654 "a failed transfer must release its parts"
3655 );
3656
3657 std::fs::remove_file(&file).ok();
3658 }
3659 }
3660