Skip to main content

max / synckit

Pin the hold queue counts and its not-found answers HoldCounts::total, the cross-scope reads and the boolean returns from clear and requeue had no test that could tell a sum from a difference, one scope from two, or a missing row from a cleared one. Cover them in deferred.rs with a two-and-three mix and holds in two scopes, and cover the facade wrappers held_counts, held_entries and retry_held over the same seed. pending_changes_counts_unpushed now inserts three rows instead of one.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-23 22:32 UTC
Signed with PGP, not checked
Commit: 08228591fff2e166c495afd880e082b416475223
Parent: 84e06f8
2 files changed, +243 insertions, -5 deletions
M synckit-client/src/store/deferred.rs +130
@@ -576,6 +576,136 @@
576 576 );
577 577 }
578 578
579 + /// An outcome holding both kinds at once, for the count arithmetic.
580 + fn mixed_outcome(deferred: Vec<Unapplied>, rejected: Vec<Unapplied>) -> ApplyOutcome {
581 + ApplyOutcome {
582 + rejected,
583 + deferred,
584 + ..ApplyOutcome::default()
585 + }
586 + }
587 +
588 + #[test]
589 + fn hold_counts_total_adds_the_two_states() {
590 + let conn = db();
591 + // Two of one and three of the other: any arithmetic other than a sum
592 + // lands somewhere else (a difference underflows, a product gives six).
593 + let entries: Vec<PulledChange> = (0..5)
594 + .map(|i| pulled("child", &format!("c{i}"), i))
595 + .collect();
596 + settle(
597 + &conn,
598 + "",
599 + &mixed_outcome(
600 + vec![unapplied("child", "c0"), unapplied("child", "c1")],
601 + vec![
602 + unapplied("child", "c2"),
603 + unapplied("child", "c3"),
604 + unapplied("child", "c4"),
605 + ],
606 + ),
607 + &batch(&entries),
608 + &HashSet::new(),
609 + )
610 + .unwrap();
611 +
612 + let counts = counts(&conn, "").unwrap();
613 + assert_eq!(counts.deferred, 2);
614 + assert_eq!(counts.rejected, 3);
615 + assert_eq!(counts.total(), 5);
616 + assert_eq!(list(&conn, "").unwrap().len(), 5);
617 + }
618 +
619 + #[test]
620 + fn the_all_scope_reads_see_every_scope_at_once() {
621 + // A row held in a group scope is as lost as one held in the personal
622 + // scope, so the status surface reads across both. Different counts per
623 + // scope, so neither read can pass by looking at one of them twice.
624 + let conn = db();
625 + let personal = pulled("child", "p1", 1);
626 + let group = [pulled("child", "g1", 2), pulled("child", "g2", 3)];
627 + settle(
628 + &conn,
629 + "",
630 + &deferred_outcome(vec![unapplied("child", "p1")]),
631 + &batch(&[personal]),
632 + &HashSet::new(),
633 + )
634 + .unwrap();
635 + settle(
636 + &conn,
637 + "group-a",
638 + &mixed_outcome(
639 + vec![unapplied("child", "g1")],
640 + vec![unapplied("child", "g2")],
641 + ),
642 + &batch(&group),
643 + &HashSet::new(),
644 + )
645 + .unwrap();
646 +
647 + let all = counts_all(&conn).unwrap();
648 + assert_eq!(all.deferred, 2, "the personal one plus the group's");
649 + assert_eq!(all.rejected, 1);
650 + assert_eq!(all.total(), 3);
651 +
652 + let listed = list_all(&conn).unwrap();
653 + assert_eq!(listed.len(), 3);
654 + let mut scopes: Vec<&str> = listed.iter().map(|e| e.scope.as_str()).collect();
655 + scopes.sort_unstable();
656 + scopes.dedup();
657 + assert_eq!(scopes, ["", "group-a"], "both scopes are represented");
658 +
659 + // The per-scope reads still see only their own, which is what makes the
660 + // pair of reads worth having.
661 + assert_eq!(counts(&conn, "").unwrap().total(), 1);
662 + assert_eq!(list(&conn, "group-a").unwrap().len(), 2);
663 + }
664 +
665 + #[test]
666 + fn clear_and_requeue_report_whether_the_row_was_there() {
667 + // Both return the rows-affected of their statement, and a caller uses it
668 + // to tell "retried" from "there was nothing to retry".
669 + let conn = db();
670 + settle(
671 + &conn,
672 + "",
673 + &ApplyOutcome {
674 + rejected: vec![unapplied("child", "c1")],
675 + ..ApplyOutcome::default()
676 + },
677 + &batch(&[pulled("child", "c1", 7)]),
678 + &HashSet::new(),
679 + )
680 + .unwrap();
681 +
682 + assert!(
683 + !requeue(&conn, "", "child", "absent").unwrap(),
684 + "no such row id"
685 + );
686 + assert!(
687 + !requeue(&conn, "", "other-scope", "c1").unwrap(),
688 + "no such table"
689 + );
690 + assert!(
691 + !clear(&conn, "elsewhere", "child", "c1").unwrap(),
692 + "no such scope"
693 + );
694 + assert_eq!(
695 + counts(&conn, "").unwrap().total(),
696 + 1,
697 + "none of that touched the held row"
698 + );
699 +
700 + assert!(requeue(&conn, "", "child", "c1").unwrap());
701 + assert!(clear(&conn, "", "child", "c1").unwrap());
702 + assert!(
703 + !clear(&conn, "", "child", "c1").unwrap(),
704 + "the second clear finds nothing left"
705 + );
706 + assert_eq!(counts(&conn, "").unwrap().total(), 0);
707 + }
708 +
579 709 #[test]
580 710 fn a_repeat_failure_replaces_the_payload_without_duplicating_the_row() {
581 711 let conn = db();
M synckit-client/src/store/facade.rs +113 -5
@@ -828,12 +828,120 @@
828 828 async fn pending_changes_counts_unpushed() {
829 829 let dir = tempdir();
830 830 let db = make_db(&dir.join("a.db"));
831 - db.open()
832 - .unwrap()
833 - .execute("INSERT INTO note (id,name) VALUES ('r','v')", [])
834 - .unwrap();
831 + let conn = db.open().unwrap();
832 + // Three rows, so a count that reported "any" or "one" would be wrong
833 + // rather than accidentally right.
834 + for id in ["r1", "r2", "r3"] {
835 + conn.execute("INSERT INTO note (id,name) VALUES (?1,'v')", [id])
836 + .unwrap();
837 + }
838 + drop(conn);
835 839 let log = Arc::new(Mutex::new(Vec::new()));
836 840 let s = store(db, Fake::new(0xA, log));
837 - assert_eq!(s.pending_changes().await.unwrap(), 1);
841 + assert_eq!(s.pending_changes().await.unwrap(), 3);
842 + }
843 +
844 + /// Hold one deferred row in the personal scope and one rejected row in a
845 + /// group scope, through the same path a failed apply takes.
846 + fn seed_holds(db: &DbSource) {
847 + use super::super::apply::{ApplyOutcome, Unapplied};
848 + use crate::types::{ChangeOp, hlc_legacy_floor};
849 +
850 + let conn = db.open().unwrap();
851 + let pulled = |row_id: &str| PulledChange {
852 + storage_version: None,
853 + entry: ChangeEntry {
854 + table: "note".into(),
855 + op: ChangeOp::Insert,
856 + row_id: row_id.into(),
857 + timestamp: Utc::now(),
858 + hlc: hlc_legacy_floor(),
859 + data: Some(serde_json::json!({"id": row_id})),
860 + extra: serde_json::Map::default(),
861 + },
862 + device_id: DeviceId::nil(),
863 + seq: 1,
864 + };
865 + let unapplied = |row_id: &str| Unapplied {
866 + table: "note".into(),
867 + row_id: row_id.into(),
868 + cause: "constraint violation".into(),
869 + };
870 +
871 + for (scope, row_id, outcome) in [
872 + (
873 + "",
874 + "held-personal",
875 + ApplyOutcome {
876 + deferred: vec![unapplied("held-personal")],
877 + ..ApplyOutcome::default()
878 + },
879 + ),
880 + (
881 + "group-a",
882 + "held-group",
883 + ApplyOutcome {
884 + rejected: vec![unapplied("held-group")],
885 + ..ApplyOutcome::default()
886 + },
887 + ),
888 + ] {
889 + let change = pulled(row_id);
890 + let batch: HashMap<(String, String), PulledChange> =
891 + [(("note".to_string(), row_id.to_string()), change)]
892 + .into_iter()
893 + .collect();
894 + deferred::settle(&conn, scope, &outcome, &batch, &HashSet::new()).unwrap();
895 + }
896 + }
897 +
898 + #[tokio::test]
899 + async fn held_reads_span_every_scope() {
900 + let dir = tempdir();
901 + let db = make_db(&dir.join("a.db"));
902 + seed_holds(&db);
903 + let log = Arc::new(Mutex::new(Vec::new()));
904 + let s = store(db, Fake::new(0xA, log));
905 +
906 + let counts = s.held_counts().await.unwrap();
907 + assert_eq!(counts.deferred, 1);
908 + assert_eq!(counts.rejected, 1);
909 + assert_eq!(counts.total(), 2);
910 +
911 + let entries = s.held_entries().await.unwrap();
912 + assert_eq!(entries.len(), 2, "the group scope's row counts too");
913 + let mut scopes: Vec<&str> = entries.iter().map(|e| e.scope.as_str()).collect();
914 + scopes.sort_unstable();
915 + assert_eq!(scopes, ["", "group-a"]);
916 + // The payload rides along so a caller can name the row for its user.
917 + let group_entry = entries.iter().find(|e| e.scope == "group-a").unwrap();
918 + assert_eq!(group_entry.row_id, "held-group");
919 + assert_eq!(group_entry.cause, "constraint violation");
920 + assert_eq!(group_entry.payload.as_ref().unwrap()["id"], "held-group");
921 + }
922 +
923 + #[tokio::test]
924 + async fn retry_held_reports_whether_there_was_anything_to_retry() {
925 + let dir = tempdir();
926 + let db = make_db(&dir.join("a.db"));
927 + seed_holds(&db);
928 + let log = Arc::new(Mutex::new(Vec::new()));
929 + let s = store(db, Fake::new(0xA, log));
930 +
931 + assert!(
932 + !s.retry_held("group-a", "note", "never-held").await.unwrap(),
933 + "no such row"
934 + );
935 + assert!(
936 + !s.retry_held("", "note", "held-group").await.unwrap(),
937 + "right row, wrong scope"
938 + );
939 + assert!(s.retry_held("group-a", "note", "held-group").await.unwrap());
940 +
941 + // Requeueing moves the row out of rejected rather than adding one.
942 + let counts = s.held_counts().await.unwrap();
943 + assert_eq!(counts.deferred, 2);
944 + assert_eq!(counts.rejected, 0);
945 + assert_eq!(counts.total(), 2);
838 946 }
839 947 }