Skip to main content

max / synckit

synckit-client: kill the survivors the last test pass left standing Nine findings from the 2026-08-04 mutation run were re-verified against current code today. All nine had real tests written for them since, and all nine still had named mutants alive. This is that residue: 40 tests across ten files, both feature configurations green (626 lib + 160 integration default, 624 + 165 under --no-default-features --features store,testing), no production code changed. The class is worth naming because every one of these tests was aimed correctly and passed anyway. The clearest case is enforce_changelog_retention, whose test ended `assert!(dropped >= 1)` where the true value is 4, so the mutant returning a constant Ok(1) sailed through a test written for it. The rest are the same shape: a bound asserted away from its boundary, a conjunction exercised only on the row where both operators agree, a persistence call whose only test stamped values a zeroed store would also produce. So the tests here are chosen to disagree. hlc save_clock and observe are pinned by reopening through load_clock at a high now_ms, since the previous fixture stamped at 50 and 200 where a no-op save is invisible. Hlc::observe's conjunction is exercised on both rows where && and || differ rather than one. The Argon2 memory floor is pinned by the REJECTION at 8191 rather than the acceptance at 8192, because the mutant's floor still accepts the latter. apply's composite-primary-key case needed a new fixture table, because a single-column PK genuinely cannot see the index increment the mutation deletes. Cargo.toml gains tokio's test-util dev-feature. The retry backoff guards gate only a sleep, so no assertion on request counts can observe them however many attempts it counts; a paused clock and an assertion on elapsed time can, and that is what the six guards at helpers.rs 199, 250 and 269 needed. Two dispositions deliberately not taken. The rotation.rs:124 mutant is NOT recorded as equivalent: it is equivalent only under the no-keychain config the tests run in, and is a live defect in the shipping build, filed as its own task. The blob.rs:172 size guard stays a visible survivor rather than an exclusion, since distinguishing it needs a real 4 GiB allocation, which is cost rather than impossibility.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session
https://claude.ai/code/session_01MptwXZ8k65v19rFmdGAyki
Author: Max Johnson <me@maxj.phd> · 2026-08-31 16:59 UTC
Signed with PGP, not checked
Commit: d85f75506a17f4955c63dd85a5200294c9f34dd4
Parent: 5876a22
14 files changed, +1829 insertions, -2 deletions
M synckit-client/Cargo.toml +5
@@ -115,6 +115,11 @@
115 115 path = "tests/integration/main.rs"
116 116
117 117 [dev-dependencies]
118 + # `test-util` for the paused clock. The retry loop's backoff guard gates only a
119 + # `tokio::time::sleep`, so nothing but elapsed time can observe it, and freezing
120 + # the clock is what lets a test assert the delays without waiting seven real
121 + # seconds. Resolver v3 keeps dev-dependency features out of a consumer build.
122 + tokio = { version = "1", features = ["test-util"] }
118 123 # The contracts in `types::Hlc`, `conflict`, and `crypto` are universally
119 124 # quantified (an order is an order for every pair, a round-trip round-trips for
120 125 # every input), while their tests were all examples. `proptest-regressions/` is
M synckit-client/src/conflict.rs +92
@@ -1144,6 +1144,13 @@
1144 1144 ];
1145 1145
1146 1146 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
1147 + // The negative side of `is_empty`: every other assertion in this file is
1148 + // `assert!(clean.is_empty())`, which a constant-`true` `is_empty` also
1149 + // satisfies. Two clean changes must report non-empty.
1150 + assert!(
1151 + !clean.is_empty(),
1152 + "two uncontested rows must not report an empty CleanChanges"
1153 + );
1147 1154 assert_eq!(clean.len(), 2);
1148 1155 assert_eq!(conflicts.len(), 1);
1149 1156 assert_eq!(conflicts[0].remote.entry.row_id, "r1");
@@ -2092,6 +2099,91 @@
2092 2099 ));
2093 2100 }
2094 2101
2102 + /// The drift window is a stated contract (5 minutes), not an arbitrary
2103 + /// number: it is the honest inter-device skew SyncKit promises to tolerate,
2104 + /// and every other test in the tree refers to it symbolically, so an
2105 + /// arithmetic slip in `5 * 60 * 1000` would change no other outcome.
2106 + #[test]
2107 + fn max_hlc_drift_is_five_minutes() {
2108 + assert_eq!(MAX_HLC_DRIFT_MS, 300_000);
2109 + }
2110 +
2111 + /// The poisoning threshold is exclusive: an entry sitting exactly on
2112 + /// `now + MAX_HLC_DRIFT_MS` is still honest, and only the next millisecond
2113 + /// is poisoned. Both sides of the boundary, because `>` and `>=` differ
2114 + /// only at equality.
2115 + #[test]
2116 + fn poisoning_threshold_is_exclusive_at_the_drift_boundary() {
2117 + let now = Utc::now();
2118 + let now_ms = now.timestamp_millis();
2119 +
2120 + let at_boundary = Hlc {
2121 + wall_ms: now_ms + MAX_HLC_DRIFT_MS,
2122 + counter: 0,
2123 + node: remote_node(),
2124 + };
2125 + assert!(
2126 + !is_clock_poisoned(&at_boundary, now),
2127 + "an entry exactly at the drift limit is within tolerated skew"
2128 + );
2129 +
2130 + let one_past = Hlc {
2131 + wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
2132 + counter: 0,
2133 + node: remote_node(),
2134 + };
2135 + assert!(
2136 + is_clock_poisoned(&one_past, now),
2137 + "one millisecond past the drift limit is poisoned"
2138 + );
2139 +
2140 + let one_before = Hlc {
2141 + wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1,
2142 + counter: 0,
2143 + node: remote_node(),
2144 + };
2145 + assert!(!is_clock_poisoned(&one_before, now));
2146 + }
2147 +
2148 + /// The boundary as `resolve_lww_at` sees it: a remote sitting exactly on the
2149 + /// limit is honest, so it wins on raw HLC order; one millisecond further out
2150 + /// loses to the older local. Same two inputs, opposite resolutions, so the
2151 + /// threshold cannot be shifted by a millisecond without this failing.
2152 + #[test]
2153 + fn lww_keeps_a_remote_at_the_drift_boundary_and_rejects_the_next_ms() {
2154 + let now = Utc::now();
2155 + let now_ms = now.timestamp_millis();
2156 +
2157 + let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
2158 + local.hlc = Hlc {
2159 + wall_ms: now_ms,
2160 + counter: 0,
2161 + node: local_node(),
2162 + };
2163 +
2164 + let mut at_boundary = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
2165 + at_boundary.entry.hlc = Hlc {
2166 + wall_ms: now_ms + MAX_HLC_DRIFT_MS,
2167 + counter: 0,
2168 + node: remote_node(),
2169 + };
2170 + assert!(matches!(
2171 + resolve_lww_at(&local, &at_boundary, now),
2172 + Resolution::KeepRemote
2173 + ));
2174 +
2175 + let mut one_past = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
2176 + one_past.entry.hlc = Hlc {
2177 + wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
2178 + counter: 0,
2179 + node: remote_node(),
2180 + };
2181 + assert!(matches!(
2182 + resolve_lww_at(&local, &one_past, now),
2183 + Resolution::KeepLocal
2184 + ));
2185 + }
2186 +
2095 2187 #[test]
2096 2188 fn canonical_payload_sorts_map_keys() {
2097 2189 // Pins the exact-HLC tiebreak's convergence invariant: serde_json must
M synckit-client/src/crypto.rs +79
@@ -2237,4 +2237,83 @@
2237 2237 assert!(encrypt_data_aad(b"x", &gck, &ctx).is_err());
2238 2238 }
2239 2239 }
2240 +
2241 + /// The `total_len > encrypted.len()` bound is a strict inequality, and the
2242 + /// equality case has to reach the chunk stream. A header claiming exactly
2243 + /// the ciphertext length is not an over-claim on the allocation bound (it
2244 + /// allocates no more than the input already occupies), so it must be
2245 + /// rejected further down, where the sealed-chunk arithmetic finds the
2246 + /// stream too short. Relaxing the bound to `>=` would short-circuit here
2247 + /// and report the wrong reason, which is what this pins.
2248 + #[test]
2249 + fn v3_total_len_equal_to_input_is_rejected_by_the_chunk_stream_not_the_bound() {
2250 + let key = generate_master_key();
2251 + let plaintext = vec![9u8; 100];
2252 + let mut sealed = encrypt_blob_chunked(&plaintext, &key, "hash").unwrap();
2253 + // 100 plaintext + tag(4) + header(13) + nonce/tag(40) = 157.
2254 + assert_eq!(sealed.len(), 157, "v3 layout for a single 100-byte chunk");
2255 +
2256 + // total_len lives at body[5..13], i.e. absolute [9..17].
2257 + let equal = sealed.len() as u64;
2258 + sealed[9..17].copy_from_slice(&equal.to_le_bytes());
2259 +
2260 + let err = decrypt_blob_chunked(&sealed, &key, "hash")
2261 + .expect_err("a 157-byte plaintext claim cannot be satisfied by 140 stream bytes");
2262 + let SyncKitError::Crypto(msg) = &err else {
2263 + panic!("expected a crypto error, got {err:?}");
2264 + };
2265 + assert!(
2266 + msg.contains("truncated mid-chunk"),
2267 + "equality must fall through the length bound and fail on the stream: {msg}"
2268 + );
2269 + assert!(
2270 + !msg.contains("exceeds encrypted input"),
2271 + "total_len == encrypted.len() is not an over-claim: {msg}"
2272 + );
2273 +
2274 + // One byte past the bound is the over-claim, and must be caught here.
2275 + let over = sealed.len() as u64 + 1;
2276 + sealed[9..17].copy_from_slice(&over.to_le_bytes());
2277 + let err = decrypt_blob_chunked(&sealed, &key, "hash")
2278 + .expect_err("a claim larger than the input is rejected");
2279 + let SyncKitError::Crypto(msg) = &err else {
2280 + panic!("expected a crypto error, got {err:?}");
2281 + };
2282 + assert!(
2283 + msg.contains("exceeds encrypted input"),
2284 + "one byte over the input length is the allocation bound: {msg}"
2285 + );
2286 + }
2287 +
2288 + /// v2 wire overhead is a sum of the tag and the AEAD overhead, pinned to
2289 + /// the number rather than recomputed from the same expression: an assertion
2290 + /// written as `WIRE_V2_TAG_BYTES.len() + ENCRYPTION_OVERHEAD` would hold
2291 + /// however that expression was mutated.
2292 + #[test]
2293 + fn v2_wire_overhead_is_forty_four_bytes() {
2294 + assert_eq!(ENCRYPTION_OVERHEAD_V2, 44);
2295 + // And the constant describes what the encoder actually emits.
2296 + let key = generate_master_key();
2297 + let ctx = AeadContext::blob("abc");
2298 + let wire = encrypt_bytes_aad(&[3u8; 70], &key, &ctx).unwrap();
2299 + assert_eq!(wire.len(), 70 + 44);
2300 + }
2301 +
2302 + /// The Argon2 memory floor is 8192 KiB exactly. Both sides of the bound are
2303 + /// asserted: 8191 is a downgrade and must be refused, 8192 must derive.
2304 + /// The pre-existing test used 8 KiB, which any plausible floor rejects.
2305 + #[test]
2306 + fn argon2_memory_floor_admits_8192_kib_and_refuses_8191() {
2307 + let salt = [7u8; 32];
2308 + let err = derive_wrapping_key_with_params("pw", &salt, 8191, 1, 1)
2309 + .expect_err("one KiB under the floor is a KDF downgrade");
2310 + assert!(
2311 + matches!(err, SyncKitError::InvalidEnvelope(_)),
2312 + "out-of-range params are an envelope error, got {err:?}"
2313 + );
2314 + assert!(
2315 + derive_wrapping_key_with_params("pw", &salt, 8192, 1, 1).is_ok(),
2316 + "the floor itself is inside the accepted range"
2317 + );
2318 + }
2240 2319 }
M synckit-client/src/types.rs +62
@@ -1359,6 +1359,68 @@
1359 1359 );
1360 1360 }
1361 1361
1362 + /// The counter carried forward has to come from the clock that actually
1363 + /// owns the winning wall component. Here the local clock wins the wall
1364 + /// (500 > 200 and > now_ms), so the remote's much larger counter at its own
1365 + /// older wall is stale and must be discarded. Merging the two counters
1366 + /// instead (taking the max whenever either wall matches) would inflate the
1367 + /// counter to 41 and burn 37 tiebreak slots that no event ever used.
1368 + #[test]
1369 + fn hlc_observe_discards_a_stale_remote_counter() {
1370 + let me = DeviceId::new(Uuid::from_u128(1));
1371 + let them = DeviceId::new(Uuid::from_u128(2));
1372 + let local = Hlc {
1373 + wall_ms: 500,
1374 + counter: 3,
1375 + node: me,
1376 + };
1377 + let remote = Hlc {
1378 + wall_ms: 200,
1379 + counter: 40,
1380 + node: them,
1381 + };
1382 + let merged = Hlc::observe(local, remote, 100, me);
1383 + assert_eq!(
1384 + merged,
1385 + Hlc {
1386 + wall_ms: 500,
1387 + counter: 4,
1388 + node: me
1389 + },
1390 + "the remote counter belongs to wall 200 and must not carry to 500"
1391 + );
1392 + }
1393 +
1394 + /// The mirror image: the remote wins the wall component (800), so our own
1395 + /// counter at wall 100 is the stale one. The result must continue the
1396 + /// remote's counter, not ours, or the merged clock claims 51 events at a
1397 + /// wall component where only 8 exist.
1398 + #[test]
1399 + fn hlc_observe_discards_a_stale_local_counter() {
1400 + let me = DeviceId::new(Uuid::from_u128(1));
1401 + let them = DeviceId::new(Uuid::from_u128(2));
1402 + let local = Hlc {
1403 + wall_ms: 100,
1404 + counter: 50,
1405 + node: me,
1406 + };
1407 + let remote = Hlc {
1408 + wall_ms: 800,
1409 + counter: 7,
1410 + node: them,
1411 + };
1412 + let merged = Hlc::observe(local, remote, 200, me);
1413 + assert_eq!(
1414 + merged,
1415 + Hlc {
1416 + wall_ms: 800,
1417 + counter: 8,
1418 + node: me
1419 + },
1420 + "our counter belongs to wall 100 and must not carry to 800"
1421 + );
1422 + }
1423 +
1362 1424 #[test]
1363 1425 fn hlc_counter_overflow_rolls_into_wall_not_backwards() {
1364 1426 let me = DeviceId::new(Uuid::from_u128(1));
M synckit-client/src/client/blob.rs +142
@@ -971,6 +971,7 @@
971 971
972 972 #[cfg(test)]
973 973 mod tests {
974 + use super::*;
974 975 use crate::types::*;
975 976
976 977 mod resume {
@@ -1069,6 +1070,40 @@
1069 1070 assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
1070 1071 }
1071 1072
1073 + #[test]
1074 + fn a_plan_with_a_zero_part_size_is_dropped() {
1075 + // part_size 0 is the hostile case the `> 0` guard exists for: it is
1076 + // also the divisor of the tiling check below it, so a guard that let
1077 + // it through would divide by zero rather than merely mis-resume.
1078 + let store = fake(60);
1079 + store
1080 + .record
1081 + .lock()
1082 + .unwrap()
1083 + .as_mut()
1084 + .unwrap()
1085 + .session
1086 + .part_size = 0;
1087 + assert!(SyncKitClient::load_resume(&store, "h", 24).is_none());
1088 + assert!(*store.cleared.lock().unwrap());
1089 + }
1090 +
1091 + #[test]
1092 + fn a_plan_with_no_parts_is_dropped_even_where_the_tiling_check_would_agree() {
1093 + // part_count 0 over a 0-byte session: 0.div_ceil(8) == 0, so the
1094 + // tiling check is satisfied and the `part_count > 0` guard is the
1095 + // only thing rejecting it. A record naming a completed part in a
1096 + // zero-part plan describes nothing.
1097 + let store = fake(60);
1098 + {
1099 + let mut held = store.record.lock().unwrap();
1100 + let session = &mut held.as_mut().unwrap().session;
1101 + session.part_count = 0;
1102 + session.size_bytes = 0;
1103 + }
1104 + assert!(SyncKitClient::load_resume(&store, "h", 0).is_none());
1105 + }
1106 +
1072 1107 #[test]
1073 1108 fn a_record_with_no_completed_parts_saves_nothing() {
1074 1109 // Not an error: the upload starts at part 1 either way. Dropping it
@@ -1243,4 +1278,111 @@
1243 1278 assert_eq!(parsed["hash"], "sha256-def456");
1244 1279 assert_eq!(parsed["size_bytes"], 2048);
1245 1280 }
1281 +
1282 + // ── The in-memory and streaming size caps ──
1283 +
1284 + #[test]
1285 + fn the_blob_cap_is_four_gibibytes_exactly() {
1286 + // Pinned as a literal rather than as the same arithmetic the constant
1287 + // uses, because that arithmetic is what can be wrong. The two readings a
1288 + // single wrong operator produces here are 1_077_936_128 and 4_195_328:
1289 + // both look like plausible caps, and either would refuse legitimate
1290 + // media the SDK documents itself as carrying. Nothing else in the suite
1291 + // can see the difference, since no fixture is anywhere near any of the
1292 + // three values.
1293 + assert_eq!(MAX_BLOB_BYTES, 4_294_967_296, "4 GiB");
1294 + }
1295 +
1296 + /// A client holding a key but no session: every blob path gets past the
1297 + /// key check and stops at `require_token`, which is what makes
1298 + /// `NotAuthenticated` mean "the size check let this through".
1299 + fn keyed_but_unauthenticated() -> SyncKitClient {
1300 + let client = SyncKitClient::new(crate::SyncKitConfig {
1301 + server_url: "https://example.invalid".to_string(),
1302 + api_key: "test-api-key".to_string(),
1303 + });
1304 + client.set_master_key_raw([9u8; 32]);
1305 + client
1306 + }
1307 +
1308 + /// A sparse file of `len` bytes: `set_len` allocates nothing, so the
1309 + /// multi-gigabyte sizes the cap is written in terms of cost no disk. The
1310 + /// cap is read off `metadata`, which is all these tests reach.
1311 + fn sparse_file(len: u64) -> std::path::PathBuf {
1312 + use std::sync::atomic::{AtomicU64, Ordering};
1313 + static N: AtomicU64 = AtomicU64::new(0);
1314 + let mut p = std::env::temp_dir();
1315 + p.push(format!(
1316 + "synckit_cap_{}_{}",
1317 + std::process::id(),
1318 + N.fetch_add(1, Ordering::Relaxed)
1319 + ));
1320 + let f = std::fs::File::create(&p).unwrap();
1321 + f.set_len(len).unwrap();
1322 + p
1323 + }
1324 +
1325 + #[tokio::test]
1326 + async fn the_streaming_cap_accepts_a_blob_of_exactly_the_cap_and_refuses_one_byte_more() {
1327 + // Both sides of the bound. `>` differs from `>=` and from `==` only at
1328 + // the cap itself, so a test that only uploads something small cannot
1329 + // see any of them: at every reachable size all three agree.
1330 + let client = keyed_but_unauthenticated();
1331 + let hash = "b".repeat(64);
1332 +
1333 + let at_cap = sparse_file(MAX_BLOB_BYTES as u64);
1334 + let err = client
1335 + .blob_upload_streaming(&hash, &at_cap)
1336 + .await
1337 + .unwrap_err();
1338 + assert!(
1339 + matches!(err, SyncKitError::NotAuthenticated),
1340 + "a blob of exactly {MAX_BLOB_BYTES} bytes is under the cap and must reach the session check, got {err:?}"
1341 + );
1342 +
1343 + let over = sparse_file(MAX_BLOB_BYTES as u64 + 1);
1344 + let err = client
1345 + .blob_upload_streaming(&hash, &over)
1346 + .await
1347 + .unwrap_err();
1348 + match err {
1349 + SyncKitError::InvalidArgument(m) => {
1350 + assert!(m.contains("client cap"), "wrong rejection: {m}");
1351 + }
1352 + other => panic!("one byte over the cap must be refused, got {other:?}"),
1353 + }
1354 +
1355 + // And a small file is not refused by a cap that has been inverted.
1356 + let small = sparse_file(1_000);
1357 + let err = client
1358 + .blob_upload_streaming(&hash, &small)
1359 + .await
1360 + .unwrap_err();
1361 + assert!(
1362 + matches!(err, SyncKitError::NotAuthenticated),
1363 + "a 1000-byte blob must reach the session check, got {err:?}"
1364 + );
1365 +
1366 + for p in [at_cap, over, small] {
1367 + let _ = std::fs::remove_file(p);
1368 + }
1369 + }
1370 +
1371 + #[tokio::test]
1372 + async fn the_in_memory_cap_passes_an_ordinary_blob_through_to_the_put() {
1373 + // The reachable half of the same bound: a blob far under the cap must
1374 + // not be rejected by it, so the call fails at the transport instead.
1375 + // (The unreachable half is the cap itself, which would need a 4 GiB
1376 + // allocation to reach.) A relative URL is a reqwest builder error,
1377 + // classified as permanent, so no network attempt is made.
1378 + let client = keyed_but_unauthenticated();
1379 + let err = client
1380 + .blob_upload(&"c".repeat(64), "not-a-url", vec![7u8; 5_000])
1381 + .await
1382 + .unwrap_err();
1383 + assert!(
1384 + matches!(err, SyncKitError::Http(_)),
1385 + "a 5000-byte blob is under the cap and must reach the PUT, got {err:?}"
1386 + );
1387 + }
1246 1388 }
M synckit-client/src/client/helpers.rs +222
@@ -1900,4 +1900,226 @@
1900 1900 .unwrap_err();
1901 1901 assert!(matches!(err, SyncKitError::Internal(_)));
1902 1902 }
1903 +
1904 + // ── the retry loop's sleep, observed on a frozen clock ──
1905 + //
1906 + // Every other retry test counts requests, and a request count cannot see
1907 + // `if attempt < max_attempts`: that guard gates only `tokio::time::sleep`,
1908 + // so flipping it to `>` (never sleep) or `<=` (sleep once more, after the
1909 + // final attempt) leaves the number of attempts untouched. These tests stamp
1910 + // the paused clock inside the operation itself, which makes each gap
1911 + // between attempts exactly the delay that was slept.
1912 +
1913 + /// The gaps between consecutive clock stamps.
1914 + fn gaps(stamps: &[tokio::time::Instant]) -> Vec<Duration> {
1915 + stamps.windows(2).map(|w| w[1] - w[0]).collect()
1916 + }
1917 +
1918 + /// Assert a slept gap lands inside `jittered`'s +/-20% window for `base_ms`.
1919 + /// The windows for 1s, 2s and 4s do not overlap, so a gap identifies which
1920 + /// backoff step produced it.
1921 + #[track_caller]
1922 + fn assert_slept(got: Duration, base_ms: u64) {
1923 + let low = Duration::from_millis(base_ms - base_ms / 5);
1924 + let high = Duration::from_millis(base_ms + base_ms / 5);
1925 + assert!(
1926 + got >= low && got <= high,
1927 + "slept {got:?}, outside the +/-20% window [{low:?}, {high:?}] around {base_ms}ms"
1928 + );
1929 + }
1930 +
1931 + #[tokio::test(start_paused = true)]
1932 + async fn retry_request_sleeps_before_each_replay_and_not_after_the_last() {
1933 + let client = SyncKitClient::new(test_config());
1934 + let stamps = std::sync::Mutex::new(Vec::new());
1935 + let start = tokio::time::Instant::now();
1936 +
1937 + let err = client
1938 + .retry_request(Idempotency::ReadOnly, || {
1939 + stamps.lock().unwrap().push(tokio::time::Instant::now());
1940 + async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1941 + })
1942 + .await
1943 + .unwrap_err();
1944 + let end = tokio::time::Instant::now();
1945 +
1946 + assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
1947 + let stamps = stamps.into_inner().unwrap();
1948 + assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
1949 +
1950 + // Nothing is slept before the first try.
1951 + assert_eq!(stamps[0] - start, Duration::ZERO);
1952 + let gaps = gaps(&stamps);
1953 + assert_slept(gaps[0], 1_000);
1954 + assert_slept(gaps[1], 2_000);
1955 + assert_slept(gaps[2], 4_000);
1956 + // And nothing after the last: the loop gives up the instant the final
1957 + // attempt fails. A guard that fired on `attempt == max_attempts` would
1958 + // burn an 8s backoff here for no replay.
1959 + assert_eq!(
1960 + end - stamps[3],
1961 + Duration::ZERO,
1962 + "slept after the final attempt, which buys nothing"
1963 + );
1964 + }
1965 +
1966 + #[tokio::test(start_paused = true)]
1967 + async fn retry_request_neither_replays_nor_sleeps_an_unsafe_operation() {
1968 + let client = SyncKitClient::new(test_config());
1969 + let calls = std::sync::Mutex::new(0u32);
1970 + let start = tokio::time::Instant::now();
1971 +
1972 + let err = client
1973 + .retry_request(Idempotency::Unsafe, || {
1974 + *calls.lock().unwrap() += 1;
1975 + async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1976 + })
1977 + .await
1978 + .unwrap_err();
1979 +
1980 + assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
1981 + assert_eq!(*calls.lock().unwrap(), 1, "Unsafe gets exactly one attempt");
1982 + // max_attempts is 0 here, so the sleep guard must not fire even once:
1983 + // the caller is told the request failed with no delay bought for a
1984 + // replay that is never going to happen.
1985 + assert_eq!(tokio::time::Instant::now() - start, Duration::ZERO);
1986 + }
1987 +
1988 + #[tokio::test(start_paused = true)]
1989 + async fn retry_request_json_sleeps_before_each_replay_and_not_after_the_last() {
1990 + let client = SyncKitClient::new(test_config());
1991 + let stamps = std::sync::Mutex::new(Vec::new());
1992 + let start = tokio::time::Instant::now();
1993 +
1994 + let err = client
1995 + .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
1996 + stamps.lock().unwrap().push(tokio::time::Instant::now());
1997 + async { Result::<reqwest::Response>::Err(server_err(503, None)) }
1998 + })
1999 + .await
2000 + .unwrap_err();
2001 + let end = tokio::time::Instant::now();
2002 +
2003 + assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
2004 + let stamps = stamps.into_inner().unwrap();
2005 + assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
2006 + assert_eq!(stamps[0] - start, Duration::ZERO);
2007 + let gaps = gaps(&stamps);
2008 + assert_slept(gaps[0], 1_000);
2009 + assert_slept(gaps[1], 2_000);
2010 + assert_slept(gaps[2], 4_000);
2011 + assert_eq!(
2012 + end - stamps[3],
2013 + Duration::ZERO,
2014 + "slept after the final attempt, which buys nothing"
2015 + );
2016 + }
2017 +
2018 + #[tokio::test]
2019 + async fn retry_request_json_sleeps_between_body_read_failures() {
2020 + // The body-read arm has its own copy of the guard, reached only when
2021 + // the request succeeds and the parse does not.
2022 + ensure_crypto_provider();
2023 + let server = wiremock::MockServer::start().await;
2024 + wiremock::Mock::given(wiremock::matchers::any())
2025 + .respond_with(wiremock::ResponseTemplate::new(200))
2026 + .mount(&server)
2027 + .await;
2028 +
2029 + // Fetch every response BEFORE freezing the clock. An empty body is
2030 + // Content-Length 0, so hyper ends the stream without another socket
2031 + // read: inside the loop there is no I/O left, and the only thing that
2032 + // can advance a paused clock is the loop's own sleep.
2033 + let mut responses = std::collections::VecDeque::new();
2034 + for _ in 0..4 {
2035 + responses.push_back(reqwest::get(server.uri()).await.unwrap());
2036 + }
2037 + let responses = std::sync::Mutex::new(responses);
2038 + let stamps = std::sync::Mutex::new(Vec::new());
2039 +
2040 + let client = SyncKitClient::new(test_config());
2041 + tokio::time::pause();
2042 + let start = tokio::time::Instant::now();
2043 + let err = client
2044 + .retry_request_json::<_, _, serde_json::Value>(Idempotency::ReadOnly, || {
2045 + stamps.lock().unwrap().push(tokio::time::Instant::now());
2046 + let resp = responses
2047 + .lock()
2048 + .unwrap()
2049 + .pop_front()
2050 + .expect("the loop asked for a fifth attempt");
2051 + async move { Result::<reqwest::Response>::Ok(resp) }
2052 + })
2053 + .await
2054 + .unwrap_err();
2055 + let end = tokio::time::Instant::now();
2056 +
2057 + // An empty body is not JSON, so every attempt fails in the parse.
2058 + assert!(matches!(err, SyncKitError::Json(_)), "got {err:?}");
2059 + let stamps = stamps.into_inner().unwrap();
2060 + assert_eq!(stamps.len(), 4, "MAX_RETRIES is 3, so 1 try plus 3 replays");
2061 + assert_eq!(stamps[0] - start, Duration::ZERO);
2062 + let gaps = gaps(&stamps);
2063 + assert_slept(gaps[0], 1_000);
2064 + assert_slept(gaps[1], 2_000);
2065 + assert_slept(gaps[2], 4_000);
2066 + assert_eq!(
2067 + end - stamps[3],
2068 + Duration::ZERO,
2069 + "slept after the final attempt, which buys nothing"
2070 + );
2071 + }
2072 +
2073 + /// Serve one chunked HTTP/1.1 response carrying `len` bytes and no
2074 + /// `Content-Length`. The whole response goes out in a single write so the
2075 + /// body reaches `bytes_stream` as one item.
2076 + async fn chunked_body_of_len(len: usize) -> reqwest::Response {
2077 + use tokio::io::{AsyncReadExt, AsyncWriteExt};
2078 + ensure_crypto_provider();
2079 + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2080 + let addr = listener.local_addr().unwrap();
2081 + tokio::spawn(async move {
2082 + let (mut sock, _) = listener.accept().await.unwrap();
2083 + // Drain the request head before replying. A single `read` is not
2084 + // enough: TCP may split the request across segments, and answering
2085 + // a half-read request races the client's own write.
2086 + let mut request = Vec::new();
2087 + let mut scratch = [0u8; 256];
2088 + while !request.windows(4).any(|w| w == b"\r\n\r\n") {
2089 + let n = sock.read(&mut scratch).await.unwrap();
2090 + assert_ne!(n, 0, "the client closed before sending a request head");
2091 + request.extend_from_slice(&scratch[..n]);
2092 + }
2093 + let mut out = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec();
2094 + out.extend_from_slice(format!("{len:x}\r\n").as_bytes());
2095 + out.extend_from_slice(&vec![b'x'; len]);
2096 + out.extend_from_slice(b"\r\n0\r\n\r\n");
2097 + sock.write_all(&out).await.unwrap();
2098 + sock.flush().await.unwrap();
2099 + });
2100 + reqwest::get(format!("http://{addr}/")).await.unwrap()
2101 + }
2102 +
2103 + #[tokio::test]
2104 + async fn read_body_capped_rejects_an_oversized_body_that_declares_no_length() {
2105 + // Every other oversized case is caught by the Content-Length fast path,
2106 + // which leaves the running `buf.len() + chunk.len()` total unobserved.
2107 + // A chunked body has no declared length, so that accumulation is the
2108 + // only thing standing between the cap and an unbounded read: with the
2109 + // buffer still empty it is the incoming chunk's own size that has to
2110 + // trip the limit.
2111 + let resp = chunked_body_of_len(100).await;
2112 + assert_eq!(
2113 + resp.content_length(),
2114 + None,
2115 + "the fast path must not be what rejects this"
2116 + );
2117 + let err = read_body_capped(resp, 50).await.unwrap_err();
2118 + match err {
2119 + SyncKitError::Internal(msg) => {
2120 + assert!(msg.contains("50"), "message did not name the cap: {msg}");
2121 + }
2122 + other => panic!("expected Internal, got {other:?}"),
2123 + }
2124 + }
1903 2125 }
M synckit-client/src/store/apply.rs +236
@@ -787,6 +787,19 @@
787 787 .pk(&["key"])
788 788 .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
789 789 SyncTable::full("reffer", &["id", "ext_id"]).references_unsynced(),
790 + // `kind` is NOT NULL *with a default*, the only shape in which
791 + // omitting a column and binding an explicit NULL differ observably.
792 + SyncTable::full("note", &["id", "body", "kind"]),
793 + // A preserved column that is also a whitelist column, so a payload
794 + // can carry it and the ON CONFLICT SET has to refuse it.
795 + SyncTable::full("vault", &["id", "label", "token"]).preserve_local(&["token"]),
796 + // Partial update on a composite key: two WHERE bindings, not one.
797 + SyncTable::full("pairflag", &["a", "b", "flag"])
798 + .pk(&["a", "b"])
799 + .partial_update(&["flag"]),
800 + // INTEGER PRIMARY KEY, so a text id is a datatype mismatch: a SQLite
801 + // failure that is not a constraint violation.
802 + SyncTable::full("tally", &["id", "label"]),
790 803 ])
791 804 }
792 805
@@ -804,6 +817,10 @@
804 817 CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT, deleted_at INTEGER);
805 818 CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);
806 819 CREATE TABLE reffer (id TEXT PRIMARY KEY, ext_id INTEGER NOT NULL REFERENCES ghost(id));
820 + CREATE TABLE note (id TEXT PRIMARY KEY, body TEXT, kind TEXT NOT NULL DEFAULT 'plain');
821 + CREATE TABLE vault (id TEXT PRIMARY KEY, label TEXT, token TEXT);
822 + CREATE TABLE pairflag (a TEXT, b TEXT, flag INTEGER, PRIMARY KEY (a, b));
823 + CREATE TABLE tally (id INTEGER PRIMARY KEY, label TEXT);
807 824 ",
808 825 )
809 826 .unwrap();
@@ -1408,4 +1425,223 @@
1408 1425 .collect();
1409 1426 assert_eq!(read, vec![0, 1], "the payload key names the row to update");
1410 1427 }
1428 +
1429 + #[test]
1430 + fn a_non_constraint_sqlite_failure_rolls_the_whole_batch_back() {
1431 + let mut conn = db();
1432 + // tally.id is an INTEGER PRIMARY KEY, so a text id SQLite cannot coerce
1433 + // raises SQLITE_MISMATCH, not SQLITE_CONSTRAINT. Only a constraint
1434 + // violation is survivable; anything else means the batch cannot be
1435 + // trusted, so it must surface as Err rather than as a deferred row.
1436 + let changes = ResolvedChanges::for_test(vec![
1437 + upsert("parent", "p1", json!({"id":"p1","name":"a"})),
1438 + upsert("tally", "t1", json!({"id":"notanint","label":"x"})),
1439 + ]);
1440 + let e = apply_remote_changes(&mut conn, &schema(), &changes, "")
1441 + .expect_err("a non-constraint SQLite error must not be swallowed as deferred");
1442 + assert!(matches!(e, crate::error::SyncKitError::Database(_)));
1443 + assert_eq!(
1444 + conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
1445 + .unwrap(),
1446 + 0,
1447 + "the valid row earlier in the same batch rolls back with it"
1448 + );
1449 + assert_eq!(
1450 + conn.query_row("SELECT COUNT(*) FROM tally", [], |r| r.get::<_, i64>(0))
1451 + .unwrap(),
1452 + 0
1453 + );
1454 + }
1455 +
1456 + #[test]
1457 + fn a_null_for_a_defaulted_not_null_column_is_omitted_not_bound() {
1458 + let mut conn = db();
1459 + // note.kind is NOT NULL DEFAULT 'plain'. Omitting it (what the NOT NULL
1460 + // set is read for) takes the default; binding an explicit NULL would
1461 + // violate instead, so the two are distinguishable here.
1462 + let o = apply(
1463 + &mut conn,
1464 + &[upsert(
1465 + "note",
1466 + "n1",
1467 + json!({"id":"n1","body":"b1","kind":null}),
1468 + )],
1469 + );
1470 + assert_eq!(o.applied, 1);
1471 + assert!(o.deferred.is_empty(), "an omitted column takes its default");
1472 + let kind: String = conn
1473 + .query_row("SELECT kind FROM note WHERE id='n1'", [], |r| r.get(0))
1474 + .unwrap();
1475 + assert_eq!(kind, "plain");
1476 +
1477 + // Same on the update leg: the column is left out of the ON CONFLICT SET,
1478 + // so a local value stands rather than being nulled.
1479 + conn.execute("UPDATE note SET kind='code' WHERE id='n1'", [])
1480 + .unwrap();
1481 + let o2 = apply(
1482 + &mut conn,
1483 + &[upsert(
1484 + "note",
1485 + "n1",
1486 + json!({"id":"n1","body":"b2","kind":null}),
1487 + )],
1488 + );
1489 + assert_eq!(o2.applied, 1);
1490 + assert!(o2.deferred.is_empty());
1491 + let (body, kind): (String, String) = conn
1492 + .query_row("SELECT body, kind FROM note WHERE id='n1'", [], |r| {
1493 + Ok((r.get(0)?, r.get(1)?))
1494 + })
1495 + .unwrap();
1496 + assert_eq!(body, "b2", "a nullable column still updates");
1497 + assert_eq!(
1498 + kind, "code",
1499 + "a NOT NULL column the payload nulled keeps its local value"
1500 + );
1501 + }
1502 +
1503 + #[test]
1504 + fn a_preserved_column_carried_in_the_payload_is_still_not_overwritten() {
1505 + let mut conn = db();
1506 + // vault.token is both a whitelist column and preserve_local, so the
1507 + // payload can carry it and the ON CONFLICT SET must still refuse it.
1508 + // Only the preserve filter keeps it out; the PK filter would not.
1509 + let o = apply(
1510 + &mut conn,
1511 + &[upsert(
1512 + "vault",
1513 + "v1",
1514 + json!({"id":"v1","label":"l1","token":"seed"}),
1515 + )],
1516 + );
1517 + assert_eq!(o.applied, 1);
1518 + conn.execute("UPDATE vault SET token='local' WHERE id='v1'", [])
1519 + .unwrap();
1520 + apply(
1521 + &mut conn,
1522 + &[upsert(
1523 + "vault",
1524 + "v1",
1525 + json!({"id":"v1","label":"l2","token":"remote"}),
1526 + )],
1527 + );
1528 + let (label, token): (String, String) = conn
1529 + .query_row("SELECT label, token FROM vault WHERE id='v1'", [], |r| {
1530 + Ok((r.get(0)?, r.get(1)?))
1531 + })
1532 + .unwrap();
1533 + assert_eq!(label, "l2", "a non-preserved column still updates");
1534 + assert_eq!(
1535 + token, "local",
1536 + "preserve_local outranks a payload that carries the column"
1537 + );
1538 + }
1539 +
1540 + #[test]
1541 + fn a_pre_existing_orphan_survives_a_sweep_of_its_own_table() {
1542 + let mut conn = db();
1543 + // An orphan left by an earlier relaxed apply. No entry in this batch
1544 + // describes it, so there is nothing to hold and deleting it would lose
1545 + // the row for good.
1546 + conn.execute_batch(
1547 + "PRAGMA foreign_keys=OFF;
1548 + INSERT INTO child (id, parent_id, note) VALUES ('ghost1', 'gone', 'g');
1549 + PRAGMA foreign_keys=ON;",
1550 + )
1551 + .unwrap();
1552 + // `reffer` turns the batch-wide relaxation on, so the sweep runs over
1553 + // `child`. The batch names the table but not this row.
1554 + let o = apply(
1555 + &mut conn,
1556 + &[
1557 + upsert("reffer", "r1", json!({"id":"r1","ext_id":404})),
1558 + upsert("parent", "p1", json!({"id":"p1","name":"a"})),
1559 + upsert(
1560 + "child",
1561 + "c1",
1562 + json!({"id":"c1","parent_id":"p1","note":"n"}),
1563 + ),
1564 + ],
1565 + );
1566 + assert_eq!(o.applied, 3);
1567 + assert!(
1568 + o.deferred.is_empty(),
1569 + "a row this batch never pulled cannot be deferred for retry"
1570 + );
1571 + let ids: Vec<String> = conn
1572 + .prepare("SELECT id FROM child ORDER BY id")
1573 + .unwrap()
1574 + .query_map([], |r| r.get(0))
1575 + .unwrap()
1576 + .map(std::result::Result::unwrap)
1577 + .collect();
1578 + assert_eq!(
1579 + ids,
1580 + vec!["c1".to_string(), "ghost1".to_string()],
1581 + "matching the table alone is not matching the row"
1582 + );
1583 + }
1584 +
1585 + #[test]
1586 + fn a_composite_partial_update_binds_every_key_component() {
1587 + let mut conn = db();
1588 + conn.execute_batch("INSERT INTO pairflag (a, b, flag) VALUES ('x','y',0), ('x','z',0);")
1589 + .unwrap();
1590 + let o = apply(
1591 + &mut conn,
1592 + &[ChangeEntry {
1593 + op: ChangeOp::Update,
1594 + ..upsert("pairflag", "x:z", json!({"a":"x","b":"z","flag":1}))
1595 + }],
1596 + );
1597 + assert_eq!(o.applied, 1);
1598 + let flags: Vec<(String, i64)> = conn
1599 + .prepare("SELECT b, flag FROM pairflag ORDER BY b")
1600 + .unwrap()
1601 + .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
1602 + .unwrap()
1603 + .map(std::result::Result::unwrap)
1604 + .collect();
1605 + // Each key component needs its own placeholder: reusing the first would
1606 + // compare `b` against the value of `a` and touch the wrong row, or none.
1607 + assert_eq!(
1608 + flags,
1609 + vec![("y".to_string(), 0), ("z".to_string(), 1)],
1610 + "only the row the whole composite key names is updated"
1611 + );
1612 + }
1613 +
1614 + #[test]
1615 + fn a_delete_with_no_payload_falls_back_to_the_wire_row_id() {
1616 + let mut conn = db();
1617 + apply(
1618 + &mut conn,
1619 + &[
1620 + upsert("parent", "p1", json!({"id":"p1","name":"a"})),
1621 + upsert("parent", "p2", json!({"id":"p2","name":"b"})),
1622 + ],
1623 + );
1624 + // An older client wrote the key into the wire row id and sent no payload.
1625 + // A single-PK table can still be addressed from it.
1626 + let mut change = delete("parent", "p1", json!({}));
1627 + change.data = None;
1628 + let o = apply(&mut conn, &[change]);
1629 + assert_eq!(o.applied, 1);
1630 + assert!(
1631 + o.rejected.is_empty(),
1632 + "a single-PK delete is reconstructable from the row id alone"
1633 + );
1634 + let left: Vec<String> = conn
1635 + .prepare("SELECT id FROM parent ORDER BY id")
1636 + .unwrap()
1637 + .query_map([], |r| r.get(0))
1638 + .unwrap()
1639 + .map(std::result::Result::unwrap)
1640 + .collect();
1641 + assert_eq!(
1642 + left,
1643 + vec!["p2".to_string()],
1644 + "the row the wire id names, and only it, goes"
1645 + );
1646 + }
1411 1647 }
M synckit-client/src/store/blob.rs +92 -1
@@ -311,6 +311,10 @@
311 311 #[derive(Clone, Default)]
312 312 struct FakeBlobs {
313 313 store: Arc<Mutex<HashMap<String, Vec<u8>>>>,
314 + /// Every `confirm` call, in order. Recorded because whether confirm
315 + /// happens is the whole difference between an upload and a dedup, and
316 + /// the counters `upload_blobs` returns are identical either way.
317 + confirms: Arc<Mutex<Vec<(String, u64)>>>,
314 318 }
315 319 impl FakeBlobs {
316 320 fn put_raw(&self, hash: &str, bytes: Vec<u8>) {
@@ -319,6 +323,9 @@
319 323 fn has(&self, hash: &str) -> bool {
320 324 self.store.lock().unwrap().contains_key(hash)
321 325 }
326 + fn confirms(&self) -> Vec<(String, u64)> {
327 + self.confirms.lock().unwrap().clone()
328 + }
322 329 }
323 330 impl BlobTransport for FakeBlobs {
324 331 async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
@@ -329,7 +336,8 @@
329 336 self.store.lock().unwrap().insert(hash.to_string(), data);
330 337 Ok(BlobUploadOutcome::Uploaded)
331 338 }
332 - async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
339 + async fn confirm(&self, hash: &str, size: u64) -> Result<()> {
340 + self.confirms.lock().unwrap().push((hash.to_string(), size));
333 341 Ok(())
334 342 }
335 343 async fn download_url(&self, hash: &str) -> Result<String> {
@@ -594,4 +602,87 @@
594 602 .unwrap();
595 603 assert_eq!(co, 1, "reconcile flips a missing-file row to cloud_only");
596 604 }
605 +
606 + #[tokio::test]
607 + async fn upload_confirms_exactly_what_it_sent_and_never_what_the_server_already_had() {
608 + // `upload_blobs` returns 1 in both cases, so the count cannot tell an
609 + // upload from a dedup. The confirm call can: it is the write that
610 + // commits the blob server-side, and firing it for content nobody sent
611 + // (or skipping it for content that was sent) is the whole failure.
612 + let dir = tempdir();
613 + let (db, policy) = setup(&dir);
614 + let server = FakeBlobs::default();
615 + let content = b"confirm-me-please".to_vec(); // 17 bytes, not 0 or 1
616 + let h = hash_of(&content);
617 + seed_meta(&db, &h, "wav", content.len() as u64, 0);
618 + std::fs::write(
619 + policy.local_path(&BlobRef {
620 + hash: h.clone(),
621 + ext: "wav".into(),
622 + size: 0,
623 + }),
624 + &content,
625 + )
626 + .unwrap();
627 +
628 + assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
629 + assert_eq!(
630 + server.confirms(),
631 + vec![(h.clone(), 17)],
632 + "an upload that sent bytes must confirm them, once, at the row's size"
633 + );
634 +
635 + // Second pass: the server now holds the hash, so it answers
636 + // AlreadyPresent and nothing may be confirmed.
637 + assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
638 + assert_eq!(
639 + server.confirms(),
640 + vec![(h, 17)],
641 + "a deduped blob sent no bytes, so there is nothing to confirm"
642 + );
643 + }
644 +
645 + #[tokio::test]
646 + async fn the_client_transport_forwards_each_call_to_the_sdk_method() {
647 + // The impl is four one-line forwards, and a forward that answers on its
648 + // own instead of calling through is invisible to every other test in
649 + // this file (they all run against FakeBlobs). Pin it by giving the
650 + // client nothing to work with: each SDK method has its own refusal, and
651 + // a body that did not call through could not produce it.
652 + let client = SyncKitClient::new(crate::SyncKitConfig {
653 + server_url: "https://example.invalid".to_string(),
654 + api_key: "test-api-key".to_string(),
655 + });
656 + let hash = "a".repeat(64);
657 +
658 + // blob_upload_streaming asks for the master key before anything else.
659 + let e = BlobTransport::upload_file(&client, &hash, Path::new("/nonexistent"))
660 + .await
661 + .unwrap_err();
662 + assert!(matches!(e, SyncKitError::NoMasterKey), "upload_file: {e:?}");
663 +
664 + // blob_confirm and blob_download_url both need a session.
665 + let e = BlobTransport::confirm(&client, &hash, 4096)
666 + .await
667 + .unwrap_err();
668 + assert!(
669 + matches!(e, SyncKitError::NotAuthenticated),
670 + "confirm: {e:?}"
671 + );
672 + let e = BlobTransport::download_url(&client, &hash)
673 + .await
674 + .unwrap_err();
675 + assert!(
676 + matches!(e, SyncKitError::NotAuthenticated),
677 + "download_url: {e:?}"
678 + );
679 +
680 + // blob_download GETs the presigned URL first, so an unusable URL is its
681 + // refusal. A relative URL is a reqwest builder error, which the retry
682 + // layer classifies as permanent, so this makes no network attempt.
683 + let e = BlobTransport::download(&client, &hash, "not-a-url")
684 + .await
685 + .unwrap_err();
686 + assert!(matches!(e, SyncKitError::Http(_)), "download: {e:?}");
687 + }
597 688 }
M synckit-client/src/store/hlc.rs +88
@@ -2065,4 +2065,92 @@
2065 2065 .unwrap();
2066 2066 assert_eq!(oldest_kept, "50");
2067 2067 }
2068 +
2069 + /// `stamp_pending` has to leave the advanced clock on disk, not only on the
2070 + /// changelog rows. The wall component is high (5_000_000) and the row count
2071 + /// is three, so the persisted clock is a value no fresh `Hlc::zero` can
2072 + /// coincide with, and the second stamp runs at a LOWER now_ms (1000) so a
2073 + /// clock that failed to persist would visibly restart at 1000 instead of
2074 + /// continuing the counter at 5_000_000.
2075 + #[test]
2076 + fn stamp_pending_persists_the_advanced_clock() {
2077 + let (conn, n) = device(1);
2078 + for id in ["r1", "r2", "r3"] {
2079 + conn.execute("INSERT INTO note (id, name) VALUES (?1, 'v')", [id])
2080 + .unwrap();
2081 + }
2082 + assert_eq!(stamp_pending(&conn, n, 5_000_000).unwrap(), 3);
2083 +
2084 + // Three ticks at one wall component: adopt 5_000_000 with counter 0,
2085 + // then bump twice.
2086 + assert_eq!(
2087 + load_clock(&conn, n).unwrap(),
2088 + Hlc {
2089 + wall_ms: 5_000_000,
2090 + counter: 2,
2091 + node: n
2092 + },
2093 + "the clock reached by stamping must survive a reload"
2094 + );
2095 +
2096 + conn.execute("INSERT INTO note (id, name) VALUES ('r4', 'v')", [])
2097 + .unwrap();
2098 + assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 1);
2099 + let r4 = load_local_pending(&conn, n)
2100 + .unwrap()
2101 + .into_iter()
2102 + .find(|e| e.row_id == "r4")
2103 + .unwrap();
2104 + assert_eq!(
2105 + r4.hlc,
2106 + Hlc {
2107 + wall_ms: 5_000_000,
2108 + counter: 3,
2109 + node: n
2110 + },
2111 + "a later stamp at an earlier now_ms must continue the reloaded clock"
2112 + );
2113 + }
2114 +
2115 + /// `observe` has to leave the merged clock on disk: the whole point is that
2116 + /// a subsequent local write outranks the remote it observed. The remote sits
2117 + /// far in the future (9_000_000_000) and the local stamp that follows runs
2118 + /// at now_ms 2000, so if the merge were not persisted the new stamp would
2119 + /// land at wall 2000 and lose to the remote by seven orders of magnitude.
2120 + #[test]
2121 + fn observe_persists_the_merged_clock() {
2122 + let (conn, n) = device(1);
2123 + let remote = Hlc {
2124 + wall_ms: 9_000_000_000,
2125 + counter: 5,
2126 + node: node(2),
2127 + };
2128 + observe(&conn, n, [remote], 1000).unwrap();
2129 + assert_eq!(
2130 + load_clock(&conn, n).unwrap(),
2131 + Hlc {
2132 + wall_ms: 9_000_000_000,
2133 + counter: 6,
2134 + node: n
2135 + },
2136 + "observe must persist the remote wall and a strictly greater counter"
2137 + );
2138 +
2139 + conn.execute("INSERT INTO note (id, name) VALUES ('r', 'v')", [])
2140 + .unwrap();
2141 + assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 1);
2142 + let stamped = load_local_pending(&conn, n).unwrap().remove(0).hlc;
2143 + assert_eq!(
2144 + stamped,
2145 + Hlc {
2146 + wall_ms: 9_000_000_000,
2147 + counter: 7,
2148 + node: n
2149 + }
2150 + );
2151 + assert!(
2152 + stamped > remote,
2153 + "a local write after observing must outrank the observed remote"
2154 + );
2155 + }
2068 2156 }
M synckit-client/src/store/schema.rs +32
@@ -876,4 +876,36 @@
876 876 // The failure carries the manifest, so the diff names what moved.
877 877 assert!(msg.contains("cols=id,name,archived"), "{msg}");
878 878 }
879 +
880 + /// `group_scope()` is public API for a consumer asserting exactly which
881 + /// tables are group-scoped, and nothing in the engine calls it (every
882 + /// production site reads the field), so this is the only place the accessor
883 + /// is observed. Three tables, three different answers, so a body returning
884 + /// any fixed value disagrees with at least two rows: the two scoped tables
885 + /// name different columns, and the third is personal-only.
886 + #[test]
887 + fn group_scope_reports_the_declared_provenance_column_per_table() {
888 + let schema = SyncSchema::new(vec![
889 + SyncTable::full("project", &["id", "name"]).group_scoped("group_id"),
890 + SyncTable::full("note", &["id", "body"]).group_scoped("owner_group"),
891 + SyncTable::full("task", &["id", "project_id", "title", "done"]),
892 + ]);
893 +
894 + let expected: [(&str, Option<&'static str>); 3] = [
895 + ("project", Some("group_id")),
896 + ("note", Some("owner_group")),
897 + ("task", None),
898 + ];
899 + for (table, want) in expected {
900 + let t = schema
901 + .tables()
902 + .iter()
903 + .find(|t| t.name() == table)
904 + .unwrap_or_else(|| panic!("{table} is in the schema"));
905 + assert_eq!(t.group_scope(), want, "group_scope() for {table}");
906 + // The accessor is the field, not a constant: the engine reads the
907 + // field directly (migrate.rs, apply.rs) and the two must not drift.
908 + assert_eq!(t.group_scope(), t.group_scope, "accessor for {table}");
909 + }
910 + }
879 911 }
M synckit-client/src/store/sync.rs +75 -1
@@ -1496,8 +1496,14 @@
1496 1496 )
1497 1497 .unwrap();
1498 1498 }
1499 + let before: i64 = conn
1500 + .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
1501 + .unwrap();
1499 1502 // Cap to 3: keep the 3 most recent rows, drop older PUSHED ones only.
1500 1503 let dropped = enforce_changelog_retention(&conn, 3).unwrap();
1504 + let after: i64 = conn
1505 + .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
1506 + .unwrap();
1501 1507 let unpushed: i64 = conn
1502 1508 .query_row(
1503 1509 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
@@ -1506,7 +1512,75 @@
1506 1512 )
1507 1513 .unwrap();
1508 1514 assert_eq!(unpushed, 2, "unpushed entries are never dropped");
1509 - assert!(dropped >= 1);
1515 + // The 3 newest rows by id are the 2 unpushed plus the newest pushed one,
1516 + // so exactly the 4 oldest pushed rows go. A count that is not 4 - a
1517 + // hardcoded 0 or 1, or the row count of some other query - disagrees.
1518 + assert_eq!(
1519 + dropped, 4,
1520 + "the four oldest pushed rows are the only ones dropped"
1521 + );
1522 + assert_eq!(
1523 + dropped,
1524 + (before - after) as u64,
1525 + "the returned count is the number of rows actually deleted"
1526 + );
1527 + assert_eq!(after, 3, "the cap is the number of surviving rows here");
1528 + }
1529 +
1530 + #[test]
1531 + fn retention_keeps_exactly_cap_rows_and_drops_the_next_one() {
1532 + let dir = tempdir();
1533 + let (da, _) = device(&dir.join("retention_edge.db"), 27);
1534 + let conn = da.open().unwrap();
1535 + conn.execute("DELETE FROM sync_changelog", []).unwrap();
1536 + // 6 pushed rows, ids ascending with insertion order.
1537 + for i in 0..6 {
1538 + conn.execute(
1539 + "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
1540 + [format!("p{i}")],
1541 + )
1542 + .unwrap();
1543 + }
1544 + let ids: Vec<i64> = {
1545 + let mut st = conn
1546 + .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
1547 + .unwrap();
1548 + let rows: Vec<i64> = st
1549 + .query_map([], |r| r.get(0))
1550 + .unwrap()
1551 + .map(|r| r.unwrap())
1552 + .collect();
1553 + rows
1554 + };
1555 + assert_eq!(ids.len(), 6);
1556 +
1557 + // Cap 4 of 6 keeps the newest four and drops two. Both sides of the
1558 + // boundary are named: ids[1] is the last row dropped, ids[2] the first
1559 + // row kept, so an off-by-one in the LIMIT changes the answer.
1560 + let dropped = enforce_changelog_retention(&conn, 4).unwrap();
1561 + assert_eq!(dropped, 2, "6 rows capped at 4 drops 2");
1562 + let survivors: Vec<i64> = {
1563 + let mut st = conn
1564 + .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
1565 + .unwrap();
1566 + st.query_map([], |r| r.get(0))
1567 + .unwrap()
1568 + .map(|r| r.unwrap())
1569 + .collect()
1570 + };
1571 + assert_eq!(survivors, ids[2..].to_vec(), "the newest four rows survive");
1572 + assert!(
1573 + !survivors.contains(&ids[1]),
1574 + "the row just past the cap is gone"
1575 + );
1576 +
1577 + // Re-running at the same cap is a no-op: nothing is left to drop, so a
1578 + // constant return value of 1 or 2 disagrees with 0 here.
1579 + assert_eq!(
1580 + enforce_changelog_retention(&conn, 4).unwrap(),
1581 + 0,
1582 + "a second pass at the same cap drops nothing"
1583 + );
1510 1584 }
1511 1585
1512 1586 #[tokio::test]
M synckit-client/tests/integration/blob.rs +130
@@ -339,3 +339,133 @@
339 339 "1MB upload should add exactly the v3 chunked overhead"
340 340 );
341 341 }
342 +
343 + // ── The v3 framing boundary on the download path ──
344 + //
345 + // `blob_download` decides three things from lengths alone: that four bytes are
346 + // in hand before it reads the format tag, that the 4-byte tag plus the 13-byte
347 + // header (17 bytes) are in hand before it parses the header, and that a chunk is
348 + // complete before it decrypts. Each is a comparison against a literal, and a
349 + // wrong one is invisible to a test that only serves whole, well-formed blobs:
350 + // every such body is far past all three boundaries, so the comparisons agree.
351 + // The bodies below sit exactly on them.
352 +
353 + /// The message `blob_download` refused `body` with, served at `path`.
354 + async fn download_refusal(
355 + kit: &MockKit,
356 + client: &SyncKitClient,
357 + path: &str,
358 + hash: &str,
359 + body: Vec<u8>,
360 + ) -> String {
361 + kit.get(path).bytes(body).await;
362 + match client.blob_download(hash, &kit.url(path)).await {
363 + Err(SyncKitError::Crypto(message)) => message,
364 + Err(other) => panic!("expected a Crypto refusal at {path}, got {other:?}"),
365 + Ok(_) => panic!("{path} must not be accepted as a blob"),
366 + }
367 + }
368 +
369 + #[tokio::test]
370 + async fn a_v3_body_that_stops_short_of_its_header_is_refused_as_a_missing_header() {
371 + let kit = MockKit::start().await;
372 + let (client, _key) = kit.keyed();
373 + let hash = hex::encode(sha2::Sha256::digest(b"never served"));
374 + let header = synckit_client::crypto::blob_header_bytes(5_000);
375 + assert_eq!(
376 + header.len(),
377 + 17,
378 + "4-byte format tag plus the 13-byte header"
379 + );
380 +
381 + // Exactly the format tag: enough to know the format, nothing to parse. The
382 + // reader must hold on for the header rather than take the four bytes as one.
383 + let tag_only =
384 + download_refusal(&kit, &client, "/s3/tag-only", &hash, header[..4].to_vec()).await;
385 + assert_eq!(
386 + tag_only, "v3 blob ended before its header",
387 + "four bytes is the tag and no more"
388 + );
389 +
390 + // Between the two boundaries: past the tag, short of the header.
391 + let partial = download_refusal(
392 + &kit,
393 + &client,
394 + "/s3/partial-header",
395 + &hash,
396 + header[..10].to_vec(),
397 + )
398 + .await;
399 + assert_eq!(
400 + partial, "v3 blob ended before its header",
401 + "ten bytes is still short of the 17-byte header"
402 + );
403 + }
404 +
405 + #[tokio::test]
406 + async fn a_v3_body_of_exactly_its_header_is_parsed_and_then_found_to_have_no_chunks() {
407 + // Dead on the boundary: the header is complete, so it must be parsed, and
408 + // the refusal must be about the missing chunks rather than the header. A
409 + // reader that waits for one more byte before parsing gives the other
410 + // message, and no whole-blob fixture can tell the two apart.
411 + let kit = MockKit::start().await;
412 + let (client, _key) = kit.keyed();
413 + let hash = hex::encode(sha2::Sha256::digest(b"never served"));
414 + let header = synckit_client::crypto::blob_header_bytes(5_000);
415 +
416 + let message = download_refusal(&kit, &client, "/s3/header-only", &hash, header).await;
417 + assert_eq!(
418 + message, "v3 blob ended mid-chunk or had trailing bytes",
419 + "a complete header with no chunk behind it is a truncated blob, not a missing header"
420 + );
421 + }
422 +
423 + #[tokio::test]
424 + async fn a_v3_blob_one_byte_short_or_one_byte_long_is_refused() {
425 + let kit = MockKit::start().await;
426 + let (client, key) = kit.keyed();
427 +
428 + // Two chunks plus a remainder, so the last chunk is a short one and the
429 + // truncation lands inside it.
430 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE + 4_321))
431 + .map(|i| i as u8)
432 + .collect();
433 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
434 + let blob = synckit_client::crypto::encrypt_blob_chunked(&plaintext, &key, &hash).unwrap();
435 +
436 + // The control: intact, these exact bytes decrypt. Without it the two
437 + // refusals below would also pass if the reader refused everything.
438 + kit.get("/s3/intact").bytes(blob.clone()).await;
439 + assert_eq!(
440 + client
441 + .blob_download(&hash, &kit.url("/s3/intact"))
442 + .await
443 + .unwrap(),
444 + plaintext,
445 + "the intact blob must round-trip"
446 + );
447 +
448 + let short = download_refusal(
449 + &kit,
450 + &client,
451 + "/s3/one-short",
452 + &hash,
453 + blob[..blob.len() - 1].to_vec(),
454 + )
455 + .await;
456 + assert_eq!(
457 + short, "v3 blob ended mid-chunk or had trailing bytes",
458 + "a chunk one byte short is incomplete and must never be decrypted"
459 + );
460 +
461 + // One byte past the end: every chunk is complete and the plaintext hashes
462 + // correctly, so nothing but the leftover byte is wrong. A reader that only
463 + // counted chunks would accept this.
464 + let mut long = blob.clone();
465 + long.push(0);
466 + let long = download_refusal(&kit, &client, "/s3/one-long", &hash, long).await;
467 + assert_eq!(
468 + long, "v3 blob ended mid-chunk or had trailing bytes",
469 + "a trailing byte is not part of any chunk and must be refused"
470 + );
471 + }
M synckit-client/tests/integration/blob_multipart.rs +299
@@ -765,3 +765,302 @@
765 765 .await;
766 766 }
767 767 }
768 +
769 + // ── Hostile part plans ──
770 + //
771 + // `part_size` and `part_count` come from the server and drive both the
772 + // allocation and the cut points of the sealed stream, so every bound on them is
773 + // arithmetic the client cannot get wrong quietly. Each case below is written as
774 + // a pair: the value that must be accepted by a gate and the adjacent one that
775 + // must not, told apart by which message came back. A test that only checked
776 + // "is_err" would pass with any gate firing, including the wrong one.
777 +
778 + /// Mount a start response carrying an arbitrary plan, plus the abort the client
779 + /// makes on its way out. No part-URL route is mounted: every case here must be
780 + /// refused before a part is requested.
781 + async fn mount_hostile_plan(kit: &MockKit, part_size: u64, part_count: u32) {
782 + kit.post(START_PATH)
783 + .json(json!({
784 + "upload_id": "hostile-upload-id",
785 + "part_size": part_size,
786 + "part_count": part_count,
787 + "already_exists": false,
788 + }))
789 + .await;
790 + kit.post(ABORT_PATH).code(204).empty().await;
791 + }
792 +
793 + /// Run a streaming upload of a 5000-byte blob against `(part_size, part_count)`
794 + /// and return the internal-error message it was refused with.
795 + async fn plan_rejection(part_size: u64, part_count: u32) -> String {
796 + let kit = MockKit::start().await;
797 + let (client, _key) = kit.keyed();
798 + let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
799 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
800 + let file = temp_blob("hostile.bin", &plaintext);
801 + mount_hostile_plan(&kit, part_size, part_count).await;
802 +
803 + let err = client
804 + .blob_upload_streaming(&hash, &file)
805 + .await
806 + .unwrap_err();
807 + std::fs::remove_file(&file).ok();
808 + assert_eq!(
809 + kit.hits(PARTS_PATH).await,
810 + 0,
811 + "a plan refused up front must not mint a single part URL"
812 + );
813 + match err {
814 + SyncKitError::Internal(message) => message,
815 + other => panic!("expected an Internal rejection, got {other:?}"),
816 + }
817 + }
818 +
819 + #[tokio::test]
820 + async fn a_plan_with_no_bytes_per_part_or_no_parts_is_refused_as_empty() {
821 + // part_size 0 is also the divisor of the tiling check below the guard, so a
822 + // guard that let it through would divide by zero rather than mis-upload.
823 + assert!(
824 + plan_rejection(0, 3).await.contains("empty multipart plan"),
825 + "part_size 0 must be refused as an empty plan"
826 + );
827 + // part_count 0 is the other half of the same `||`: with an `&&` in its
828 + // place, a plan that is empty in only one of the two ways gets through.
829 + assert!(
830 + plan_rejection(1024 * 1024, 0)
831 + .await
832 + .contains("empty multipart plan"),
833 + "part_count 0 must be refused as an empty plan"
834 + );
835 + }
836 +
837 + #[tokio::test]
838 + async fn the_part_size_ceiling_admits_exactly_one_gibibyte_and_refuses_one_byte_more() {
839 + // At the ceiling the plan is legal geometry and is judged on whether it
840 + // tiles the blob (it does not: 5000 bytes is one part, not two). One byte
841 + // over is refused by the ceiling itself. The two messages name which gate
842 + // fired, which is the only thing that separates `>` from `>=` and `==`.
843 + let at = plan_rejection(1 << 30, 2).await;
844 + assert!(
845 + at.contains("does not match"),
846 + "a part_size of exactly 1 GiB is under the ceiling: {at}"
847 + );
848 + let over = plan_rejection((1 << 30) + 1, 2).await;
849 + assert!(
850 + over.contains("exceeds the 1073741824-byte ceiling"),
851 + "one byte over the ceiling must be refused by it: {over}"
852 + );
853 + }
854 +
855 + #[tokio::test]
856 + async fn the_part_count_ceiling_admits_exactly_ten_thousand_and_refuses_one_more() {
857 + // S3's own hard limit, so 10_000 parts is a legal plan and must reach the
858 + // tiling check; 10_001 is not.
859 + let at = plan_rejection(1024 * 1024, 10_000).await;
860 + assert!(
861 + at.contains("does not match"),
862 + "a part_count of exactly 10000 is under the ceiling: {at}"
863 + );
864 + let over = plan_rejection(1024 * 1024, 10_001).await;
865 + assert!(
866 + over.contains("exceeds the 10000-part ceiling"),
867 + "one part over the ceiling must be refused by it: {over}"
868 + );
869 + }
870 +
871 + /// Mints one part URL per request with a caller-chosen `part_number` and
872 + /// `content_length`, so a test can make the server's signed geometry disagree
873 + /// with the bytes the client holds.
874 + struct LyingPartsResponder {
875 + part_number: i64,
876 + content_length: u64,
877 + base: String,
878 + }
879 +
880 + impl wiremock::Respond for LyingPartsResponder {
881 + fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
882 + ResponseTemplate::new(200).set_body_json(json!({
883 + "parts": [{
884 + "part_number": self.part_number,
885 + "content_length": self.content_length,
886 + "url": format!("{}{PART_PUT_PATH}?partNumber={}", self.base, self.part_number),
887 + }],
888 + }))
889 + }
890 + }
891 +
892 + /// A single-part session whose minted URL carries the given geometry. Returns
893 + /// the error the upload was refused with, or `None` if it went through.
894 + async fn minted_part_rejection(part_number: i64, content_length_delta: i64) -> Option<String> {
895 + let kit = MockKit::start().await;
896 + let (client, _key) = kit.keyed();
897 + let plaintext: Vec<u8> = (0..5_000u32).map(|i| i as u8).collect();
898 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
899 + let file = temp_blob("mismatched-part.bin", &plaintext);
900 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
901 +
902 + // One part holding the whole blob, so the plan itself is beyond reproach and
903 + // only the minted URL disagrees.
904 + kit.post(START_PATH)
905 + .json(json!({
906 + "upload_id": "mismatch-upload-id",
907 + "part_size": cipher_len,
908 + "part_count": 1,
909 + "already_exists": false,
910 + }))
911 + .await;
912 + kit.post(PARTS_PATH)
913 + .responder(LyingPartsResponder {
914 + part_number,
915 + content_length: (cipher_len as i64 + content_length_delta) as u64,
916 + base: kit.uri(),
917 + })
918 + .await;
919 + kit.put(PART_PUT_PATH)
920 + .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
921 + .await;
922 + kit.post(COMPLETE_PATH).code(204).empty().await;
923 + kit.post(ABORT_PATH).code(204).empty().await;
924 +
925 + let outcome = client.blob_upload_streaming(&hash, &file).await;
926 + std::fs::remove_file(&file).ok();
927 + match outcome {
928 + Ok(_) => {
929 + assert_eq!(kit.hits(PART_PUT_PATH).await, 1, "an accepted plan is PUT");
930 + None
931 + }
932 + Err(SyncKitError::Internal(message)) => {
933 + assert_eq!(
934 + kit.hits(PART_PUT_PATH).await,
935 + 0,
936 + "a part whose geometry is disputed must not be sent anyway"
937 + );
938 + Some(message)
939 + }
940 + Err(other) => panic!("expected an Internal rejection, got {other:?}"),
941 + }
942 + }
943 +
944 + #[tokio::test]
945 + async fn a_minted_part_url_that_disagrees_with_the_bytes_in_hand_is_refused() {
946 + // The agreeing case first, so the two disagreements below are known to be
947 + // the only difference: part 1, exactly the bytes the client sealed.
948 + assert!(
949 + minted_part_rejection(1, 0).await.is_none(),
950 + "a URL signed for the part the client actually holds must be used"
951 + );
952 +
953 + // Signed for a different part: PUTting anyway would store the bytes at the
954 + // wrong index and assemble a scrambled object.
955 + let wrong_number = minted_part_rejection(2, 0)
956 + .await
957 + .expect("a URL signed for part 2 must not be used for part 1");
958 + assert!(
959 + wrong_number.contains("part geometry mismatch"),
960 + "wrong part_number: {wrong_number}"
961 + );
962 +
963 + // Signed for a different length: Content-Length is a signed header, so this
964 + // fails SigV4 at S3 with a far less legible error if it is sent.
965 + let wrong_length = minted_part_rejection(1, -1)
966 + .await
967 + .expect("a URL signed for one byte less must not be used");
968 + assert!(
969 + wrong_length.contains("part geometry mismatch"),
970 + "wrong content_length: {wrong_length}"
971 + );
972 + }
973 +
974 + #[tokio::test]
975 + async fn a_resume_that_lands_exactly_on_a_chunk_boundary_seals_the_chunk_afresh() {
976 + // The other resume test picks part boundaries that can never coincide with
977 + // a chunk boundary, which is the common case but only one side of the
978 + // question. Here the first part is exactly the header plus chunk 0, so the
979 + // resume restarts with `within == 0`: nothing of the boundary chunk is at
980 + // S3, and it must therefore be sealed from scratch. Demanding a recorded
981 + // nonce here would fail the upload outright, since no nonce was ever
982 + // recorded for a chunk that was never sent.
983 + let kit = MockKit::start().await;
984 + let key = synckit_client::crypto::generate_master_key();
985 + let store = resume_store("aligned");
986 +
987 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 4_321))
988 + .map(|i| i as u8)
989 + .collect();
990 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
991 + let file = temp_blob("aligned-resume.bin", &plaintext);
992 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
993 + // Header plus one whole sealed chunk: the one part size that puts the
994 + // second part's first byte on chunk 1's first byte.
995 + let part_size = synckit_client::crypto::blob_header_bytes(plaintext.len()).len()
996 + + synckit_client::crypto::sealed_blob_chunk_len(plaintext.len(), 0);
997 + let part_count = cipher_len.div_ceil(part_size);
998 + assert_eq!(part_count, 3, "three parts, resuming at the second");
999 +
1000 + // ── First attempt: one part lands, then the transfer dies ──
1001 + let client = kit.authed();
1002 + client.set_master_key_raw(key);
1003 + client.set_resume_store(Arc::clone(&store));
1004 +
1005 + mount_session_without_put(&kit, cipher_len, part_size).await;
1006 + kit.put(PART_PUT_PATH)
1007 + .responder(DiesAfter {
1008 + ok: 1,
1009 + seen: std::sync::atomic::AtomicUsize::new(0),
1010 + })
1011 + .await;
1012 + kit.post(ABORT_PATH).code(204).empty().await;
1013 +
1014 + let err = client
1015 + .blob_upload_streaming(&hash, &file)
1016 + .await
1017 + .unwrap_err();
1018 + assert!(
1019 + matches!(err, SyncKitError::Server { status: 403, .. }),
1020 + "got {err:?}"
1021 + );
1022 + let first: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(1).collect();
1023 + assert_eq!(
1024 + first[0].len(),
1025 + part_size,
1026 + "the first part is the header and chunk 0 exactly"
1027 + );
1028 +
1029 + let record = store.load(&hash).unwrap().expect("a session was recorded");
1030 + assert_eq!(record.usable_parts().len(), 1);
1031 + assert!(
1032 + record.chunk(1).is_none(),
1033 + "chunk 1 was never sent, so no nonce for it can have been recorded"
1034 + );
1035 +
1036 + // ── Second attempt ──
1037 + kit.reset().await;
1038 + mount_session(&kit, cipher_len, part_size).await;
1039 + kit.post(ABORT_PATH).code(204).empty().await;
1040 +
1041 + let restarted = kit.authed();
1042 + restarted.set_master_key_raw(key);
1043 + restarted.set_resume_store(Arc::clone(&store));
1044 + restarted.blob_upload_streaming(&hash, &file).await.unwrap();
1045 +
1046 + let resumed = put_bodies(&kit).await;
1047 + assert_eq!(
1048 + resumed.len(),
1049 + part_count - 1,
1050 + "only the missing parts go up"
1051 + );
1052 +
1053 + let assembled: Vec<u8> = first
1054 + .iter()
1055 + .chain(resumed.iter())
1056 + .flat_map(Clone::clone)
1057 + .collect();
1058 + assert_eq!(assembled.len(), cipher_len);
1059 + assert_eq!(
1060 + synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
1061 + plaintext,
1062 + "a resume aligned to a chunk boundary must still assemble"
1063 + );
1064 +
1065 + std::fs::remove_file(&file).ok();
1066 + }
M synckit-client/tests/integration/rotation.rs +275
@@ -363,3 +363,278 @@
363 363 assert_eq!(a.hlc, b.hlc, "the rotation changed a row's clock");
364 364 }
365 365 }
366 +
367 + // ── The re-encrypt loop pages, and the old key stops working ──
368 +
369 + /// The rows the paging test carries, split into the two server pages below.
370 + /// Distinct payloads per row so a re-encryption that swapped two rows, or
371 + /// re-sealed one row's plaintext under another's AAD, shows up as a mismatch
372 + /// rather than as two interchangeable blobs.
373 + fn paged_rows() -> [Vec<(&'static str, &'static str, serde_json::Value)>; 2] {
374 + [
375 + vec![
376 + ("tasks", "row-1", json!({"title": "first page, first row"})),
377 + ("tasks", "row-2", json!({"title": "first page, second row"})),
378 + ],
379 + vec![
380 + ("notes", "row-3", json!({"body": "second page, first row"})),
381 + ("notes", "row-4", json!({"body": "second page, second row"})),
382 + ],
383 + ]
384 + }
385 +
386 + /// A `/keys/rotate/entries` page: the rows sealed under `old_key`, numbered from
387 + /// `first_seq`, with the server's `has_more` verdict.
388 + fn entries_page(
389 + old_key: &[u8; 32],
390 + rows: &[(&'static str, &'static str, serde_json::Value)],
391 + first_seq: i64,
392 + has_more: bool,
393 + ) -> serde_json::Value {
394 + let entries: Vec<serde_json::Value> = rows
395 + .iter()
396 + .enumerate()
397 + .map(|(i, (table, row_id, payload))| {
398 + let ctx = synckit_client::crypto::AeadContext::entry(table, row_id);
399 + let sealed = synckit_client::crypto::encrypt_json_aad(payload, old_key, &ctx).unwrap();
400 + json!({ "seq": first_seq + i as i64, "table": table, "row_id": row_id, "data": sealed })
401 + })
402 + .collect();
403 + json!({ "entries": entries, "has_more": has_more })
404 + }
405 +
406 + /// A server that hands back two pages of work drives two re-encrypt rounds and
407 + /// two batch pushes, and every payload it gets back is sealed under the NEW key
408 + /// only.
409 + ///
410 + /// The `has_more = true` page is the point: it is what forces
411 + /// `reencrypt_batch`'s `Ok(!has_more)` to say "not done", so a client that
412 + /// stopped after the first page would leave page two readable under the old key
413 + /// forever. Both halves are asserted, the second page really was pulled and
414 + /// pushed, and the old key really is dead against all four re-encrypted rows.
415 + ///
416 + /// `pending_key` puts the rotation on the resume path so the new key is one this
417 + /// test knows and can decrypt with, rather than a fresh key only the client saw.
418 + #[tokio::test]
419 + async fn a_second_entries_page_is_pulled_re_encrypted_and_pushed() {
420 + let kit = MockKit::start().await;
421 + let old_key = synckit_client::crypto::generate_master_key();
422 + let new_key = synckit_client::crypto::generate_master_key();
423 + assert_ne!(
424 + old_key, new_key,
425 + "the two keys must differ or the old-key assertions below prove nothing"
426 + );
427 + let [page_one, page_two] = paged_rows();
428 +
429 + kit.get(KEYS_PATH)
430 + .json(json!({
431 + "encrypted_key": synckit_client::crypto::wrap_master_key(&old_key, ROTATE_PW).unwrap(),
432 + "key_version": 1,
433 + "key_id": 1,
434 + "pending_key": {
435 + "encrypted_key": synckit_client::crypto::wrap_master_key(&new_key, ROTATE_PW).unwrap(),
436 + "key_id": 2,
437 + },
438 + }))
439 + .await;
440 + kit.post(ROTATE_PATH).json(begin_body(4)).await;
441 + // Page one says has_more, page two drains. Mounted in order so the first
442 + // `once()` mock answers the first pull and the second answers the next.
443 + kit.post(ENTRIES_PATH)
444 + .once()
445 + .json(entries_page(&old_key, &page_one, 1, true))
446 + .await;
447 + kit.post(ENTRIES_PATH)
448 + .once()
449 + .json(entries_page(&old_key, &page_two, 3, false))
450 + .await;
451 + // Anything past those two pages would be a third round the loop must not run.
452 + kit.post(ENTRIES_PATH)
453 + .json(json!({ "entries": [], "has_more": false }))
454 + .await;
455 + kit.post(BATCH_PATH)
456 + .json(json!({ "updated_count": 2 }))
457 + .await;
458 + kit.post(COMPLETE_PATH).empty().await;
459 +
460 + kit.authed()
461 + .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
462 + .await
463 + .expect("a two-page rotation should complete");
464 +
465 + assert_eq!(
466 + kit.hits(ENTRIES_PATH).await,
467 + 2,
468 + "one pull per page: has_more on page one must run a second round, and page two must end it"
469 + );
470 + assert_eq!(
471 + kit.hits(BATCH_PATH).await,
472 + 2,
473 + "each page is pushed back as its own batch"
474 + );
475 +
476 + // Flatten what the client pushed and check it row by row against what it was
477 + // given, in seq order.
478 + let pushed: Vec<serde_json::Value> = kit
479 + .bodies("POST", BATCH_PATH)
480 + .await
481 + .iter()
482 + .flat_map(|body| {
483 + body["entries"]
484 + .as_array()
485 + .expect("a batch body carries entries")
486 + .clone()
487 + })
488 + .collect();
489 + let all_rows: Vec<_> = page_one.iter().chain(page_two.iter()).collect();
490 + assert_eq!(
491 + pushed.len(),
492 + all_rows.len(),
493 + "every row from both pages must come back re-encrypted"
494 + );
495 +
496 + for (i, (entry, (table, row_id, payload))) in pushed.iter().zip(&all_rows).enumerate() {
497 + let seq = i as i64 + 1;
498 + assert_eq!(
499 + entry["seq"].as_i64(),
500 + Some(seq),
501 + "row {row_id} came back under the wrong seq"
502 + );
503 + let ctx = synckit_client::crypto::AeadContext::entry(table, row_id);
504 + let opened = synckit_client::crypto::decrypt_json_aad(&entry["data"], &new_key, &ctx)
505 + .unwrap_or_else(|e| panic!("row {row_id} does not open under the new key: {e}"));
506 + assert_eq!(
507 + opened, *payload,
508 + "row {row_id} came back holding a different payload"
509 + );
510 + // The whole point of a rotation: the old key is retired. A no-op
511 + // re-encrypt would leave this opening cleanly.
512 + assert!(
513 + synckit_client::crypto::decrypt_json_aad(&entry["data"], &old_key, &ctx).is_err(),
514 + "row {row_id} still opens under the OLD key, so it was never re-encrypted"
515 + );
516 + }
517 + }
518 +
519 + // ── The straggler round cap ──
520 +
521 + /// `MAX_ROTATION_ROUNDS` from `src/client/rotation.rs`. Private there, so it is
522 + /// restated here; the assertions below pin the cap exactly, which is the only
523 + /// way the counter's arithmetic is observable at all (a counter that never
524 + /// advances and a counter that advances differ nowhere except at the cap).
525 + const MAX_ROTATION_ROUNDS: usize = 100_000;
526 +
527 + /// A wiremock responder that counts what it served.
528 + ///
529 + /// The house `MockKit` would answer this test's routes, but its `hits()` reads
530 + /// wiremock's recorded-request log, and this test provokes 200,000 requests: the
531 + /// log would hold every one of them in memory for the duration. Counting in the
532 + /// responder and turning recording off keeps the test's footprint flat.
533 + struct Counted {
534 + hits: std::sync::Arc<std::sync::atomic::AtomicUsize>,
535 + code: u16,
536 + body: serde_json::Value,
537 + }
538 +
539 + impl wiremock::Respond for Counted {
540 + fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate {
541 + self.hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
542 + ResponseTemplate::new(self.code).set_body_json(self.body.clone())
543 + }
544 + }
545 +
546 + /// A server that answers `POST /keys/rotate/complete` with 409 forever must not
547 + /// spin the straggler loop forever: `rotate_key` gives up at
548 + /// `MAX_ROTATION_ROUNDS` and returns the round-cap `Internal` error.
549 + ///
550 + /// This is the only test that can see the straggler counter at all. Every other
551 + /// rotation test either never gets a 409 or gets exactly one, and on those the
552 + /// counter's value is never read; only crossing the cap turns it into an
553 + /// observable. The exact-count assertion is deliberate: it pins both the
554 + /// increment (a counter that stalled at zero never returns) and the boundary
555 + /// (the cap fires on the round that reaches it, not one round later).
556 + #[tokio::test]
557 + async fn a_server_that_reports_stragglers_forever_stops_at_the_round_cap() {
558 + use std::sync::Arc;
559 + use std::sync::atomic::{AtomicUsize, Ordering};
560 +
561 + ensure_crypto_provider();
562 + let old_key = synckit_client::crypto::generate_master_key();
563 +
564 + // Recording off: see `Counted`.
565 + let server = wiremock::MockServer::builder()
566 + .disable_request_recording()
567 + .start()
568 + .await;
569 +
570 + let completes = Arc::new(AtomicUsize::new(0));
571 + let entries_pulls = Arc::new(AtomicUsize::new(0));
572 +
573 + wiremock::Mock::given(wiremock::matchers::method("GET"))
574 + .and(wiremock::matchers::path(KEYS_PATH))
575 + .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
576 + .mount(&server)
577 + .await;
578 + wiremock::Mock::given(wiremock::matchers::method("POST"))
579 + .and(wiremock::matchers::path(ROTATE_PATH))
580 + .respond_with(ResponseTemplate::new(200).set_body_json(begin_body(0)))
581 + .mount(&server)
582 + .await;
583 + // Nothing left to re-encrypt, so each straggler round costs one pull and
584 + // returns immediately: the loop that is being bounded is the straggler loop,
585 + // not the re-encrypt loop inside it.
586 + wiremock::Mock::given(wiremock::matchers::method("POST"))
587 + .and(wiremock::matchers::path(ENTRIES_PATH))
588 + .respond_with(Counted {
589 + hits: Arc::clone(&entries_pulls),
590 + code: 200,
591 + body: json!({ "entries": [], "has_more": false }),
592 + })
593 + .mount(&server)
594 + .await;
595 + // The stall: stragglers, always, no matter how many rounds the client runs.
596 + wiremock::Mock::given(wiremock::matchers::method("POST"))
597 + .and(wiremock::matchers::path(COMPLETE_PATH))
598 + .respond_with(Counted {
599 + hits: Arc::clone(&completes),
600 + code: 409,
601 + body: json!({ "message": "stragglers" }),
602 + })
603 + .mount(&server)
604 + .await;
605 +
606 + let client = SyncKitClient::new(SyncKitConfig {
607 + server_url: server.uri(),
608 + api_key: "test-api-key".to_string(),
609 + });
610 + let (user_id, app_id) = test_ids();
611 + client.restore_session(&fresh_token(), user_id, app_id);
612 +
613 + let err = client
614 + .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
615 + .await
616 + .expect_err("a server that never converges must not be waited on forever");
617 +
618 + // The straggler cap, not the re-encrypt cap: the two share a constant and a
619 + // variant, so the message is what tells them apart.
620 + match &err {
621 + SyncKitError::Internal(msg) => assert!(
622 + msg.contains("server kept reporting stragglers past the round cap"),
623 + "wrong give-up path: {msg}"
624 + ),
625 + other => panic!("expected the round-cap Internal error, got {other:?}"),
626 + }
627 +
628 + assert_eq!(
629 + completes.load(Ordering::Relaxed),
630 + MAX_ROTATION_ROUNDS,
631 + "the cap must fire on the round that reaches it: one completion attempt per round, no more and no fewer"
632 + );
633 + // Every round but the last re-ran the re-encrypt loop; the capped round
634 + // returns before it does.
635 + assert_eq!(
636 + entries_pulls.load(Ordering::Relaxed),
637 + MAX_ROTATION_ROUNDS,
638 + "one initial re-encrypt pass plus one per straggler round short of the cap"
639 + );
640 + }