Skip to main content

max / synckit

Observe the push drain past its batch limit and the pull counters The push loop's batch terminator was only ever exercised with two rows, so a break after the first batch would have lost every row past 500 and still reported a clean sync. Seed 1001 pending changelog rows, drain once, and assert the count returned, the rows the fake server received, their distinctness, and that nothing is left unpushed. A second drain sends nothing. Add a pull over a mix that produces all four PullOutcome counters at once, with a distinct count for each, and assert cleanup_changelog's return against the rows it actually deleted, leaving unpushed rows alone.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-23 22:17 UTC
Signed with PGP, not checked
Commit: fec69d949fe235a462ccc2cd806097af40b7f9fa
Parent: 1296f2a
1 file changed, +221 insertions, -0 deletions
M synckit-client/src/store/sync.rs +221
@@ -1288,6 +1288,227 @@
1288 1288 assert!(dropped >= 1);
1289 1289 }
1290 1290
1291 + #[tokio::test]
1292 + async fn push_drains_past_the_batch_limit() {
1293 + let dir = tempdir();
1294 + let (db, node) = device(&dir.join("big.db"), 21);
1295 + let server = FakeServer::default();
1296 +
1297 + // One more than two full batches, so the drain has to come back for a
1298 + // third: a loop that stops after the first batch loses 501 rows and
1299 + // still reports a clean sync.
1300 + let total = PUSH_BATCH_LIMIT * 2 + 1;
1301 + {
1302 + let conn = db.open().unwrap();
1303 + conn.execute("BEGIN", []).unwrap();
1304 + for i in 0..total {
1305 + conn.execute(
1306 + "INSERT INTO note (id, name) VALUES (?1, 'n')",
1307 + [format!("r{i:04}")],
1308 + )
1309 + .unwrap();
1310 + }
1311 + conn.execute("COMMIT", []).unwrap();
1312 + let pending: i64 = conn
1313 + .query_row(
1314 + "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
1315 + [],
1316 + |r| r.get(0),
1317 + )
1318 + .unwrap();
1319 + assert_eq!(pending as usize, total, "the triggers logged every insert");
1320 + }
1321 +
1322 + let pushed = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
1323 + .await
1324 + .unwrap();
1325 + assert_eq!(pushed as usize, total);
1326 +
1327 + {
1328 + let log = server.log.lock().unwrap();
1329 + assert_eq!(log.len(), total, "every row reached the server");
1330 + let mut seen: Vec<String> = log.iter().map(|(_, e)| e.row_id.clone()).collect();
1331 + seen.sort();
1332 + seen.dedup();
1333 + assert_eq!(
1334 + seen.len(),
1335 + total,
1336 + "no row was sent twice in place of another"
1337 + );
1338 + }
1339 +
1340 + let left: i64 = db
1341 + .open()
1342 + .unwrap()
1343 + .query_row(
1344 + "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
1345 + [],
1346 + |r| r.get(0),
1347 + )
1348 + .unwrap();
1349 + assert_eq!(left, 0, "every row is marked pushed");
1350 +
1351 + // A second drain has nothing to do and sends nothing.
1352 + let again = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
1353 + .await
1354 + .unwrap();
1355 + assert_eq!(again, 0);
1356 + assert_eq!(server.log.lock().unwrap().len(), total);
1357 + }
1358 +
1359 + /// A schema wide enough for one pull to produce all four outcomes at once.
1360 + fn mix_schema() -> SyncSchema {
1361 + SyncSchema::new(vec![
1362 + SyncTable::full("parent", &["id", "name"]),
1363 + SyncTable::full("child", &["id", "parent_id"]),
1364 + SyncTable::full("cfg", &["key", "value"])
1365 + .pk(&["key"])
1366 + .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
1367 + ])
1368 + }
1369 +
1370 + fn mix_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
1371 + let db = DbSource::path(path);
1372 + let conn = db.open().unwrap();
1373 + conn.execute_batch(
1374 + "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
1375 + CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));
1376 + CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
1377 + )
1378 + .unwrap();
1379 + conn.execute_batch(&mix_schema().migration_sql()).unwrap();
1380 + (db, DeviceId::new(uuid::Uuid::from_u128(n)))
1381 + }
1382 +
1383 + /// Put an arbitrary entry on the fake server, including one with no payload.
1384 + fn serve_entry(server: &FakeServer, entry: ChangeEntry) {
1385 + server
1386 + .log
1387 + .lock()
1388 + .unwrap()
1389 + .push((DeviceId::new(uuid::Uuid::from_u128(0xAA)), entry));
1390 + }
1391 +
1392 + #[tokio::test]
1393 + async fn pull_reports_all_four_outcome_counters() {
1394 + let dir = tempdir();
1395 + let (db, node) = mix_device(&dir.join("mix.db"), 22);
1396 + let server = FakeServer::default();
1397 +
1398 + // applied x4
1399 + for i in 1..=4 {
1400 + serve(
1401 + &server,
1402 + "parent",
1403 + &format!("p{i}"),
1404 + serde_json::json!({"id": format!("p{i}"), "name": "p"}),
1405 + );
1406 + }
1407 + // filtered x3: cfg's include predicate excludes the sync_ prefix.
1408 + for k in ["sync_a", "sync_b", "sync_c"] {
1409 + serve(
1410 + &server,
1411 + "cfg",
1412 + k,
1413 + serde_json::json!({"key": k, "value": "v"}),
1414 + );
1415 + }
1416 + // rejected x1: no payload, so the same bytes always fail.
1417 + serve_entry(
1418 + &server,
1419 + ChangeEntry {
1420 + table: "parent".into(),
1421 + op: ChangeOp::Insert,
1422 + row_id: "p9".into(),
1423 + timestamp: Utc::now(),
1424 + hlc: crate::types::hlc_legacy_floor(),
1425 + data: None,
1426 + extra: serde_json::Map::default(),
1427 + },
1428 + );
1429 + // deferred x2: a parent that is not in this page and never will be.
1430 + for i in 1..=2 {
1431 + serve(
1432 + &server,
1433 + "child",
1434 + &format!("c{i}"),
1435 + serde_json::json!({"id": format!("c{i}"), "parent_id": "absent"}),
1436 + );
1437 + }
1438 +
1439 + let out = pull_scope(&db, &server, &mix_schema(), node, SyncScope::Personal)
1440 + .await
1441 + .unwrap();
1442 +
1443 + assert_eq!(out.applied, 4);
1444 + assert_eq!(out.filtered, 3);
1445 + assert_eq!(out.rejected, 1);
1446 + assert_eq!(out.deferred, 2);
1447 + assert!(out.changed_tables.contains("parent"));
1448 + assert!(
1449 + !out.changed_tables.contains("child"),
1450 + "nothing landed in child"
1451 + );
1452 + assert_eq!(row_count(&db, "parent"), 4);
1453 + assert_eq!(row_count(&db, "child"), 0);
1454 + assert_eq!(row_count(&db, "cfg"), 0);
1455 + }
1456 +
1457 + #[test]
1458 + fn cleanup_changelog_returns_what_it_deleted() {
1459 + let dir = tempdir();
1460 + let (da, _) = device(&dir.join("cleanup.db"), 23);
1461 + let conn = da.open().unwrap();
1462 + conn.execute("DELETE FROM sync_changelog", []).unwrap();
1463 +
1464 + // 4 pushed and old (deletable), 2 pushed and recent, 3 unpushed and old.
1465 + let insert = |row_id: &str, pushed: i64, ts: &str| {
1466 + conn.execute(
1467 + "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed, timestamp) \
1468 + VALUES ('note','INSERT',?1,'{}',?2,?3)",
1469 + rusqlite::params![row_id, pushed, ts],
1470 + )
1471 + .unwrap();
1472 + };
1473 + let old = "2000-01-01T00:00:00.000Z";
1474 + let now = "2999-01-01T00:00:00.000Z";
1475 + for i in 0..4 {
1476 + insert(&format!("old{i}"), 1, old);
1477 + }
1478 + for i in 0..2 {
1479 + insert(&format!("new{i}"), 1, now);
1480 + }
1481 + for i in 0..3 {
1482 + insert(&format!("unp{i}"), 0, old);
1483 + }
1484 +
1485 + let before: i64 = conn
1486 + .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
1487 + .unwrap();
1488 + let removed = cleanup_changelog(&conn).unwrap();
1489 + let after: i64 = conn
1490 + .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
1491 + .unwrap();
1492 +
1493 + assert_eq!(removed, 4, "only the old pushed rows go");
1494 + assert_eq!(
1495 + removed,
1496 + (before - after) as u64,
1497 + "the returned count is the number of rows actually deleted"
1498 + );
1499 + let unpushed: i64 = conn
1500 + .query_row(
1501 + "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
1502 + [],
1503 + |r| r.get(0),
1504 + )
1505 + .unwrap();
1506 + assert_eq!(unpushed, 3, "an unpushed row is never pruned by age");
1507 +
1508 + // Nothing left to prune, so a second pass reports zero.
1509 + assert_eq!(cleanup_changelog(&conn).unwrap(), 0);
1510 + }
1511 +
1291 1512 // minimal temp-dir helper (no external dep)
1292 1513 fn tempdir() -> std::path::PathBuf {
1293 1514 let mut p = std::env::temp_dir();