Skip to main content

max / synckit

52.8 KB · 1456 lines History Blame Raw
1 //! Internal HTTP plumbing shared across the client modules.
2 //!
3 //! Response-status to [`SyncKitError`] mapping, idempotency-key generation,
4 //! size-capped body reads (so a hostile server cannot exhaust memory on a
5 //! control response), wire-version negotiation, and JWT-expiry parsing. This
6 //! module is `pub(crate)`: none of it is part of the SDK's public API.
7
8 use base64::Engine;
9
10 use crate::{
11 crypto,
12 error::{Result, SyncKitError},
13 types::{ChangeEntry, Hlc, PullChangeEntry, WireChangeEntry},
14 };
15
16 use super::{BASE_DELAY, MAX_RETRIES, SyncKitClient};
17
18 /// Proof, supplied at the call site, that retrying an operation is sound.
19 ///
20 /// Retry replays the request, so it is only safe when the server treats the
21 /// operation idempotently. Requiring this argument makes that judgement an
22 /// explicit, reviewable decision, and the judgement has runtime teeth:
23 /// [`may_retry`](Self::may_retry) gates the retry loop, so an
24 /// [`Unsafe`](Self::Unsafe) operation physically cannot be auto-replayed.
25 ///
26 /// The variants split "safe to replay" into a **read** and a **write that the
27 /// server dedups**, and the write variant must *name* what makes replay
28 /// idempotent. That is the forcing function: a mutating create tagged
29 /// auto-retryable without naming its dedup key does not compile, so the
30 /// `ota_create_release`/`ota_register_artifact` class, a `CREATE` that silently
31 /// relied on a server unique constraint nobody had written down, cannot recur
32 /// unnoticed.
33 #[derive(Clone, Copy, Debug)]
34 pub(super) enum Idempotency {
35 /// A read with no server-side effect (a `GET`, or a `POST` that only reads).
36 /// Replay is trivially safe.
37 ReadOnly,
38 /// A write whose replay collapses to a single effect because the server
39 /// dedups on `on`, a content-addressed key, an optimistic version, or a
40 /// `UNIQUE` constraint. `on` documents *why* replay is harmless, at the call
41 /// site, and naming it is mandatory.
42 IdempotentWrite {
43 /// The key/constraint the server dedups on (e.g. `"(app_id, version)
44 /// unique"`). Documentation, surfaced in review and logs.
45 on: &'static str,
46 },
47 /// Carries a client-generated idempotency key (e.g. push's `batch_id`) so
48 /// the server collapses duplicate deliveries into one effect.
49 Keyed,
50 /// Not idempotent and not idempotency-keyed: replaying it would mint
51 /// duplicate external state (e.g. a second Stripe Checkout session). The
52 /// retry helpers attempt it exactly once.
53 Unsafe,
54 }
55
56 impl Idempotency {
57 /// Whether the retry helpers may replay this operation on a transient error.
58 fn may_retry(self) -> bool {
59 matches!(
60 self,
61 Idempotency::ReadOnly | Idempotency::IdempotentWrite { .. } | Idempotency::Keyed
62 )
63 }
64
65 /// The documented dedup basis for an idempotent write (`None` for the other
66 /// variants), surfaced in the retry log so a replayed request records *why*
67 /// replay was safe.
68 fn dedup_basis(self) -> Option<&'static str> {
69 match self {
70 Idempotency::IdempotentWrite { on } => Some(on),
71 _ => None,
72 }
73 }
74 }
75
76 /// Upper bound on a control/JSON/error response body read into memory. Blob
77 /// bodies stream through [`super::SyncKitClient::blob_download`]'s own 4 GiB cap;
78 /// every *other* response, JSON control replies, OTA manifests, SSE-open errors,
79 /// 4xx/5xx error bodies, is small, so a hostile or buggy server streaming a
80 /// multi-gigabyte body into one of them is a pure OOM lever. 8 MiB is generous
81 /// for any legitimate control body.
82 pub(super) const MAX_CONTROL_BODY_BYTES: usize = 8 * 1024 * 1024;
83
84 /// Read a response body into memory with a hard byte cap, the ONE sanctioned
85 /// way to buffer a non-blob body.
86 ///
87 /// `reqwest`'s `Response::json`/`text`/`bytes` read the whole body with no size
88 /// limit, so a hostile server can stream an arbitrarily large body and OOM the
89 /// client (worst case: the public, unauthenticated OTA updater check). This
90 /// drains `bytes_stream()` and aborts the instant the running total exceeds
91 /// `limit`, and fast-rejects on an honest oversized `Content-Length` before
92 /// reading a byte. Those raw reader methods are banned by `clippy.toml`
93 /// (`disallowed-methods`) so a new call site cannot reintroduce an uncapped read.
94 pub(super) async fn read_body_capped(
95 resp: reqwest::Response,
96 limit: usize,
97 ) -> Result<bytes::Bytes> {
98 use tokio_stream::StreamExt;
99 if let Some(len) = resp.content_length()
100 && len > limit as u64
101 {
102 return Err(SyncKitError::Internal(format!(
103 "response Content-Length {len} exceeds {limit}-byte cap"
104 )));
105 }
106 let mut stream = resp.bytes_stream();
107 let mut buf = bytes::BytesMut::new();
108 while let Some(chunk) = stream.next().await {
109 let chunk = chunk.map_err(SyncKitError::Http)?;
110 if buf.len() + chunk.len() > limit {
111 return Err(SyncKitError::Internal(format!(
112 "response body exceeds {limit}-byte cap"
113 )));
114 }
115 buf.extend_from_slice(&chunk);
116 }
117 Ok(buf.freeze())
118 }
119
120 /// [`read_body_capped`] + JSON deserialize, the capped replacement for
121 /// `resp.json::<T>()`.
122 pub(super) async fn read_json_capped<T: serde::de::DeserializeOwned>(
123 resp: reqwest::Response,
124 limit: usize,
125 ) -> Result<T> {
126 let bytes = read_body_capped(resp, limit).await?;
127 Ok(serde_json::from_slice(&bytes)?)
128 }
129
130 /// [`read_body_capped`] + lossy UTF-8, the capped replacement for `resp.text()`
131 /// on error bodies (which are only logged, so lossy decoding and a `""` fallback
132 /// on an oversized/failed read are fine).
133 pub(super) async fn read_text_capped(resp: reqwest::Response, limit: usize) -> String {
134 match read_body_capped(resp, limit).await {
135 Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
136 Err(_) => String::new(),
137 }
138 }
139
140 /// Current on-wire HLC-envelope version, written into the `__skver` tag.
141 const ENVELOPE_VERSION: u64 = 2;
142
143 /// The on-wire HLC-envelope version, parsed from the `__skver` tag.
144 ///
145 /// Parsing is the explicit dispatch point: an unknown (future) version is a
146 /// hard error, never a silent fallback. That closes the X2 hazard, a protocol
147 /// bump that an older client cannot understand fails loudly instead of being
148 /// mis-decoded as a bare row and corrupting the clock.
149 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
150 enum WireVersion {
151 /// The current envelope: `{ __skver: 2, __skhlc, data }`.
152 V2,
153 }
154
155 impl WireVersion {
156 fn parse(tag: u64) -> Result<Self> {
157 match tag {
158 2 => Ok(WireVersion::V2),
159 other => Err(SyncKitError::Crypto(format!(
160 "unknown sync envelope version {other}; this client is too old to read it"
161 ))),
162 }
163 }
164 }
165
166 impl SyncKitClient {
167 /// Retry an async HTTP operation with exponential backoff.
168 ///
169 /// Retries on transient errors (network failures, 5xx, 429) up to [`MAX_RETRIES`]
170 /// times with delays of 1s, 2s, 4s. Returns the last error if all attempts fail.
171 /// Client errors (4xx except 429) are considered permanent and returned immediately.
172 ///
173 /// `idempotency` is the caller's proof that replay is safe, see [`Idempotency`].
174 pub(super) async fn retry_request<F, Fut>(
175 &self,
176 idempotency: Idempotency,
177 mut operation: F,
178 ) -> Result<reqwest::Response>
179 where
180 F: FnMut() -> Fut,
181 Fut: std::future::Future<Output = Result<reqwest::Response>>,
182 {
183 // A non-idempotent operation gets exactly one attempt: zero retries.
184 let max_attempts = if idempotency.may_retry() {
185 MAX_RETRIES
186 } else {
187 0
188 };
189 let mut last_err = None;
190
191 for attempt in 0..=max_attempts {
192 match operation().await {
193 Ok(resp) => return Ok(resp),
194 Err(err) => {
195 if !is_transient(&err) {
196 return Err(err);
197 }
198
199 if attempt < max_attempts {
200 let delay = retry_delay(&err, attempt);
201 tracing::debug!(
202 attempt = attempt + 1,
203 max_retries = MAX_RETRIES,
204 delay_ms = delay.as_millis() as u64,
205 error = %err,
206 idempotency_basis = idempotency.dedup_basis(),
207 "Transient error, retrying after backoff",
208 );
209 tokio::time::sleep(delay).await;
210 }
211
212 last_err = Some(err);
213 }
214 }
215 }
216
217 Err(last_err.expect("loop ran at least once"))
218 }
219
220 /// Retry an HTTP operation and deserialize the JSON response body inside
221 /// the retry loop. This ensures a transient body-read failure (truncated
222 /// response, connection reset mid-body) is retried rather than surfacing
223 /// as a permanent error after the server already committed the operation.
224 ///
225 /// `idempotency` is the caller's proof that replay is safe, see [`Idempotency`].
226 pub(super) async fn retry_request_json<F, Fut, T>(
227 &self,
228 idempotency: Idempotency,
229 mut operation: F,
230 ) -> Result<T>
231 where
232 F: FnMut() -> Fut,
233 Fut: std::future::Future<Output = Result<reqwest::Response>>,
234 T: serde::de::DeserializeOwned,
235 {
236 // A non-idempotent operation gets exactly one attempt: zero retries.
237 let max_attempts = if idempotency.may_retry() {
238 MAX_RETRIES
239 } else {
240 0
241 };
242 let mut last_err = None;
243
244 for attempt in 0..=max_attempts {
245 match operation().await {
246 Ok(resp) => match read_json_capped::<T>(resp, MAX_CONTROL_BODY_BYTES).await {
247 Ok(parsed) => return Ok(parsed),
248 Err(e) => {
249 let err = e;
250 if attempt < max_attempts {
251 let delay = retry_delay(&err, attempt);
252 tracing::debug!(
253 attempt = attempt + 1,
254 max_retries = MAX_RETRIES,
255 delay_ms = delay.as_millis() as u64,
256 error = %err,
257 "Response body read failed, retrying",
258 );
259 tokio::time::sleep(delay).await;
260 }
261 last_err = Some(err);
262 }
263 },
264 Err(err) => {
265 if !is_transient(&err) {
266 return Err(err);
267 }
268
269 if attempt < max_attempts {
270 let delay = retry_delay(&err, attempt);
271 tracing::debug!(
272 attempt = attempt + 1,
273 max_retries = max_attempts,
274 delay_ms = delay.as_millis() as u64,
275 error = %err,
276 "Transient error, retrying after backoff",
277 );
278 tokio::time::sleep(delay).await;
279 }
280
281 last_err = Some(err);
282 }
283 }
284 }
285
286 Err(last_err.expect("loop ran at least once"))
287 }
288
289 /// Encrypt a change entry for the wire. Every change now seals an HLC
290 /// envelope (Deletes included), so the master key is always required.
291 #[cfg(test)]
292 pub(super) fn encrypt_change(&self, entry: ChangeEntry) -> Result<WireChangeEntry> {
293 let master_key = self.require_master_key()?;
294 Self::encrypt_change_with_key(entry, &master_key)
295 }
296
297 /// Decrypt a pulled legacy entry that has no encrypted payload (a pre-HLC
298 /// Delete). The HLC is synthesized from the entry's `client_timestamp`. No key
299 /// needed.
300 #[cfg(test)]
301 pub(super) fn decrypt_change_no_data(entry: PullChangeEntry) -> Result<ChangeEntry> {
302 debug_assert!(entry.data.is_none());
303 Ok(ChangeEntry {
304 table: entry.table,
305 op: entry.op,
306 row_id: entry.row_id,
307 hlc: Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
308 timestamp: entry.timestamp,
309 data: None,
310 extra: serde_json::Map::default(),
311 })
312 }
313
314 /// Wrap a change's HLC and payload into one JSON envelope. Encrypting the
315 /// envelope (rather than the bare row payload) is what carries the HLC inside
316 /// the E2E ciphertext, including for Deletes, which have no row payload. The
317 /// server stores the ciphertext opaquely and never sees the clock.
318 ///
319 /// The `__skver` tag is a positive, explicit version marker. A reader
320 /// dispatches on it (see [`WireVersion`]) rather than structurally guessing,
321 /// so a future format bump is rejected loudly instead of silently misread.
322 fn hlc_envelope(hlc: &Hlc, data: Option<&serde_json::Value>) -> serde_json::Value {
323 serde_json::json!({ "__skver": ENVELOPE_VERSION, "__skhlc": hlc, "data": data })
324 }
325
326 /// Split a decrypted payload back into `(hlc, data)`.
327 ///
328 /// Dispatch is explicit, not structural:
329 /// - A `__skver` tag means a versioned envelope; the version is parsed via
330 /// [`WireVersion::parse`], which **errors loudly** on an unknown future
331 /// version rather than silently falling back to a bare-row read (which
332 /// would corrupt the clock, the X2 hazard).
333 /// - No `__skver` but an embedded `__skhlc` that parses is a gen-1 envelope
334 /// (predates the version tag).
335 /// - Anything else is a legacy bare-row payload whose HLC is synthesized
336 /// from `node` + `timestamp_ms`.
337 fn split_hlc_envelope(
338 decrypted: serde_json::Value,
339 node: crate::ids::DeviceId,
340 timestamp_ms: i64,
341 ) -> Result<(Hlc, Option<serde_json::Value>)> {
342 if let Some(obj) = decrypted.as_object() {
343 if let Some(tag) = obj.get("__skver") {
344 // Explicit version present: dispatch, rejecting unknown loudly.
345 let tag = tag.as_u64().ok_or_else(|| {
346 SyncKitError::Crypto("envelope __skver tag is not an integer".into())
347 })?;
348 return match WireVersion::parse(tag)? {
349 WireVersion::V2 => {
350 let hlc = obj
351 .get("__skhlc")
352 .and_then(|v| serde_json::from_value::<Hlc>(v.clone()).ok())
353 .ok_or_else(|| {
354 SyncKitError::Crypto("v2 envelope missing __skhlc".into())
355 })?;
356 let data = obj.get("data").cloned().filter(|v| !v.is_null());
357 Ok((hlc, data))
358 }
359 };
360 }
361 // gen-1 envelope: embedded HLC, no version tag.
362 if let Some(hlc) = obj
363 .get("__skhlc")
364 .and_then(|v| serde_json::from_value::<Hlc>(v.clone()).ok())
365 {
366 let data = obj.get("data").cloned().filter(|v| !v.is_null());
367 return Ok((hlc, data));
368 }
369 }
370 // Legacy bare-row payload: the decrypted value is the row data itself.
371 Ok((Hlc::from_legacy(timestamp_ms, node), Some(decrypted)))
372 }
373
374 /// Encrypt with a pre-loaded key. Used by `push()` to avoid per-entry lock
375 /// acquisition. Every change (including Deletes) is encrypted, because the HLC
376 /// envelope always needs sealing. The ciphertext is bound to its
377 /// `(table, row_id)` address via AEAD associated data, so a server cannot
378 /// relocate it to another row/table without the open failing closed.
379 pub(super) fn encrypt_change_with_key(
380 entry: ChangeEntry,
381 master_key: &[u8; 32],
382 ) -> Result<WireChangeEntry> {
383 let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
384 let envelope = Self::hlc_envelope(&entry.hlc, entry.data.as_ref());
385 let encrypted_data = Some(crypto::encrypt_json_aad(&envelope, master_key, &ctx)?);
386
387 Ok(WireChangeEntry {
388 table: entry.table,
389 op: entry.op,
390 row_id: entry.row_id,
391 timestamp: entry.timestamp,
392 data: encrypted_data,
393 })
394 }
395
396 /// Decrypt the data field of a pulled change entry.
397 #[cfg(test)]
398 pub(super) fn decrypt_change(&self, entry: PullChangeEntry) -> Result<ChangeEntry> {
399 if entry.data.is_some() {
400 let master_key = self.require_master_key()?;
401 Self::decrypt_change_with_key(entry, &master_key)
402 } else {
403 Self::decrypt_change_no_data(entry)
404 }
405 }
406
407 /// Decrypt with a pre-loaded key, preserving `device_id` and `seq` in a [`PulledChange`].
408 ///
409 /// Used by `pull_rich()` to produce conflict-detection-ready results.
410 pub(super) fn decrypt_change_to_pulled(
411 entry: PullChangeEntry,
412 master_key: &[u8; 32],
413 ) -> Result<crate::types::PulledChange> {
414 let device_id = entry.device_id;
415 let seq = entry.seq;
416 let decrypted = Self::decrypt_change_with_key(entry, master_key)?;
417 Ok(crate::types::PulledChange {
418 entry: decrypted,
419 device_id,
420 seq,
421 })
422 }
423
424 /// Decrypt a pulled entry during a rotation window, selecting the key by the
425 /// entry's `key_id`. Generic over the decrypt step so both the plain-pull
426 /// (`ChangeEntry`) and rich-pull (`PulledChange`) paths share exactly this
427 /// selection logic, including the unknown-`key_id` fallback that tries the
428 /// primary key then the pending key. Previously the live pull path
429 /// reimplemented a fallback-less variant while the tested one sat unused.
430 pub(super) fn decrypt_with_rotation_keys<T, F>(
431 entry: PullChangeEntry,
432 primary_key: &[u8; 32],
433 primary_key_id: i32,
434 pending_key: &[u8; 32],
435 pending_key_id: i32,
436 decrypt_fn: &F,
437 ) -> Result<T>
438 where
439 F: Fn(PullChangeEntry, &[u8; 32]) -> Result<T>,
440 {
441 let effective_key_id = entry.key_id.unwrap_or(1);
442 if effective_key_id == pending_key_id {
443 decrypt_fn(entry, pending_key)
444 } else if effective_key_id == primary_key_id || effective_key_id <= 1 {
445 decrypt_fn(entry, primary_key)
446 } else {
447 // Unknown key_id, try primary, then fall back to pending.
448 match decrypt_fn(entry.clone(), primary_key) {
449 Ok(result) => Ok(result),
450 Err(_) => decrypt_fn(entry, pending_key),
451 }
452 }
453 }
454
455 /// Encrypt a change for a **group** changelog, sealed under the group's GCK
456 /// and bound to `(group_id, table, row_id)` as AEAD associated data. Same HLC
457 /// envelope as the personal path; the extra `group_id` binding makes a
458 /// ciphertext non-relocatable across groups (the p1 crypto guarantee).
459 pub(super) fn encrypt_group_change_with_key(
460 group_id: &str,
461 entry: ChangeEntry,
462 gck: &[u8; 32],
463 ) -> Result<WireChangeEntry> {
464 let ctx = crypto::AeadContext::group_entry(group_id, &entry.table, &entry.row_id);
465 let envelope = Self::hlc_envelope(&entry.hlc, entry.data.as_ref());
466 let encrypted_data = Some(crypto::encrypt_json_aad(&envelope, gck, &ctx)?);
467
468 Ok(WireChangeEntry {
469 table: entry.table,
470 op: entry.op,
471 row_id: entry.row_id,
472 timestamp: entry.timestamp,
473 data: encrypted_data,
474 })
475 }
476
477 /// Decrypt a pulled group entry under the group's GCK, preserving `device_id`
478 /// and `seq` in a [`PulledChange`]. The counterpart to
479 /// [`encrypt_group_change_with_key`](Self::encrypt_group_change_with_key).
480 /// Group pull has no master-key-rotation window (a group's key rotates
481 /// server-side, re-encrypted in place), so this is the whole decrypt path.
482 pub(super) fn decrypt_group_change_to_pulled(
483 group_id: &str,
484 entry: PullChangeEntry,
485 gck: &[u8; 32],
486 ) -> Result<crate::types::PulledChange> {
487 let device_id = entry.device_id;
488 let seq = entry.seq;
489 let ctx = crypto::AeadContext::group_entry(group_id, &entry.table, &entry.row_id);
490 let (hlc, data) = match entry.data {
491 Some(ref value) => {
492 let decrypted = crypto::decrypt_json_aad(value, gck, &ctx)?;
493 Self::split_hlc_envelope(decrypted, device_id, entry.timestamp.timestamp_millis())?
494 }
495 None => (
496 Hlc::from_legacy(entry.timestamp.timestamp_millis(), device_id),
497 None,
498 ),
499 };
500 Ok(crate::types::PulledChange {
501 entry: ChangeEntry {
502 table: entry.table,
503 op: entry.op,
504 row_id: entry.row_id,
505 hlc,
506 timestamp: entry.timestamp,
507 data,
508 extra: serde_json::Map::default(),
509 },
510 device_id,
511 seq,
512 })
513 }
514
515 /// Decrypt with a pre-loaded key. Used by `pull()` to avoid per-entry lock acquisition.
516 pub(super) fn decrypt_change_with_key(
517 entry: PullChangeEntry,
518 master_key: &[u8; 32],
519 ) -> Result<ChangeEntry> {
520 let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
521 let (hlc, data) = match entry.data {
522 Some(ref value) => {
523 let decrypted = crypto::decrypt_json_aad(value, master_key, &ctx)?;
524 Self::split_hlc_envelope(
525 decrypted,
526 entry.device_id,
527 entry.timestamp.timestamp_millis(),
528 )?
529 }
530 None => (
531 Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
532 None,
533 ),
534 };
535
536 Ok(ChangeEntry {
537 table: entry.table,
538 op: entry.op,
539 row_id: entry.row_id,
540 hlc,
541 timestamp: entry.timestamp,
542 data,
543 extra: serde_json::Map::default(),
544 })
545 }
546 }
547
548 /// Extract the `exp` claim from a JWT without verifying the signature.
549 ///
550 /// JWTs are `header.payload.signature` where the payload is base64url-encoded JSON.
551 /// We decode the payload segment and read the `exp` field. Returns `None` if
552 /// the token is malformed or `exp` is missing.
553 pub(super) fn jwt_exp(token: &str) -> Option<i64> {
554 let payload = token.split('.').nth(1)?;
555 let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
556 .decode(payload)
557 .ok()?;
558 let claims: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
559 claims["exp"].as_i64()
560 }
561
562 /// Returns `true` if the token's `exp` claim is within [`TOKEN_EXPIRY_BUFFER_SECS`]
563 /// of the current time (or already past). Returns `false` if the token cannot
564 /// be decoded, in that case, let the server decide.
565 #[cfg(test)]
566 pub(super) fn token_is_expired(token: &str) -> bool {
567 let Some(exp) = jwt_exp(token) else {
568 return false;
569 };
570 let now = chrono::Utc::now().timestamp();
571 now >= exp - super::TOKEN_EXPIRY_BUFFER_SECS
572 }
573
574 /// Check an HTTP response for errors, returning the response on success.
575 ///
576 /// For 429 responses, parses the `Retry-After` header so the retry loop
577 /// can use it instead of the default exponential backoff delay.
578 pub(super) async fn check_response(resp: reqwest::Response) -> Result<reqwest::Response> {
579 let status = resp.status().as_u16();
580 if status >= 400 {
581 let retry_after_secs = parse_retry_after(&resp);
582 // Capped read: this error path sits in front of every request (including
583 // the otherwise-capped blob/SSE streams), so an uncapped `resp.text()`
584 // here let a hostile server bypass those caps with a giant error body.
585 let message = read_text_capped(resp, MAX_CONTROL_BODY_BYTES).await;
586 return Err(SyncKitError::Server {
587 status,
588 message,
589 retry_after_secs,
590 });
591 }
592 Ok(resp)
593 }
594
595 /// Extract the `Retry-After` header from an HTTP response as seconds.
596 /// Returns `None` if the header is absent, non-numeric, or zero.
597 fn parse_retry_after(resp: &reqwest::Response) -> Option<u64> {
598 resp.headers()
599 .get("retry-after")
600 .and_then(|v| v.to_str().ok())
601 .and_then(|v| v.parse::<u64>().ok())
602 .filter(|&secs| secs > 0)
603 }
604
605 /// Compute the backoff delay for a retry attempt.
606 ///
607 /// Uses the server's `Retry-After` header (if present on a 429) capped at
608 /// 60 seconds. Otherwise exponential backoff (1s, 2s, 4s) with ±20% jitter so
609 /// many clients that hit a transient blip simultaneously don't retry in lockstep
610 /// and re-amplify the load.
611 fn retry_delay(err: &SyncKitError, attempt: u32) -> std::time::Duration {
612 if let SyncKitError::Server {
613 retry_after_secs: Some(secs),
614 ..
615 } = err
616 {
617 let capped = (*secs).min(60);
618 return std::time::Duration::from_secs(capped);
619 }
620 jittered(BASE_DELAY * 2u32.pow(attempt))
621 }
622
623 /// Spread a backoff delay by ±20% to avoid synchronized retries (thundering herd).
624 fn jittered(base: std::time::Duration) -> std::time::Duration {
625 use rand::RngExt;
626 let millis = base.as_millis() as u64;
627 let span = millis / 5; // 20%
628 if span == 0 {
629 return base;
630 }
631 // Uniform in [millis - span, millis + span].
632 let delta = rand::rng().random_range(0..=2 * span);
633 std::time::Duration::from_millis(millis - span + delta)
634 }
635
636 /// Returns true if the error is transient and worth retrying.
637 ///
638 /// Transient errors:
639 /// - Network-level failures (connection refused, timeout, DNS, etc.)
640 /// - Server errors (5xx)
641 /// - Rate limiting (429)
642 ///
643 /// Permanent errors (not retried):
644 /// - Client errors (4xx except 429), bad request, auth failure, not found, etc.
645 /// - Serialization errors, encryption errors, missing session, etc.
646 pub(super) fn is_transient(err: &SyncKitError) -> bool {
647 match err {
648 SyncKitError::Http(e) => {
649 // Transport errors (timeout, connect, DNS) are transient.
650 // Builder errors are programming mistakes, redirect loops and
651 // body decode failures are permanent, retrying won't help.
652 !e.is_builder() && !e.is_redirect() && !e.is_decode()
653 }
654 SyncKitError::Server { status, .. } => {
655 // 5xx = server error (transient), 429 = rate limited (transient)
656 *status >= 500 || *status == 429
657 }
658 // Everything else (auth, crypto, serialization, keychain) is permanent.
659 // Note: some keychain errors (e.g. locked keychain, unavailable
660 // secret-service) are conceptually transient, but keychain operations
661 // are synchronous and not wrapped by retry_request.
662 _ => false,
663 }
664 }
665
666 #[cfg(test)]
667 mod tests {
668 use super::*;
669 use crate::ids::DeviceId;
670 use crate::types::ChangeOp;
671 use base64::Engine;
672 use chrono::Utc;
673 use std::time::Duration;
674 use uuid::Uuid;
675
676 use super::super::TOKEN_EXPIRY_BUFFER_SECS;
677
678 fn test_config() -> super::super::SyncKitConfig {
679 super::super::SyncKitConfig {
680 server_url: "https://example.com".to_string(),
681 api_key: "test-api-key-123".to_string(),
682 }
683 }
684
685 /// Build a fake JWT with the given `exp` claim (no real signature).
686 fn fake_jwt(exp: i64) -> String {
687 let header = base64::engine::general_purpose::URL_SAFE_NO_PAD
688 .encode(r#"{"alg":"HS256","typ":"JWT"}"#);
689 let payload_json = serde_json::json!({
690 "sub": "550e8400-e29b-41d4-a716-446655440000",
691 "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
692 "exp": exp,
693 "iat": exp - 3600,
694 });
695 let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD
696 .encode(payload_json.to_string().as_bytes());
697 let signature = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
698 format!("{header}.{payload}.{signature}")
699 }
700
701 // ── wire-version envelope dispatch ──
702
703 #[test]
704 fn split_envelope_dispatches_on_explicit_version() {
705 let node = DeviceId::new(Uuid::from_u128(1));
706 let hlc = Hlc {
707 wall_ms: 5,
708 counter: 2,
709 node,
710 };
711
712 // v2 envelope: explicit __skver, parsed by version.
713 let v2 = serde_json::json!({ "__skver": 2, "__skhlc": hlc, "data": {"k": "v"} });
714 let (got, data) = SyncKitClient::split_hlc_envelope(v2, node, 0).unwrap();
715 assert_eq!(got, hlc);
716 assert_eq!(data, Some(serde_json::json!({"k": "v"})));
717
718 // gen-1 envelope: __skhlc present, no version tag.
719 let gen1 = serde_json::json!({ "__skhlc": hlc, "data": null });
720 let (got, data) = SyncKitClient::split_hlc_envelope(gen1, node, 0).unwrap();
721 assert_eq!(got, hlc);
722 assert_eq!(data, None);
723
724 // Bare legacy row: HLC synthesized from node + timestamp.
725 let bare = serde_json::json!({ "title": "buy milk" });
726 let (got, data) = SyncKitClient::split_hlc_envelope(bare.clone(), node, 1234).unwrap();
727 assert_eq!(got, Hlc::from_legacy(1234, node));
728 assert_eq!(data, Some(bare));
729 }
730
731 #[test]
732 fn split_envelope_rejects_unknown_version_loudly() {
733 // The X2 hazard: a future envelope version must error, not silently
734 // fall back to a bare-row read (which would corrupt the clock).
735 let node = DeviceId::new(Uuid::from_u128(1));
736 let hlc = Hlc {
737 wall_ms: 5,
738 counter: 0,
739 node,
740 };
741 let future = serde_json::json!({ "__skver": 3, "__skhlc": hlc, "data": null });
742 let err = SyncKitClient::split_hlc_envelope(future, node, 0).unwrap_err();
743 assert!(
744 matches!(err, SyncKitError::Crypto(ref m) if m.contains("envelope version 3")),
745 "unexpected error: {err:?}"
746 );
747 }
748
749 // ── encrypt_change / decrypt_change ──
750
751 #[test]
752 fn delete_seals_hlc_envelope_and_roundtrips() {
753 // A Delete carries no row payload, but its HLC must still travel, so it is
754 // now encrypted into an envelope (data = Some), and decrypt restores the
755 // op, a None payload, and the exact HLC.
756 let client = SyncKitClient::new(test_config());
757 let key = crypto::generate_master_key();
758 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
759
760 let device = DeviceId::new(Uuid::new_v4());
761 let hlc = Hlc {
762 wall_ms: 12_345,
763 counter: 7,
764 node: device,
765 };
766 let entry = ChangeEntry {
767 table: "tasks".to_string(),
768 op: ChangeOp::Delete,
769 row_id: "row-1".to_string(),
770 timestamp: Utc::now(),
771 hlc,
772 data: None,
773 extra: serde_json::Map::default(),
774 };
775
776 let wire = client.encrypt_change(entry).unwrap();
777 assert_eq!(wire.op, ChangeOp::Delete);
778 assert!(
779 wire.data.is_some(),
780 "delete now seals an encrypted HLC envelope"
781 );
782
783 let pull_entry = PullChangeEntry {
784 seq: 1,
785 device_id: device,
786 table: wire.table,
787 op: wire.op,
788 row_id: wire.row_id,
789 timestamp: wire.timestamp,
790 data: wire.data,
791 key_id: None,
792 };
793 let decrypted = client.decrypt_change(pull_entry).unwrap();
794 assert_eq!(decrypted.op, ChangeOp::Delete);
795 assert_eq!(decrypted.row_id, "row-1");
796 assert!(
797 decrypted.data.is_none(),
798 "payload is still None after the envelope unwraps"
799 );
800 assert_eq!(decrypted.hlc, hlc, "HLC survives the round trip");
801 }
802
803 #[test]
804 fn encrypt_change_fails_without_master_key() {
805 let client = SyncKitClient::new(test_config());
806 let entry = ChangeEntry {
807 table: "tasks".to_string(),
808 op: ChangeOp::Insert,
809 row_id: "row-1".to_string(),
810 timestamp: Utc::now(),
811 hlc: Hlc::zero(DeviceId::nil()),
812 data: Some(serde_json::json!({"title": "test"})),
813 extra: serde_json::Map::default(),
814 };
815
816 let err = client.encrypt_change(entry).unwrap_err();
817 assert!(matches!(err, SyncKitError::NoMasterKey));
818 }
819
820 #[test]
821 fn encrypt_change_produces_encrypted_data() {
822 let client = SyncKitClient::new(test_config());
823 let key = crypto::generate_master_key();
824 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
825
826 let original_data = serde_json::json!({"title": "Buy milk", "priority": 3});
827 let entry = ChangeEntry {
828 table: "tasks".to_string(),
829 op: ChangeOp::Insert,
830 row_id: "row-1".to_string(),
831 timestamp: Utc::now(),
832 hlc: Hlc::zero(DeviceId::nil()),
833 data: Some(original_data.clone()),
834 extra: serde_json::Map::default(),
835 };
836
837 let wire = client.encrypt_change(entry).unwrap();
838 assert!(wire.data.is_some());
839 let encrypted = wire.data.unwrap();
840 assert!(encrypted.is_string());
841 assert_ne!(encrypted, original_data);
842 }
843
844 #[test]
845 fn encrypt_decrypt_roundtrip() {
846 let client = SyncKitClient::new(test_config());
847 let key = crypto::generate_master_key();
848 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
849
850 let original_data = serde_json::json!({
851 "title": "Buy milk",
852 "tags": ["groceries", "urgent"],
853 "count": 42
854 });
855 let ts = Utc::now();
856 let entry = ChangeEntry {
857 table: "tasks".to_string(),
858 op: ChangeOp::Update,
859 row_id: "row-abc".to_string(),
860 timestamp: ts,
861 hlc: Hlc::zero(DeviceId::nil()),
862 data: Some(original_data.clone()),
863 extra: serde_json::Map::default(),
864 };
865
866 let wire = client.encrypt_change(entry).unwrap();
867 let pull_entry = PullChangeEntry {
868 seq: 1,
869 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
870 table: wire.table,
871 op: wire.op,
872 row_id: wire.row_id,
873 timestamp: wire.timestamp,
874 data: wire.data,
875 key_id: None,
876 };
877
878 let decrypted = client.decrypt_change(pull_entry).unwrap();
879 assert_eq!(decrypted.table, "tasks");
880 assert_eq!(decrypted.op, ChangeOp::Update);
881 assert_eq!(decrypted.row_id, "row-abc");
882 assert_eq!(decrypted.data.unwrap(), original_data);
883 }
884
885 #[test]
886 fn decrypt_change_with_no_data() {
887 let client = SyncKitClient::new(test_config());
888 let pull_entry = PullChangeEntry {
889 seq: 5,
890 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
891 table: "events".to_string(),
892 op: ChangeOp::Delete,
893 row_id: "evt-1".to_string(),
894 timestamp: Utc::now(),
895 data: None,
896 key_id: None,
897 };
898
899 let decrypted = client.decrypt_change(pull_entry).unwrap();
900 assert_eq!(decrypted.table, "events");
901 assert_eq!(decrypted.op, ChangeOp::Delete);
902 assert!(decrypted.data.is_none());
903 }
904
905 #[test]
906 fn decrypt_change_fails_without_master_key() {
907 let client = SyncKitClient::new(test_config());
908 let pull_entry = PullChangeEntry {
909 seq: 1,
910 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
911 table: "tasks".to_string(),
912 op: ChangeOp::Insert,
913 row_id: "row-1".to_string(),
914 timestamp: Utc::now(),
915 data: Some(serde_json::json!("some-encrypted-string")),
916 key_id: None,
917 };
918
919 let err = client.decrypt_change(pull_entry).unwrap_err();
920 assert!(matches!(err, SyncKitError::NoMasterKey));
921 }
922
923 // ── is_transient error classification ──
924
925 #[test]
926 fn is_transient_server_5xx() {
927 let err = SyncKitError::Server {
928 status: 500,
929 message: "Internal Server Error".to_string(),
930 retry_after_secs: None,
931 };
932 assert!(is_transient(&err));
933 let err = SyncKitError::Server {
934 status: 502,
935 message: "Bad Gateway".to_string(),
936 retry_after_secs: None,
937 };
938 assert!(is_transient(&err));
939 let err = SyncKitError::Server {
940 status: 503,
941 message: "Service Unavailable".to_string(),
942 retry_after_secs: None,
943 };
944 assert!(is_transient(&err));
945 let err = SyncKitError::Server {
946 status: 504,
947 message: "Gateway Timeout".to_string(),
948 retry_after_secs: None,
949 };
950 assert!(is_transient(&err));
951 }
952
953 #[test]
954 fn is_transient_rate_limited_429() {
955 let err = SyncKitError::Server {
956 status: 429,
957 message: "Too Many Requests".to_string(),
958 retry_after_secs: None,
959 };
960 assert!(is_transient(&err));
961 }
962
963 #[test]
964 fn is_not_transient_client_4xx() {
965 let err = SyncKitError::Server {
966 status: 400,
967 message: "Bad Request".to_string(),
968 retry_after_secs: None,
969 };
970 assert!(!is_transient(&err));
971 let err = SyncKitError::Server {
972 status: 401,
973 message: "Unauthorized".to_string(),
974 retry_after_secs: None,
975 };
976 assert!(!is_transient(&err));
977 let err = SyncKitError::Server {
978 status: 403,
979 message: "Forbidden".to_string(),
980 retry_after_secs: None,
981 };
982 assert!(!is_transient(&err));
983 let err = SyncKitError::Server {
984 status: 404,
985 message: "Not Found".to_string(),
986 retry_after_secs: None,
987 };
988 assert!(!is_transient(&err));
989 let err = SyncKitError::Server {
990 status: 409,
991 message: "Conflict".to_string(),
992 retry_after_secs: None,
993 };
994 assert!(!is_transient(&err));
995 let err = SyncKitError::Server {
996 status: 422,
997 message: "Unprocessable Entity".to_string(),
998 retry_after_secs: None,
999 };
1000 assert!(!is_transient(&err));
1001 }
1002
1003 #[test]
1004 fn is_not_transient_not_authenticated() {
1005 assert!(!is_transient(&SyncKitError::NotAuthenticated));
1006 }
1007
1008 #[test]
1009 fn is_not_transient_no_master_key() {
1010 assert!(!is_transient(&SyncKitError::NoMasterKey));
1011 }
1012
1013 #[test]
1014 fn is_not_transient_decryption_failed() {
1015 assert!(!is_transient(&SyncKitError::DecryptionFailed));
1016 }
1017
1018 #[test]
1019 fn is_not_transient_invalid_envelope() {
1020 assert!(!is_transient(&SyncKitError::InvalidEnvelope(
1021 "bad version".to_string()
1022 )));
1023 }
1024
1025 #[test]
1026 fn is_not_transient_crypto() {
1027 assert!(!is_transient(&SyncKitError::Crypto(
1028 "encrypt failed".to_string()
1029 )));
1030 }
1031
1032 #[test]
1033 fn is_not_transient_json() {
1034 let err: SyncKitError = serde_json::from_str::<serde_json::Value>("not json")
1035 .unwrap_err()
1036 .into();
1037 assert!(!is_transient(&err));
1038 }
1039
1040 #[test]
1041 fn is_not_transient_base64() {
1042 let err: SyncKitError = base64::engine::general_purpose::STANDARD
1043 .decode("!!!invalid!!!")
1044 .unwrap_err()
1045 .into();
1046 assert!(!is_transient(&err));
1047 }
1048
1049 #[test]
1050 fn is_not_transient_token_expired() {
1051 assert!(!is_transient(&SyncKitError::TokenExpired));
1052 }
1053
1054 #[test]
1055 fn is_not_transient_internal() {
1056 assert!(!is_transient(&SyncKitError::Internal(
1057 "lock poisoned".to_string()
1058 )));
1059 }
1060
1061 // ── Retry constants ──
1062
1063 #[test]
1064 fn retry_constants_are_sensible() {
1065 assert_eq!(MAX_RETRIES, 3);
1066 assert_eq!(BASE_DELAY, Duration::from_secs(1));
1067 }
1068
1069 #[test]
1070 fn backoff_delays_are_exponential() {
1071 let delay_0 = BASE_DELAY * 2u32.pow(0);
1072 let delay_1 = BASE_DELAY * 2u32.pow(1);
1073 let delay_2 = BASE_DELAY * 2u32.pow(2);
1074
1075 assert_eq!(delay_0, Duration::from_secs(1));
1076 assert_eq!(delay_1, Duration::from_secs(2));
1077 assert_eq!(delay_2, Duration::from_secs(4));
1078 }
1079
1080 // ── is_transient boundary: 429 vs 428, 499 vs 500 ──
1081
1082 #[test]
1083 fn is_transient_boundary_values() {
1084 assert!(!is_transient(&SyncKitError::Server {
1085 status: 428,
1086 message: String::new(),
1087 retry_after_secs: None
1088 }));
1089 assert!(is_transient(&SyncKitError::Server {
1090 status: 429,
1091 message: String::new(),
1092 retry_after_secs: None
1093 }));
1094 assert!(!is_transient(&SyncKitError::Server {
1095 status: 430,
1096 message: String::new(),
1097 retry_after_secs: None
1098 }));
1099 assert!(!is_transient(&SyncKitError::Server {
1100 status: 499,
1101 message: String::new(),
1102 retry_after_secs: None
1103 }));
1104 assert!(is_transient(&SyncKitError::Server {
1105 status: 500,
1106 message: String::new(),
1107 retry_after_secs: None
1108 }));
1109 }
1110
1111 // ── Token expiry detection ──
1112
1113 #[test]
1114 fn jwt_exp_extracts_expiry() {
1115 let exp = Utc::now().timestamp() + 3600;
1116 let token = fake_jwt(exp);
1117 assert_eq!(jwt_exp(&token), Some(exp));
1118 }
1119
1120 #[test]
1121 fn jwt_exp_returns_none_for_garbage() {
1122 assert_eq!(jwt_exp("not-a-jwt"), None);
1123 assert_eq!(jwt_exp("a.b.c"), None);
1124 assert_eq!(jwt_exp(""), None);
1125 }
1126
1127 #[test]
1128 fn token_is_expired_for_past_exp() {
1129 let token = fake_jwt(Utc::now().timestamp() - 3600);
1130 assert!(token_is_expired(&token));
1131 }
1132
1133 #[test]
1134 fn token_is_expired_within_buffer() {
1135 let token = fake_jwt(Utc::now().timestamp() + 10);
1136 assert!(token_is_expired(&token));
1137 }
1138
1139 #[test]
1140 fn token_is_not_expired_when_fresh() {
1141 let token = fake_jwt(Utc::now().timestamp() + 3600);
1142 assert!(!token_is_expired(&token));
1143 }
1144
1145 #[test]
1146 fn token_is_not_expired_for_garbage() {
1147 assert!(!token_is_expired("garbage"));
1148 }
1149
1150 #[test]
1151 fn token_expires_exactly_at_buffer_boundary() {
1152 let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS);
1153 assert!(token_is_expired(&token));
1154 }
1155
1156 #[test]
1157 fn token_expires_just_past_buffer() {
1158 let token = fake_jwt(Utc::now().timestamp() + TOKEN_EXPIRY_BUFFER_SECS + 1);
1159 assert!(!token_is_expired(&token));
1160 }
1161
1162 // ── encrypt_change preserves metadata ──
1163
1164 #[test]
1165 fn encrypt_change_preserves_all_metadata() {
1166 let client = SyncKitClient::new(test_config());
1167 let key = crypto::generate_master_key();
1168 client.set_master_key_raw(key);
1169
1170 let ts = Utc::now();
1171 let entry = ChangeEntry {
1172 table: "contacts".to_string(),
1173 op: ChangeOp::Update,
1174 row_id: "unique-row-id".to_string(),
1175 timestamp: ts,
1176 hlc: Hlc::zero(DeviceId::nil()),
1177 data: Some(serde_json::json!({"name": "Alice"})),
1178 extra: serde_json::Map::default(),
1179 };
1180
1181 let wire = client.encrypt_change(entry).unwrap();
1182 assert_eq!(wire.table, "contacts");
1183 assert_eq!(wire.op, ChangeOp::Update);
1184 assert_eq!(wire.row_id, "unique-row-id");
1185 assert_eq!(wire.timestamp, ts);
1186 }
1187
1188 // ── Multiple entries encrypt/decrypt ──
1189
1190 #[test]
1191 fn multiple_entries_encrypt_decrypt_roundtrip() {
1192 let client = SyncKitClient::new(test_config());
1193 let key = crypto::generate_master_key();
1194 client.set_master_key_raw(key);
1195
1196 let entries = [
1197 ChangeEntry {
1198 table: "tasks".to_string(),
1199 op: ChangeOp::Insert,
1200 row_id: "r1".to_string(),
1201 timestamp: Utc::now(),
1202 hlc: Hlc::zero(DeviceId::nil()),
1203 data: Some(serde_json::json!({"title": "Task 1"})),
1204 extra: serde_json::Map::default(),
1205 },
1206 ChangeEntry {
1207 table: "tasks".to_string(),
1208 op: ChangeOp::Update,
1209 row_id: "r2".to_string(),
1210 timestamp: Utc::now(),
1211 hlc: Hlc::zero(DeviceId::nil()),
1212 data: Some(serde_json::json!({"title": "Task 2", "done": true})),
1213 extra: serde_json::Map::default(),
1214 },
1215 ChangeEntry {
1216 table: "events".to_string(),
1217 op: ChangeOp::Delete,
1218 row_id: "r3".to_string(),
1219 timestamp: Utc::now(),
1220 hlc: Hlc::zero(DeviceId::nil()),
1221 data: None,
1222 extra: serde_json::Map::default(),
1223 },
1224 ];
1225
1226 let wire_entries: Vec<_> = entries
1227 .iter()
1228 .cloned()
1229 .map(|e| client.encrypt_change(e).unwrap())
1230 .collect();
1231
1232 assert_eq!(wire_entries.len(), 3);
1233 assert!(wire_entries[0].data.is_some());
1234 assert!(wire_entries[1].data.is_some());
1235 // The Delete now also seals an encrypted HLC envelope (was None before HLC).
1236 assert!(wire_entries[2].data.is_some());
1237
1238 for (i, wire) in wire_entries.into_iter().enumerate() {
1239 let pull = PullChangeEntry {
1240 seq: i as i64,
1241 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
1242 table: wire.table,
1243 op: wire.op,
1244 row_id: wire.row_id,
1245 timestamp: wire.timestamp,
1246 data: wire.data,
1247 key_id: None,
1248 };
1249 let decrypted = client.decrypt_change(pull).unwrap();
1250 assert_eq!(decrypted.table, entries[i].table);
1251 assert_eq!(decrypted.op, entries[i].op);
1252 assert_eq!(decrypted.data, entries[i].data);
1253 }
1254 }
1255
1256 // ── Unicode and edge-case roundtrips ──
1257
1258 #[test]
1259 fn encrypt_decrypt_roundtrip_unicode_table() {
1260 let client = SyncKitClient::new(test_config());
1261 let key = crypto::generate_master_key();
1262 client.set_master_key_raw(key);
1263
1264 let entry = ChangeEntry {
1265 table: "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}".into(),
1266 op: ChangeOp::Insert,
1267 row_id: "row-1".into(),
1268 timestamp: Utc::now(),
1269 hlc: Hlc::zero(DeviceId::nil()),
1270 data: Some(serde_json::json!({"name": "\u{30C6}\u{30B9}\u{30C8}"})),
1271 extra: serde_json::Map::default(),
1272 };
1273
1274 let wire = client.encrypt_change(entry).unwrap();
1275 let pull = PullChangeEntry {
1276 seq: 1,
1277 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
1278 table: wire.table,
1279 op: wire.op,
1280 row_id: wire.row_id,
1281 timestamp: wire.timestamp,
1282 data: wire.data,
1283 key_id: None,
1284 };
1285 let decrypted = client.decrypt_change(pull).unwrap();
1286 assert_eq!(
1287 decrypted.table,
1288 "\u{65E5}\u{672C}\u{8A9E}\u{30C6}\u{30FC}\u{30D6}\u{30EB}"
1289 );
1290 }
1291
1292 #[test]
1293 fn encrypt_decrypt_roundtrip_empty_row_id() {
1294 let client = SyncKitClient::new(test_config());
1295 let key = crypto::generate_master_key();
1296 client.set_master_key_raw(key);
1297
1298 let entry = ChangeEntry {
1299 table: "t".into(),
1300 op: ChangeOp::Insert,
1301 row_id: String::new(),
1302 timestamp: Utc::now(),
1303 hlc: Hlc::zero(DeviceId::nil()),
1304 data: Some(serde_json::json!(42)),
1305 extra: serde_json::Map::default(),
1306 };
1307
1308 let wire = client.encrypt_change(entry).unwrap();
1309 let pull = PullChangeEntry {
1310 seq: 1,
1311 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
1312 table: wire.table,
1313 op: wire.op,
1314 row_id: wire.row_id,
1315 timestamp: wire.timestamp,
1316 data: wire.data,
1317 key_id: None,
1318 };
1319 let decrypted = client.decrypt_change(pull).unwrap();
1320 assert_eq!(decrypted.row_id, "");
1321 assert_eq!(decrypted.data.unwrap(), serde_json::json!(42));
1322 }
1323
1324 // ── decrypt_change_multi_key key selection ──
1325 //
1326 // Pins the key-id boundary logic that picks between primary and pending
1327 // master keys during a rotation window. The mutations targeted here:
1328 // * `effective_key_id == pending_key_id` (== ↔ !=)
1329 // * `effective_key_id == primary_key_id || effective_key_id <= 1`
1330 // (`||` ↔ `&&`, `<= 1` ↔ `< 1`/`<= 0`)
1331 // * `entry.key_id.unwrap_or(1)` default
1332 // * the Err-fallthrough that tries pending if primary fails
1333
1334 fn encrypt_with(key: &[u8; 32], value: &serde_json::Value) -> serde_json::Value {
1335 crypto::encrypt_json(value, key).unwrap()
1336 }
1337
1338 fn pull_entry_with(encrypted: serde_json::Value, key_id: Option<i32>) -> PullChangeEntry {
1339 PullChangeEntry {
1340 seq: 1,
1341 device_id: crate::ids::DeviceId::new(uuid::Uuid::new_v4()),
1342 table: "tasks".to_string(),
1343 op: ChangeOp::Insert,
1344 row_id: "row-multikey".to_string(),
1345 timestamp: Utc::now(),
1346 data: Some(encrypted),
1347 key_id,
1348 }
1349 }
1350
1351 #[test]
1352 fn multi_key_picks_pending_when_key_id_matches() {
1353 let primary = crypto::generate_master_key();
1354 let pending = crypto::generate_master_key();
1355 let plaintext = serde_json::json!({"v": "pending-payload"});
1356 let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(7));
1357
1358 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
1359 entry,
1360 &primary,
1361 1,
1362 &pending,
1363 7,
1364 &SyncKitClient::decrypt_change_with_key,
1365 )
1366 .unwrap();
1367 assert_eq!(decrypted.data.unwrap(), plaintext);
1368 }
1369
1370 #[test]
1371 fn multi_key_picks_primary_when_key_id_matches_primary() {
1372 let primary = crypto::generate_master_key();
1373 let pending = crypto::generate_master_key();
1374 let plaintext = serde_json::json!({"v": "primary-payload"});
1375 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(3));
1376
1377 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
1378 entry,
1379 &primary,
1380 3,
1381 &pending,
1382 9,
1383 &SyncKitClient::decrypt_change_with_key,
1384 )
1385 .unwrap();
1386 assert_eq!(decrypted.data.unwrap(), plaintext);
1387 }
1388
1389 #[test]
1390 fn multi_key_treats_missing_key_id_as_primary() {
1391 // `entry.key_id.unwrap_or(1)` defaults to 1; `effective_key_id <= 1`
1392 // arm routes to primary. A mutation changing `unwrap_or(1)` to
1393 // `unwrap_or(99)` would route to "unknown" path and try primary anyway
1394 // via the fallback, but a mutation to `<= 1` → `< 1` would skip the
1395 // direct-primary branch and fall into the fallback path.
1396 let primary = crypto::generate_master_key();
1397 let pending = crypto::generate_master_key();
1398 let plaintext = serde_json::json!({"v": "legacy-no-key-id"});
1399 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), None);
1400
1401 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
1402 entry,
1403 &primary,
1404 5,
1405 &pending,
1406 6,
1407 &SyncKitClient::decrypt_change_with_key,
1408 )
1409 .unwrap();
1410 assert_eq!(decrypted.data.unwrap(), plaintext);
1411 }
1412
1413 #[test]
1414 fn multi_key_unknown_key_id_falls_back_to_primary_first() {
1415 // effective_key_id (42) matches neither primary (5) nor pending (6),
1416 // and is > 1. The fallback first tries primary; here the data WAS
1417 // encrypted with primary, so the fallback succeeds.
1418 let primary = crypto::generate_master_key();
1419 let pending = crypto::generate_master_key();
1420 let plaintext = serde_json::json!({"v": "via-fallback-primary"});
1421 let entry = pull_entry_with(encrypt_with(&primary, &plaintext), Some(42));
1422
1423 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
1424 entry,
1425 &primary,
1426 5,
1427 &pending,
1428 6,
1429 &SyncKitClient::decrypt_change_with_key,
1430 )
1431 .unwrap();
1432 assert_eq!(decrypted.data.unwrap(), plaintext);
1433 }
1434
1435 #[test]
1436 fn multi_key_unknown_key_id_falls_back_to_pending_when_primary_fails() {
1437 // Unknown key_id and the data was encrypted with pending, fallback
1438 // must try primary first (fails), then pending (succeeds).
1439 let primary = crypto::generate_master_key();
1440 let pending = crypto::generate_master_key();
1441 let plaintext = serde_json::json!({"v": "via-fallback-pending"});
1442 let entry = pull_entry_with(encrypt_with(&pending, &plaintext), Some(42));
1443
1444 let decrypted = SyncKitClient::decrypt_with_rotation_keys(
1445 entry,
1446 &primary,
1447 5,
1448 &pending,
1449 6,
1450 &SyncKitClient::decrypt_change_with_key,
1451 )
1452 .unwrap();
1453 assert_eq!(decrypted.data.unwrap(), plaintext);
1454 }
1455 }
1456