Skip to main content

max / synckit

Rotate group keys on removal, and stash what LWW throws away Two gaps in the group/conflict layer, both losses that left no trace. Rotation. remove_member revoked server access and left the GCK alone, so a removed member holding a copy could still read anything written after. It now drives rotate_group_key: mint a fresh GCK, seal it to every remaining member's stored pubkey, post the batch. The server treats that batch as the new membership, so removal and re-key commit together. revoke_member_without_rekey keeps the old behaviour for when a member's stored pubkey cannot be read and revoking now beats waiting. Rotation does not re-encrypt history, so the GCK cache is keyed by (group, generation) and group_pull_rich resolves each entry under the generation it was sealed with, falling back to the caller's current one for a server that does not stamp them. group_grant_at fetches one generation; group_grant still returns the newest, which is what a writer wants. Stash. LWW always discards one side and the bytes were simply gone. sync_conflict_stash keeps them, local-only: absent from every manifest, so never group-scoped and never pushed, since pushing it would put one device's rejected plaintext into a shared log. Three discard sites, and the third is the one worth naming: the committed-HLC gate drops a superseded remote change without ever building a ConflictPair, so it is invisible to resolve_lww. CleanChanges::gated_at gained an on_superseded sink for it and stays pure. Not stashed, deliberately: byte-identical payloads (every echo would qualify), clock-poisoned drops (refused as hostile, not outvoted, and stashing them is a way to fill the table), merges, and ServerOrder. A stash write failing logs rather than failing the sync. Bounded at 1000 rows. Nothing in the engine reads the stash back; how a loss surfaces is the consuming app's call.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-03 18:47 UTC
Signed with PGP, not checked
Commit: 5e88f2d1092cd0b2020ff8ef1c1dbc0a170a4d0d
Parent: 5c0abd8
11 files changed, +1439 insertions, -84 deletions
M synckit-client/src/conflict.rs +35 -24
@@ -57,11 +57,12 @@
57 57 /// [`gated_at`](Self::gated_at) to pin `now` across a whole sync batch (or in
58 58 /// tests) so every change in the batch is classified against the same instant.
59 59 pub fn gated(self, committed_hlc: impl Fn(&str, &str) -> Option<Hlc>) -> Vec<ChangeEntry> {
60 - self.gated_at(Utc::now(), committed_hlc)
60 + self.gated_at(Utc::now(), committed_hlc, |_, _| {})
61 61 }
62 62
63 63 /// [`gated`](Self::gated) with an explicit `now`, so the clock-poisoning
64 - /// classification is deterministic (testable) and consistent across a batch.
64 + /// classification is deterministic (testable) and consistent across a batch,
65 + /// and a sink for what the committed-HLC gate discards.
65 66 ///
66 67 /// The poisoning threshold is still fundamentally wall-clock-relative: two
67 68 /// devices whose clocks straddle the drift boundary by less than their skew
@@ -69,38 +70,48 @@
69 70 /// removes the *within-device* drift between the batch's first and last change;
70 71 /// the residual cross-device boundary window is the bounded, documented
71 72 /// tradeoff of [`MAX_HLC_DRIFT_MS`] (see [`resolve_lww_at`]).
73 + ///
74 + /// `on_superseded(dropped, committed)` fires for each change the committed-HLC
75 + /// gate rejects: a remote edit older than what this device already applied and
76 + /// pushed. That is a lost edit exactly as much as a resolved conflict is, but
77 + /// it never becomes a [`ConflictPair`] and so is invisible to
78 + /// [`resolve_lww`], which is what made it the easiest kind of data loss to
79 + /// miss. A **clock-poisoned** drop deliberately does not fire it: that entry
80 + /// is rejected as hostile or broken, not outvoted, and stashing it would
81 + /// hand an attacker a way to fill the stash. The function stays pure; the
82 + /// caller decides what persistence means.
72 83 pub fn gated_at(
73 84 self,
74 85 now: DateTime<Utc>,
75 86 committed_hlc: impl Fn(&str, &str) -> Option<Hlc>,
87 + mut on_superseded: impl FnMut(&PulledChange, &Hlc),
76 88 ) -> Vec<ChangeEntry> {
77 - self.0
78 - .into_iter()
79 - .filter(|p| {
80 - // A clean change is applied without passing through `resolve_lww`,
81 - // so the clock-poisoning guard must also live here: never apply an
82 - // entry whose wall clock is implausibly far in the future.
83 - if is_clock_poisoned(&p.entry.hlc, now) {
84 - tracing::warn!(
85 - table = %p.entry.table,
86 - wall_ms = p.entry.hlc.wall_ms,
87 - "dropping clean change with implausibly-future HLC"
88 - );
89 - return false;
90 - }
91 - let apply = committed_hlc(&p.entry.table, &p.entry.row_id)
92 - .is_none_or(|committed| p.entry.hlc > committed);
93 - if !apply {
89 + let mut kept = Vec::with_capacity(self.0.len());
90 + for p in self.0 {
91 + // A clean change is applied without passing through `resolve_lww`,
92 + // so the clock-poisoning guard must also live here: never apply an
93 + // entry whose wall clock is implausibly far in the future.
94 + if is_clock_poisoned(&p.entry.hlc, now) {
95 + tracing::warn!(
96 + table = %p.entry.table,
97 + wall_ms = p.entry.hlc.wall_ms,
98 + "dropping clean change with implausibly-future HLC"
99 + );
100 + continue;
101 + }
102 + match committed_hlc(&p.entry.table, &p.entry.row_id) {
103 + Some(committed) if p.entry.hlc <= committed => {
94 104 tracing::debug!(
95 105 table = %p.entry.table,
96 106 row_id = %p.entry.row_id,
97 107 "clean change gated out: older than the committed HLC"
98 108 );
109 + on_superseded(&p, &committed);
99 110 }
100 - apply
101 - })
102 - .map(|p| p.entry)
103 - .collect()
111 + _ => kept.push(p.entry),
112 + }
113 + }
114 + kept
104 115 }
105 116
106 117 /// The `(table, row_id)` addresses of the clean changes, so a caller can
@@ -359,7 +370,7 @@
359 370
360 371 /// [`canonical_value`] for an optional payload; `None` (a delete) canonicalizes
361 372 /// to empty bytes.
362 - fn canonical_payload(data: Option<&serde_json::Value>) -> Vec<u8> {
373 + pub(crate) fn canonical_payload(data: Option<&serde_json::Value>) -> Vec<u8> {
363 374 data.map(canonical_value).unwrap_or_default()
364 375 }
365 376
M synckit-client/src/types.rs +20
@@ -268,6 +268,18 @@
268 268 pub created_at: DateTime<Utc>,
269 269 }
270 270
271 + /// A member's identity public key, as returned by
272 + /// [`SyncKitClient::list_member_pubkeys`](crate::SyncKitClient::list_member_pubkeys).
273 + /// The admin seals a rotated GCK to each of these.
274 + #[derive(Debug, Clone, Deserialize)]
275 + #[non_exhaustive]
276 + pub struct GroupMemberPubkey {
277 + /// The member's account id.
278 + pub user_id: UserId,
279 + /// Their X25519 identity public key, base64.
280 + pub pubkey: String,
281 + }
282 +
271 283 /// A member of a group, as returned by
272 284 /// [`SyncKitClient::list_members`](crate::SyncKitClient::list_members). Grants are
273 285 /// not included, each member fetches only their own.
@@ -403,6 +415,14 @@
403 415 /// Which encryption key was used. None means key_id 1 (pre-rotation).
404 416 #[serde(default)]
405 417 pub key_id: Option<i32>,
418 + /// For a group entry, the GCK generation its ciphertext is sealed under. A
419 + /// rotation bumps the group's generation without re-encrypting what came
420 + /// before, so a pull can span generations and each row says which key opens
421 + /// it. `None` on personal entries, and on group entries from a server
422 + /// predating per-generation grants, where the caller falls back to the
423 + /// group's current generation.
424 + #[serde(default)]
425 + pub gck_version: Option<i32>,
406 426 }
407 427
408 428 // ── Filtered pull ──
M synckit-client/tests/integration.rs +497
@@ -3658,3 +3658,544 @@
3658 3658 std::fs::remove_file(&file).ok();
3659 3659 }
3660 3660 }
3661 +
3662 + // ── Group key rotation ──
3663 +
3664 + /// Rotation is the removal primitive: the batch an admin posts becomes the new
3665 + /// membership, so the server drops anyone absent from it and re-keys in the same
3666 + /// transaction. These tests pin the batch the client builds, because everything
3667 + /// the server can enforce depends on the client getting that batch right.
3668 + mod group_rotation {
3669 + use super::*;
3670 + use synckit_client::{
3671 + GroupId, IdentityKeypair, IdentityPublicKey, generate_group_key, open_gck_grant,
3672 + seal_gck_to_member,
3673 + };
3674 + use wiremock::matchers::path_regex;
3675 +
3676 + /// A member we control both halves of, so a grant sealed to them can be
3677 + /// opened and checked rather than merely counted.
3678 + struct Member {
3679 + user_id: UserId,
3680 + keypair: IdentityKeypair,
3681 + }
3682 +
3683 + impl Member {
3684 + fn new() -> Self {
3685 + Self {
3686 + user_id: UserId::new(Uuid::new_v4()),
3687 + keypair: IdentityKeypair::generate(),
3688 + }
3689 + }
3690 +
3691 + fn pubkey_json(&self) -> serde_json::Value {
3692 + json!({
3693 + "user_id": self.user_id,
3694 + "pubkey": self.keypair.public_key().to_base64(),
3695 + })
3696 + }
3697 + }
3698 +
3699 + /// The client whose master key seeds the admin identity, plus that identity.
3700 + fn admin_client(server: &MockServer) -> (SyncKitClient, IdentityKeypair) {
3701 + let client = authed_client(server);
3702 + let master = synckit_client::crypto::generate_master_key();
3703 + let identity = IdentityKeypair::from_master_key(&master);
3704 + client.set_master_key_raw(master);
3705 + (client, identity)
3706 + }
3707 +
3708 + /// Mount the two reads a rotation makes: the admin's own grant (for the
3709 + /// current generation) and the member pubkey list (the re-seal inputs).
3710 + async fn mount_reads(
3711 + server: &MockServer,
3712 + group_id: GroupId,
3713 + gck: &[u8; 32],
3714 + admin: &IdentityKeypair,
3715 + admin_id: UserId,
3716 + version: i32,
3717 + members: &[&Member],
3718 + ) {
3719 + let sealed = seal_gck_to_member(gck, &admin.public_key(), &group_id.to_string(), version)
3720 + .expect("seal admin grant");
3721 + Mock::given(method("GET"))
3722 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
3723 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3724 + "sealed_gck": sealed,
3725 + "gck_version": version,
3726 + })))
3727 + .mount(server)
3728 + .await;
3729 +
3730 + let mut pubkeys = vec![json!({
3731 + "user_id": admin_id,
3732 + "pubkey": admin.public_key().to_base64(),
3733 + })];
3734 + pubkeys.extend(members.iter().map(|m| m.pubkey_json()));
3735 + Mock::given(method("GET"))
3736 + .and(path(format!("/api/v1/sync/groups/{group_id}/pubkeys")))
3737 + .respond_with(ResponseTemplate::new(200).set_body_json(pubkeys))
3738 + .mount(server)
3739 + .await;
3740 + }
3741 +
3742 + async fn mount_rotate(server: &MockServer) {
3743 + Mock::given(method("POST"))
3744 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/rotate$"))
3745 + .respond_with(ResponseTemplate::new(204))
3746 + .mount(server)
3747 + .await;
3748 + }
3749 +
3750 + /// The body the client POSTed to `/rotate`.
3751 + async fn posted_batch(server: &MockServer) -> serde_json::Value {
3752 + let reqs = server.received_requests().await.expect("requests");
3753 + let rotate = reqs
3754 + .iter()
3755 + .find(|r| r.url.path().ends_with("/rotate"))
3756 + .expect("a rotate request was sent");
3757 + serde_json::from_slice(&rotate.body).expect("rotate body is JSON")
3758 + }
3759 +
3760 + #[tokio::test]
3761 + async fn removing_a_member_rekeys_and_reseals_to_everyone_who_stays() {
3762 + let server = MockServer::start().await;
3763 + let (client, admin_identity) = admin_client(&server);
3764 + let (admin_id, _) = test_ids();
3765 + let group_id = GroupId::new(Uuid::new_v4());
3766 + let old_gck = generate_group_key();
3767 +
3768 + let bob = Member::new();
3769 + let carol = Member::new();
3770 + mount_reads(
3771 + &server,
3772 + group_id,
3773 + &old_gck,
3774 + &admin_identity,
3775 + admin_id,
3776 + 7,
3777 + &[&bob, &carol],
3778 + )
3779 + .await;
3780 + mount_rotate(&server).await;
3781 +
3782 + client
3783 + .remove_member(group_id, carol.user_id)
3784 + .await
3785 + .expect("remove member");
3786 +
3787 + let batch = posted_batch(&server).await;
3788 + assert_eq!(
3789 + batch["gck_version"], 8,
3790 + "the generation must advance past the one our grant reports"
3791 + );
3792 +
3793 + let grants = batch["grants"].as_array().expect("grants array");
3794 + assert_eq!(grants.len(), 2, "admin and bob, not carol: {grants:?}");
3795 + let recipients: Vec<&str> = grants
3796 + .iter()
3797 + .map(|g| g["user_id"].as_str().expect("user_id"))
3798 + .collect();
3799 + assert!(recipients.contains(&admin_id.to_string().as_str()));
3800 + assert!(recipients.contains(&bob.user_id.to_string().as_str()));
3801 + assert!(
3802 + !recipients.contains(&carol.user_id.to_string().as_str()),
3803 + "the removed member must not be re-granted"
3804 + );
3805 +
3806 + // The grants are real seals of one new key, not placeholders: Bob's opens,
3807 + // and what comes out is neither the old GCK nor something private to the
3808 + // admin's copy.
3809 + let bobs = grants
3810 + .iter()
3811 + .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
3812 + .expect("bob's grant");
3813 + let new_gck = open_gck_grant(
3814 + bobs["sealed_gck"].as_str().expect("sealed_gck"),
3815 + &bob.keypair,
3816 + &group_id.to_string(),
3817 + 8,
3818 + )
3819 + .expect("bob opens his grant");
3820 + assert_ne!(new_gck, old_gck, "rotation must mint a fresh key");
3821 +
3822 + let admins = grants
3823 + .iter()
3824 + .find(|g| g["user_id"].as_str() == Some(&admin_id.to_string()))
3825 + .expect("admin's grant");
3826 + let admin_copy = open_gck_grant(
3827 + admins["sealed_gck"].as_str().expect("sealed_gck"),
3828 + &admin_identity,
3829 + &group_id.to_string(),
3830 + 8,
3831 + )
3832 + .expect("admin opens their own grant");
3833 + assert_eq!(
3834 + admin_copy, new_gck,
3835 + "every member must be sealed the same new key"
3836 + );
3837 + }
3838 +
3839 + #[tokio::test]
3840 + async fn a_grant_cannot_be_opened_by_the_member_it_was_not_sealed_to() {
3841 + let server = MockServer::start().await;
3842 + let (client, admin_identity) = admin_client(&server);
3843 + let (admin_id, _) = test_ids();
3844 + let group_id = GroupId::new(Uuid::new_v4());
3845 +
3846 + let bob = Member::new();
3847 + let carol = Member::new();
3848 + mount_reads(
3849 + &server,
3850 + group_id,
3851 + &generate_group_key(),
3852 + &admin_identity,
3853 + admin_id,
3854 + 1,
3855 + &[&bob, &carol],
3856 + )
3857 + .await;
3858 + mount_rotate(&server).await;
3859 +
3860 + client
3861 + .rotate_group_key(group_id, &[])
3862 + .await
3863 + .expect("rotate without removing anyone");
3864 +
3865 + let batch = posted_batch(&server).await;
3866 + let bobs = batch["grants"]
3867 + .as_array()
3868 + .expect("grants")
3869 + .iter()
3870 + .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
3871 + .expect("bob's grant")["sealed_gck"]
3872 + .as_str()
3873 + .expect("sealed_gck")
3874 + .to_string();
3875 +
3876 + assert!(
3877 + open_gck_grant(&bobs, &carol.keypair, &group_id.to_string(), 2).is_err(),
3878 + "a grant sealed to bob must not open under carol's key"
3879 + );
3880 + }
3881 +
3882 + #[tokio::test]
3883 + async fn an_empty_removal_set_rekeys_without_dropping_anyone() {
3884 + let server = MockServer::start().await;
3885 + let (client, admin_identity) = admin_client(&server);
3886 + let (admin_id, _) = test_ids();
3887 + let group_id = GroupId::new(Uuid::new_v4());
3888 +
3889 + let bob = Member::new();
3890 + mount_reads(
3891 + &server,
3892 + group_id,
3893 + &generate_group_key(),
3894 + &admin_identity,
3895 + admin_id,
3896 + 3,
3897 + &[&bob],
3898 + )
3899 + .await;
3900 + mount_rotate(&server).await;
3901 +
3902 + client
3903 + .rotate_group_key(group_id, &[])
3904 + .await
3905 + .expect("rekey after a suspected compromise");
3906 +
3907 + let batch = posted_batch(&server).await;
3908 + assert_eq!(batch["gck_version"], 4);
3909 + assert_eq!(
3910 + batch["grants"].as_array().expect("grants").len(),
3911 + 2,
3912 + "a bare re-key keeps the whole membership"
3913 + );
3914 + }
3915 +
3916 + #[tokio::test]
3917 + async fn a_member_pubkey_the_client_cannot_parse_aborts_the_rotation() {
3918 + let server = MockServer::start().await;
3919 + let (client, admin_identity) = admin_client(&server);
3920 + let (admin_id, _) = test_ids();
3921 + let group_id = GroupId::new(Uuid::new_v4());
3922 +
3923 + let sealed = seal_gck_to_member(
3924 + &generate_group_key(),
3925 + &admin_identity.public_key(),
3926 + &group_id.to_string(),
3927 + 1,
3928 + )
3929 + .expect("seal admin grant");
3930 + Mock::given(method("GET"))
3931 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
3932 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
3933 + "sealed_gck": sealed,
3934 + "gck_version": 1,
3935 + })))
3936 + .mount(&server)
3937 + .await;
3938 + Mock::given(method("GET"))
3939 + .and(path(format!("/api/v1/sync/groups/{group_id}/pubkeys")))
3940 + .respond_with(ResponseTemplate::new(200).set_body_json(json!([
3941 + { "user_id": admin_id, "pubkey": admin_identity.public_key().to_base64() },
3942 + { "user_id": Uuid::new_v4(), "pubkey": "not-a-key" },
3943 + ])))
3944 + .mount(&server)
3945 + .await;
3946 + mount_rotate(&server).await;
3947 +
3948 + client
3949 + .rotate_group_key(group_id, &[])
3950 + .await
3951 + .expect_err("an unreadable member key must not produce a partial rotation");
3952 +
3953 + let reqs = server.received_requests().await.expect("requests");
3954 + assert_eq!(
3955 + reqs.iter()
3956 + .filter(|r| r.url.path().ends_with("/rotate"))
3957 + .count(),
3958 + 0,
3959 + "nothing may be posted when the batch could not be built in full"
3960 + );
3961 + }
3962 +
3963 + /// `IdentityPublicKey` round-trips through the wire form the pubkey list uses.
3964 + /// If this ever stops holding, every rotation silently degrades to the error
3965 + /// path above.
3966 + #[test]
3967 + fn member_pubkeys_round_trip_through_base64() {
3968 + let identity = IdentityKeypair::generate();
3969 + let encoded = identity.public_key().to_base64();
3970 + let decoded = IdentityPublicKey::from_base64(&encoded).expect("round-trip");
3971 + assert_eq!(decoded.as_bytes(), identity.public_key().as_bytes());
3972 + }
3973 + }
3974 +
3975 + /// The client half of history-survives-rotation: one pull can span GCK
3976 + /// generations, and each entry is opened under the key it was sealed with.
3977 + mod group_generations {
3978 + use super::*;
3979 + use synckit_client::{
3980 + ChangeEntry, GroupId, IdentityKeypair, generate_group_key, seal_gck_to_member,
3981 + };
3982 + use wiremock::matchers::{path_regex, query_param};
3983 +
3984 + fn change(row: &str, title: &str) -> ChangeEntry {
3985 + ChangeEntry {
3986 + table: "tasks".to_string(),
3987 + op: ChangeOp::Insert,
3988 + row_id: row.to_string(),
3989 + timestamp: Utc::now(),
3990 + hlc: Hlc::zero(DeviceId::nil()),
3991 + data: Some(json!({ "title": title })),
3992 + extra: serde_json::Map::default(),
3993 + }
3994 + }
3995 +
3996 + /// Push one change under `gck` and return the ciphertext the client produced,
3997 + /// so it can be served straight back in a pull. Going through the real push
3998 + /// path keeps the fixture honest: no test-local reimplementation of the AAD
3999 + /// binding to drift from the one the client uses.
4000 + async fn sealed_entry(
4001 + client: &SyncKitClient,
4002 + server: &MockServer,
4003 + group_id: GroupId,
4004 + gck: &[u8; 32],
4005 + device: DeviceId,
4006 + row: &str,
4007 + title: &str,
4008 + ) -> serde_json::Value {
4009 + let before = server.received_requests().await.expect("requests").len();
4010 + client
4011 + .group_push(group_id, gck, device, vec![change(row, title)])
4012 + .await
4013 + .expect("group push");
4014 + let reqs = server.received_requests().await.expect("requests");
4015 + let pushed = reqs[before..]
4016 + .iter()
4017 + .find(|r| r.url.path().ends_with("/push"))
4018 + .expect("a push was sent");
4019 + let body: serde_json::Value = serde_json::from_slice(&pushed.body).expect("push body");
4020 + body["changes"][0].clone()
4021 + }
4022 +
4023 + #[tokio::test]
4024 + async fn a_pull_spanning_two_generations_opens_each_under_its_own_key() {
4025 + let server = MockServer::start().await;
4026 + let client = authed_client(&server);
4027 + let master = synckit_client::crypto::generate_master_key();
4028 + let identity = IdentityKeypair::from_master_key(&master);
4029 + client.set_master_key_raw(master);
4030 +
4031 + let group_id = GroupId::new(Uuid::new_v4());
4032 + let device = DeviceId::new(Uuid::new_v4());
4033 + let group_ref = group_id.to_string();
4034 + let gck_v1 = generate_group_key();
4035 + let gck_v2 = generate_group_key();
4036 +
4037 + Mock::given(method("POST"))
4038 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
4039 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "cursor": 1 })))
4040 + .mount(&server)
4041 + .await;
4042 +
4043 + let old = sealed_entry(
4044 + &client,
4045 + &server,
4046 + group_id,
4047 + &gck_v1,
4048 + device,
4049 + "r-old",
4050 + "before rotation",
4051 + )
4052 + .await;
4053 + let new = sealed_entry(
4054 + &client,
4055 + &server,
4056 + group_id,
4057 + &gck_v2,
4058 + device,
4059 + "r-new",
4060 + "after rotation",
4061 + )
4062 + .await;
4063 +
4064 + // Each generation's grant is fetched by version. Serving only these two
4065 + // means a client that ignored the per-entry version and asked for one key
4066 + // would still get an answer, and then fail to decrypt half the batch.
4067 + for (version, gck) in [(1, &gck_v1), (2, &gck_v2)] {
4068 + let sealed =
4069 + seal_gck_to_member(gck, &identity.public_key(), &group_ref, version).expect("seal");
4070 + Mock::given(method("GET"))
4071 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
4072 + .and(query_param("version", version.to_string()))
4073 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
4074 + "sealed_gck": sealed,
4075 + "gck_version": version,
4076 + })))
4077 + .mount(&server)
4078 + .await;
4079 + }
4080 +
4081 + Mock::given(method("POST"))
4082 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/pull$"))
4083 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
4084 + "changes": [
4085 + {
4086 + "seq": 1,
4087 + "device_id": device,
4088 + "table": old["table"],
4089 + "op": old["op"],
4090 + "row_id": old["row_id"],
4091 + "timestamp": old["timestamp"],
4092 + "data": old["data"],
4093 + "gck_version": 1,
4094 + },
4095 + {
4096 + "seq": 2,
4097 + "device_id": device,
4098 + "table": new["table"],
4099 + "op": new["op"],
4100 + "row_id": new["row_id"],
4101 + "timestamp": new["timestamp"],
4102 + "data": new["data"],
4103 + "gck_version": 2,
4104 + },
4105 + ],
4106 + "cursor": 2,
4107 + "has_more": false,
4108 + })))
4109 + .mount(&server)
4110 + .await;
4111 +
4112 + let (changes, cursor, has_more) = client
4113 + .group_pull_rich(group_id, 2, device, 0)
4114 + .await
4115 + .expect("a pull spanning generations must succeed");
4116 +
4117 + assert_eq!(cursor, 2);
4118 + assert!(!has_more);
4119 + assert_eq!(changes.len(), 2);
4120 + assert_eq!(
4121 + changes[0].entry.data.as_ref().expect("old plaintext"),
4122 + &json!({ "title": "before rotation" }),
4123 + "the pre-rotation entry must open under generation 1"
4124 + );
4125 + assert_eq!(
4126 + changes[1].entry.data.as_ref().expect("new plaintext"),
4127 + &json!({ "title": "after rotation" }),
4128 + "the post-rotation entry must open under generation 2"
4129 + );
4130 + }
4131 +
4132 + /// An entry with no generation is what a server predating per-generation
4133 + /// grants returns. The caller's current generation is the fallback, so an old
4134 + /// server keeps working rather than failing every pull.
4135 + #[tokio::test]
4136 + async fn an_entry_without_a_generation_falls_back_to_the_current_one() {
4137 + let server = MockServer::start().await;
4138 + let client = authed_client(&server);
4139 + let master = synckit_client::crypto::generate_master_key();
4140 + let identity = IdentityKeypair::from_master_key(&master);
4141 + client.set_master_key_raw(master);
4142 +
4143 + let group_id = GroupId::new(Uuid::new_v4());
4144 + let device = DeviceId::new(Uuid::new_v4());
4145 + let gck = generate_group_key();
4146 +
4147 + Mock::given(method("POST"))
4148 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
4149 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "cursor": 1 })))
4150 + .mount(&server)
4151 + .await;
4152 + let entry = sealed_entry(
4153 + &client,
4154 + &server,
4155 + group_id,
4156 + &gck,
4157 + device,
Lines truncated
M synckit-client/src/client/groups.rs +190 -42
@@ -1,11 +1,12 @@
1 1 //! Group sync: list the caller's groups, fetch their sealed GCK grant, and
2 2 //! push/pull a group's shared changelog under its Group Content Key.
3 3 //!
4 - //! The GCK is supplied by the caller, the `SyncStore` resolves it from the grant
5 - //! via the identity key (a later slice); these methods only encrypt/decrypt with
6 - //! it. Group entries bind `(group_id, table, row_id)` as AEAD associated data, so
7 - //! a ciphertext cannot be relocated across groups. Design: wiki
8 - //! synckit-multiscope-design.
4 + //! Push takes the GCK from the caller; pull resolves one per entry, because a
5 + //! rotation bumps the group's generation without re-encrypting what came before
6 + //! it and a single pull can therefore span generations. Group entries bind
7 + //! `(group_id, table, row_id)` as AEAD associated data, so a ciphertext cannot be
8 + //! relocated across groups. Design: wiki synckit-multiscope-design,
9 + //! synckit-groups-design.
9 10
10 11 use bytes::Bytes;
11 12 use tracing::instrument;
@@ -17,8 +18,8 @@
17 18 identity::{IdentityKeypair, IdentityPublicKey, generate_group_key, seal_gck_to_member},
18 19 ids::{DeviceId, GroupId, UserId},
19 20 types::{
20 - ChangeEntry, GroupGrant, GroupMember, PullRequest, PullResponse, PulledChange,
21 - PushResponse, SyncGroup, WirePushRequest,
21 + ChangeEntry, GroupGrant, GroupMember, GroupMemberPubkey, PullRequest, PullResponse,
22 + PulledChange, PushResponse, SyncGroup, WirePushRequest,
22 23 },
23 24 };
24 25
@@ -35,6 +36,21 @@
35 36 admin_pubkey: String,
36 37 }
37 38
39 + /// One member's re-sealed grant in a rotation batch.
40 + #[derive(serde::Serialize)]
41 + struct RotateGrant {
42 + user_id: UserId,
43 + sealed_gck: String,
44 + }
45 +
46 + /// Request body for `POST /groups/{id}/rotate`. The grant set is the new
47 + /// membership: the server drops anyone absent from it.
48 + #[derive(serde::Serialize)]
49 + struct RotateBody {
50 + gck_version: i32,
51 + grants: Vec<RotateGrant>,
52 + }
53 +
38 54 /// Request body for `POST /groups/{id}/members`.
39 55 #[derive(serde::Serialize)]
40 56 struct AddMemberBody<'a> {
@@ -163,15 +179,117 @@
163 179 .await
164 180 }
165 181
166 - /// Remove a member from a group by user id. Admin only. The server membership
167 - /// ACL revokes the member's group read/write access immediately.
182 + /// List every member's identity public key. Admin only. The input to a
183 + /// re-seal: [`rotate_group_key`](Self::rotate_group_key) consumes this.
184 + #[instrument(skip(self))]
185 + pub async fn list_member_pubkeys(&self, group_id: GroupId) -> Result<Vec<GroupMemberPubkey>> {
186 + let token = self.require_token()?;
187 + let url = self.endpoints.group_pubkeys(group_id);
188 + self.retry_request_json(Idempotency::ReadOnly, || {
189 + let req = self.http.get(&url).bearer_auth(&token);
190 + async move { check_response(req.send().await?).await }
191 + })
192 + .await
193 + }
194 +
195 + /// Rotate the group's Group Content Key, keeping everyone except `remove`.
196 + /// Admin only.
168 197 ///
169 - /// Forward secrecy for writes made *after* removal requires rotating the GCK
170 - /// (re-mint, re-seal to the remaining members, bump the generation), which the
171 - /// server does not yet expose, a later slice. This method performs the
172 - /// membership revocation only; data the member already pulled is in their hands.
198 + /// Mints a fresh GCK, seals it to every remaining member's stored public key,
199 + /// and posts the batch. The server treats the grant set as the new membership,
200 + /// so the removal and the re-key commit together: no window exists in which a
201 + /// removed member's key is still the current one.
202 + ///
203 + /// Pass an empty `remove` to re-key without dropping anyone (a suspected key
204 + /// compromise). Removing the admin is refused server-side; that would orphan
205 + /// the group.
206 + ///
207 + /// This does NOT re-encrypt history. Entries written before the rotation stay
208 + /// under the old generation, and anything a removed member already pulled is
209 + /// in their hands regardless.
210 + #[instrument(skip(self))]
211 + pub async fn rotate_group_key(&self, group_id: GroupId, remove: &[UserId]) -> Result<()> {
212 + let token = self.require_token()?;
213 +
214 + // The generation to write is one past what our own grant reports. Reading
215 + // it from the grant rather than from `list_groups` keeps the rotation
216 + // anchored to the key we can actually open.
217 + let current = self.group_grant(group_id).await?.gck_version;
218 + let next = current
219 + .checked_add(1)
220 + .ok_or_else(|| crate::error::SyncKitError::Crypto("gck_version overflow".into()))?;
221 +
222 + let pubkeys = self.list_member_pubkeys(group_id).await?;
223 + let gck = generate_group_key();
224 + let group_ref = group_id.to_string();
225 +
226 + let mut grants = Vec::with_capacity(pubkeys.len());
227 + for entry in &pubkeys {
228 + if remove.contains(&entry.user_id) {
229 + continue;
230 + }
231 + let pubkey = IdentityPublicKey::from_base64(&entry.pubkey)?;
232 + grants.push(RotateGrant {
233 + user_id: entry.user_id,
234 + sealed_gck: seal_gck_to_member(&gck, &pubkey, &group_ref, next)?,
235 + });
236 + }
237 +
238 + let body = Bytes::from(serde_json::to_vec(&RotateBody {
239 + gck_version: next,
240 + grants,
241 + })?);
242 + let url = self.endpoints.group_rotate(group_id);
243 +
244 + // The generation is the optimistic version: a replay of a rotation that
245 + // already committed carries a `gck_version` the group has moved past and
246 + // is refused, so a duplicate delivery cannot mint a second rotation.
247 + self.retry_request(
248 + Idempotency::IdempotentWrite {
249 + on: "gck_version must strictly advance; a replayed rotation is refused",
250 + },
251 + || {
252 + let req = self
253 + .http
254 + .post(&url)
255 + .bearer_auth(&token)
256 + .header("content-type", "application/json")
257 + .body(body.clone());
258 + async move { check_response(req.send().await?).await }
259 + },
260 + )
261 + .await?;
262 +
263 + // Our own cached GCK is now a generation behind.
264 + self.invalidate_gck(group_id);
265 + Ok(())
266 + }
267 +
268 + /// Remove a member from a group by user id, rotating the group key in the
269 + /// same operation. Admin only.
270 + ///
271 + /// Revocation alone would leave the removed member holding a working GCK, so
272 + /// this drives [`rotate_group_key`](Self::rotate_group_key): the server drops
273 + /// the member and re-keys the group in one transaction. Data the member
274 + /// already pulled is in their hands and no rotation changes that.
173 275 #[instrument(skip(self))]
174 276 pub async fn remove_member(&self, group_id: GroupId, member: UserId) -> Result<()> {
277 + self.rotate_group_key(group_id, &[member]).await
278 + }
279 +
280 + /// Remove a member without rotating the group key. Admin only.
281 + ///
282 + /// Revocation only: the member loses server-side access immediately, but a
283 + /// copy of the GCK they kept still opens group ciphertext, including entries
284 + /// written after this call. Prefer [`remove_member`](Self::remove_member).
285 + /// This exists for the case where the re-seal cannot be performed (a member's
286 + /// stored public key is unreadable, say) and revoking access now beats waiting.
287 + #[instrument(skip(self))]
288 + pub async fn revoke_member_without_rekey(
289 + &self,
290 + group_id: GroupId,
291 + member: UserId,
292 + ) -> Result<()> {
175 293 let token = self.require_token()?;
176 294 let url = self.endpoints.group_member(group_id, member);
177 295 self.retry_request(Idempotency::Keyed, || {
@@ -182,8 +300,10 @@
182 300 Ok(())
183 301 }
184 302
185 - /// Fetch the caller's own sealed GCK grant for a group. Open it with the
186 - /// member's identity private key to recover the GCK.
303 + /// Fetch the caller's newest sealed GCK grant for a group. Open it with the
304 + /// member's identity private key to recover the GCK. This is the key to write
305 + /// under; reading an entry from before a rotation wants
306 + /// [`group_grant_at`](Self::group_grant_at).
187 307 #[instrument(skip(self))]
188 308 pub async fn group_grant(&self, group_id: GroupId) -> Result<GroupGrant> {
189 309 let token = self.require_token()?;
@@ -195,6 +315,25 @@
195 315 .await
196 316 }
197 317
318 + /// Fetch the caller's sealed grant for one specific GCK generation, so a
319 + /// device can decrypt entries written before a rotation it lived through.
320 + ///
321 + /// A generation the caller never held (it predates their joining, or they were
322 + /// removed and re-added) is a 403.
323 + #[instrument(skip(self))]
324 + pub async fn group_grant_at(&self, group_id: GroupId, gck_version: i32) -> Result<GroupGrant> {
325 + let token = self.require_token()?;
326 + let url = format!(
327 + "{}?version={gck_version}",
328 + self.endpoints.group_grant(group_id)
329 + );
330 + self.retry_request_json(Idempotency::ReadOnly, || {
331 + let req = self.http.get(&url).bearer_auth(&token);
332 + async move { check_response(req.send().await?).await }
333 + })
334 + .await
335 + }
336 +
198 337 /// Push encrypted changes to a group's shared changelog under its GCK.
199 338 /// Returns the server cursor after the push.
200 339 #[instrument(skip(self, gck, changes))]
@@ -233,16 +372,21 @@
233 372 Ok(push_resp.cursor)
234 373 }
235 374
236 - /// Pull a group's changes since `cursor`, decrypting under its GCK. Returns
237 - /// `(changes, new_cursor, has_more)` with per-row device/seq metadata, ready
238 - /// for conflict resolution. Group pull has no master-key-rotation window (a
239 - /// group's key rotates server-side, re-encrypted in place), so the decrypt is
240 - /// a single-key pass.
241 - #[instrument(skip(self, gck))]
375 + /// Pull a group's changes since `cursor`, decrypting each entry under the GCK
376 + /// generation it was sealed with. Returns `(changes, new_cursor, has_more)`
377 + /// with per-row device/seq metadata, ready for conflict resolution.
378 + ///
379 + /// A rotation bumps the group's generation without re-encrypting history, so a
380 + /// single pull can span generations. Each entry carries its own, resolved
381 + /// through the cache in [`group_content_key`](Self::group_content_key), so the
382 + /// common case (one generation) is still one key and no extra request.
383 + /// `current_gck_version` is the fallback for an entry with no generation, which
384 + /// is what a server predating per-generation grants returns.
385 + #[instrument(skip(self))]
242 386 pub async fn group_pull_rich(
243 387 &self,
244 388 group_id: GroupId,
245 - gck: &[u8; 32],
389 + current_gck_version: i32,
246 390 device_id: DeviceId,
247 391 cursor: i64,
248 392 ) -> Result<(Vec<PulledChange>, i64, bool)> {
@@ -263,23 +407,28 @@
263 407 .await?;
264 408
265 409 let group_str = group_id.to_string();
266 - let changes = pull_resp
267 - .changes
268 - .into_iter()
269 - .map(|c| Self::decrypt_group_change_to_pulled(&group_str, c, gck))
270 - .collect::<Result<Vec<_>>>()?;
410 + let mut changes = Vec::with_capacity(pull_resp.changes.len());
411 + for entry in pull_resp.changes {
412 + let version = entry.gck_version.unwrap_or(current_gck_version);
413 + let gck = self.group_content_key(group_id, version).await?;
414 + changes.push(Self::decrypt_group_change_to_pulled(
415 + &group_str, entry, &gck,
416 + )?);
417 + }
271 418 Ok((changes, pull_resp.cursor, pull_resp.has_more))
272 419 }
273 420
274 421 /// Resolve a group's decrypted Group Content Key for `gck_version`, fetching
275 - /// and opening the sealed grant on a cache miss.
422 + /// and opening that generation's sealed grant on a cache miss.
276 423 ///
277 - /// Fast path: a cached GCK at the requested version is returned without a
278 - /// network call. On a miss (or a version mismatch, the group's key rotated),
279 - /// the sealed grant is fetched, opened with the identity keypair derived from
280 - /// this client's master key, and cached under the grant's actual version. The
281 - /// master key and identity secret never leave the client, the `SyncStore`
282 - /// only ever receives the resolved GCK.
424 + /// Fast path: a cached GCK for exactly that generation is returned without a
425 + /// network call. On a miss the grant for that generation is fetched, opened
426 + /// with the identity keypair derived from this client's master key, and
427 + /// cached. The master key and identity secret never leave the client, the
428 + /// `SyncStore` only ever receives the resolved GCK.
429 + ///
430 + /// Asking for a generation the caller never held is a server-side 403: a
431 + /// member sees the group from when they joined, not before.
283 432 // Consumed by `sync_now`'s scope iteration in slice 4.
284 433 #[allow(dead_code)]
285 434 pub(crate) async fn group_content_key(
@@ -287,19 +436,17 @@
287 436 group_id: GroupId,
288 437 gck_version: i32,
289 438 ) -> Result<crypto::ZeroizeOnDrop> {
290 - if let Some((v, gck)) = self.gck_cache.read().get(&group_id)
291 - && *v == gck_version
292 - {
439 + if let Some(gck) = self.gck_cache.read().get(&(group_id, gck_version)) {
293 440 return Ok(crypto::ZeroizeOnDrop(**gck));
294 441 }
295 442
296 - let grant = self.group_grant(group_id).await?;
443 + let grant = self.group_grant_at(group_id, gck_version).await?;
297 444 let master = self.require_master_key()?;
298 445 let gck = Self::open_group_grant(&grant, &master, group_id)?;
299 446 let out = crypto::ZeroizeOnDrop(*gck);
300 447 self.gck_cache
301 448 .write()
302 - .insert(group_id, (grant.gck_version, gck));
449 + .insert((group_id, grant.gck_version), gck);
303 450 Ok(out)
304 451 }
305 452
@@ -320,13 +467,13 @@
320 467 Ok(crypto::ZeroizeOnDrop(gck))
321 468 }
322 469
323 - /// Drop a group's cached GCK, forcing the next
324 - /// [`group_content_key`](Self::group_content_key) to re-fetch its grant. Call
470 + /// Drop every cached generation for a group, forcing the next
471 + /// [`group_content_key`](Self::group_content_key) to re-fetch its grants. Call
325 472 /// on a decrypt failure (a stale key after a rotation this client missed).
326 473 // Consumed by `sync_now`'s scope iteration in slice 4.
327 474 #[allow(dead_code)]
328 475 pub(crate) fn invalidate_gck(&self, group_id: GroupId) {
329 - self.gck_cache.write().remove(&group_id);
476 + self.gck_cache.write().retain(|(g, _), _| *g != group_id);
330 477 }
331 478 }
332 479
@@ -360,6 +507,7 @@
360 507 timestamp: wire.timestamp,
361 508 data: wire.data,
362 509 key_id: None,
510 + gck_version: None,
363 511 }
364 512 }
365 513
M synckit-client/src/client/helpers.rs +8
@@ -789,6 +789,7 @@
789 789 timestamp: wire.timestamp,
790 790 data: wire.data,
791 791 key_id: None,
792 + gck_version: None,
792 793 };
793 794 let decrypted = client.decrypt_change(pull_entry).unwrap();
794 795 assert_eq!(decrypted.op, ChangeOp::Delete);
@@ -873,6 +874,7 @@
873 874 timestamp: wire.timestamp,
874 875 data: wire.data,
875 876 key_id: None,
877 + gck_version: None,
876 878 };
877 879
878 880 let decrypted = client.decrypt_change(pull_entry).unwrap();
@@ -894,6 +896,7 @@
894 896 timestamp: Utc::now(),
895 897 data: None,
896 898 key_id: None,
899 + gck_version: None,
897 900 };
898 901
899 902 let decrypted = client.decrypt_change(pull_entry).unwrap();
@@ -914,6 +917,7 @@
914 917 timestamp: Utc::now(),
915 918 data: Some(serde_json::json!("some-encrypted-string")),
916 919 key_id: None,
920 + gck_version: None,
917 921 };
918 922
919 923 let err = client.decrypt_change(pull_entry).unwrap_err();
@@ -1245,6 +1249,7 @@
1245 1249 timestamp: wire.timestamp,
1246 1250 data: wire.data,
1247 1251 key_id: None,
1252 + gck_version: None,
1248 1253 };
1249 1254 let decrypted = client.decrypt_change(pull).unwrap();
1250 1255 assert_eq!(decrypted.table, entries[i].table);
@@ -1281,6 +1286,7 @@
1281 1286 timestamp: wire.timestamp,
1282 1287 data: wire.data,
1283 1288 key_id: None,
1289 + gck_version: None,
1284 1290 };
1285 1291 let decrypted = client.decrypt_change(pull).unwrap();
1286 1292 assert_eq!(
@@ -1315,6 +1321,7 @@
1315 1321 timestamp: wire.timestamp,
1316 1322 data: wire.data,
1317 1323 key_id: None,
1324 + gck_version: None,
1318 1325 };
1319 1326 let decrypted = client.decrypt_change(pull).unwrap();
1320 1327 assert_eq!(decrypted.row_id, "");
@@ -1345,6 +1352,7 @@
1345 1352 timestamp: Utc::now(),
1346 1353 data: Some(encrypted),
1347 1354 key_id,
1355 + gck_version: None,
1348 1356 }
1349 1357 }
1350 1358
M synckit-client/src/client/mod.rs +20 -5
@@ -195,6 +195,17 @@
195 195 format!("{}/{group_id}/members/{member}", self.groups_base)
196 196 }
197 197
198 + /// `GET` every member's identity public key (admin only), the input to a
199 + /// re-seal.
200 + fn group_pubkeys(&self, group_id: GroupId) -> String {
201 + format!("{}/{group_id}/pubkeys", self.groups_base)
202 + }
203 +
204 + /// `POST` a rotated GCK sealed to each remaining member (admin only).
205 + fn group_rotate(&self, group_id: GroupId) -> String {
206 + format!("{}/{group_id}/rotate", self.groups_base)
207 + }
208 +
198 209 /// `POST` encrypted changes to a group's shared changelog.
199 210 fn group_push(&self, group_id: GroupId) -> String {
200 211 format!("{}/{group_id}/push", self.groups_base)
@@ -324,11 +335,15 @@
324 335 master_key_id: RwLock<i32>,
325 336 /// Pending rotation key, if a rotation is in progress.
326 337 pending_key: RwLock<Option<PendingKeyState>>,
327 - /// Per-group decrypted Group Content Key cache: `group_id -> (gck_version,
328 - /// gck)`. Populated lazily by [`group_content_key`](Self::group_content_key)
329 - /// from the sealed grant; a version mismatch (rotation) or a decrypt failure
330 - /// forces a re-fetch. Holds one entry per group, the current generation.
331 - gck_cache: RwLock<std::collections::HashMap<GroupId, (i32, crypto::ZeroizeOnDrop)>>,
338 + /// Decrypted Group Content Key cache, keyed by `(group_id, gck_version)`.
339 + /// Populated lazily by [`group_content_key`](Self::group_content_key) from
340 + /// that generation's sealed grant.
341 + ///
342 + /// Keyed by generation rather than by group because a rotation does not
343 + /// re-encrypt history: a pull can return entries from several generations at
344 + /// once, and each needs its own key. A decrypt failure drops every generation
345 + /// for the group, forcing a re-fetch.
346 + gck_cache: RwLock<std::collections::HashMap<(GroupId, i32), crypto::ZeroizeOnDrop>>,
332 347 }
333 348
334 349 impl SyncKitClient {
M synckit-client/src/store/hlc.rs +424 -7
@@ -21,6 +21,7 @@
21 21
22 22 use super::db::{get_sync_state, set_sync_state};
23 23 use super::schema::{ConflictStrategy, SyncSchema};
24 + use super::stash;
24 25 use crate::conflict::{Resolution, detect_conflicts, resolve_lww_at};
25 26 use crate::error::Result;
26 27 use crate::ids::DeviceId;
@@ -164,12 +165,23 @@
164 165 /// clean vs conflicting against local pending, HLC-gate the clean set against the
165 166 /// committed ledger, resolve conflicts by `resolve_lww`, and collapse to one
166 167 /// entry per row (highest HLC wins, operation-agnostic).
168 + ///
169 + /// Every discard along that path is stashed under `scope` first (see
170 + /// [`super::stash`]). Three of them exist and they are easy to miscount: the
171 + /// conflict layer throws away the local side on `KeepRemote` and the remote side
172 + /// on `KeepLocal`/`Skip`, and the committed-HLC gate drops a superseded remote
173 + /// change without ever building a `ConflictPair` for it. A stash bolted onto the
174 + /// match arms alone would miss the third and quietest one.
175 + ///
176 + /// `ServerOrder` stashes nothing: it does not compare versions, so there is no
177 + /// loser to name. An app that picks it has chosen last-delivered-wins.
167 178 pub fn resolve_pull(
168 179 conn: &Connection,
169 180 schema: &SyncSchema,
170 181 node: DeviceId,
171 182 pulled: Vec<PulledChange>,
172 183 now: DateTime<Utc>,
184 + scope: &str,
173 185 ) -> Result<Vec<ChangeEntry>> {
174 186 match schema.conflict {
175 187 ConflictStrategy::ServerOrder => Ok(pulled.into_iter().map(|p| p.entry).collect()),
@@ -181,13 +193,46 @@
181 193 let local_pending = load_local_pending(conn, node)?;
182 194 let (clean, conflicts) = detect_conflicts(pulled, &local_pending, node);
183 195
184 - let mut resolved: Vec<ChangeEntry> =
185 - clean.gated_at(now, |t, r| lookup_committed(conn, t, r));
196 + // Site 3: superseded by the committed clock. A stash failure must not
197 + // fail the sync, so it is logged rather than propagated; losing the
198 + // evidence is bad, losing the pull because we could not record the
199 + // evidence is worse.
200 + let mut resolved: Vec<ChangeEntry> = clean.gated_at(
201 + now,
202 + |t, r| lookup_committed(conn, t, r),
203 + |dropped, committed| {
204 + if let Err(e) = stash::stash_superseded(conn, scope, dropped, committed) {
205 + tracing::warn!("could not stash a superseded change: {e}");
206 + }
207 + },
208 + );
186 209
187 210 for pair in conflicts {
188 211 match resolve_lww_at(&pair.local, &pair.remote, now) {
189 - Resolution::KeepRemote => resolved.push(pair.remote.entry),
190 - Resolution::KeepLocal | Resolution::Skip => {}
212 + // Site 1: our own edit is discarded.
213 + Resolution::KeepRemote => {
214 + stash_or_warn(
215 + conn,
216 + scope,
217 + stash::LosingSide::Local,
218 + &pair.local,
219 + node,
220 + &pair.remote.entry,
221 + );
222 + resolved.push(pair.remote.entry);
223 + }
224 + // Site 2: the other writer's edit is discarded.
225 + Resolution::KeepLocal | Resolution::Skip => {
226 + stash_or_warn(
227 + conn,
228 + scope,
229 + stash::LosingSide::Remote,
230 + &pair.remote.entry,
231 + pair.remote.device_id,
232 + &pair.local,
233 + );
234 + }
235 + // A merge keeps both sides' fields, so nothing was thrown away.
191 236 Resolution::Merged(data) => {
192 237 let hlc = max_hlc(pair.local.hlc, pair.remote.entry.hlc);
193 238 resolved.push(ChangeEntry {
@@ -203,11 +248,30 @@
203 248 }
204 249 }
205 250
251 + if let Err(e) = stash::trim_stash(conn) {
252 + tracing::warn!("could not trim the conflict stash: {e}");
253 + }
254 +
206 255 Ok(collapse_max_hlc(resolved))
207 256 }
208 257 }
209 258 }
210 259
260 + /// Stash one discarded side, logging rather than failing. Same reasoning as the
261 + /// gate's sink: the stash is evidence about a sync, never a reason to fail one.
262 + fn stash_or_warn(
263 + conn: &Connection,
264 + scope: &str,
265 + side: stash::LosingSide,
266 + losing: &ChangeEntry,
267 + losing_device: DeviceId,
268 + winning: &ChangeEntry,
269 + ) {
270 + if let Err(e) = stash::stash_loser(conn, scope, side, losing, losing_device, winning) {
271 + tracing::warn!("could not stash a discarded change: {e}");
272 + }
273 + }
274 +
211 275 /// Committed-HLC lookup for the gate. A read error is logged and treated as
212 276 /// "never applied", the gate then keeps the change, which the idempotent apply
213 277 /// path can safely re-run.
@@ -372,7 +436,7 @@
372 436 pulled: Vec<PulledChange>,
373 437 ) {
374 438 let now = Utc::now();
375 - let resolved = resolve_pull(conn, s, node, pulled, now).unwrap();
439 + let resolved = resolve_pull(conn, s, node, pulled, now, "").unwrap();
376 440 apply_remote_changes(conn, s, &resolved, "").unwrap();
377 441 record_committed(conn, &resolved).unwrap();
378 442 }
@@ -453,7 +517,7 @@
453 517 let a_pending = load_local_pending(&a, an);
454 518 record_committed(&a, &a_pending.unwrap()).unwrap();
455 519 // Re-pulling B's OLD change must be gated out (older than committed).
456 - let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now()).unwrap();
520 + let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now(), "").unwrap();
457 521 assert!(
458 522 resolved.iter().all(|e| e.row_id != "r"),
459 523 "stale re-pull must be gated"
@@ -499,7 +563,7 @@
499 563 let first = local_edit_as_pulled(&src, sn, "r", "first", 999, 1); // higher wall
500 564 let (src2, sn2) = device(3);
501 565 let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
502 - let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now()).unwrap();
566 + let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now(), "").unwrap();
503 567 apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
504 568 assert_eq!(
505 569 note_name(&a, "r").as_deref(),
@@ -507,4 +571,357 @@
507 571 "server order: last delivered wins"
508 572 );
509 573 }
574 +
575 + // ── Conflict stash ──
576 + //
577 + // LWW always discards one side; these pin that the discarded bytes are kept
578 + // rather than dropped. The one that matters most is the superseded case,
579 + // which never becomes a ConflictPair and so is invisible to resolve_lww.
580 +
581 + #[derive(Debug, PartialEq)]
582 + struct StashRow {
583 + table_name: String,
584 + row_id: String,
585 + scope: String,
586 + losing_side: String,
587 + losing_payload: Option<String>,
588 + }
589 +
590 + fn stash_rows(conn: &Connection) -> Vec<StashRow> {
591 + let mut stmt = conn
592 + .prepare(
593 + "SELECT table_name, row_id, scope, losing_side, losing_payload
594 + FROM sync_conflict_stash ORDER BY id",
595 + )
596 + .unwrap();
597 + stmt.query_map([], |r| {
598 + Ok(StashRow {
599 + table_name: r.get(0)?,
600 + row_id: r.get(1)?,
601 + scope: r.get(2)?,
602 + losing_side: r.get(3)?,
603 + losing_payload: r.get(4)?,
604 + })
605 + })
606 + .unwrap()
607 + .map(|r| r.unwrap())
608 + .collect()
609 + }
610 +
611 + /// A remote change with an explicit HLC and payload, as a peer would send it.
612 + fn remote_change(from: DeviceId, id: &str, name: &str, hlc: Hlc, seq: i64) -> PulledChange {
613 + PulledChange {
614 + entry: ChangeEntry {
615 + table: "note".to_string(),
616 + op: ChangeOp::Update,
617 + row_id: id.to_string(),
618 + timestamp: Utc::now(),
619 + hlc,
620 + data: Some(serde_json::json!({ "id": id, "name": name })),
621 + extra: serde_json::Map::default(),
622 + },
623 + device_id: from,
624 + seq,
625 + }
626 + }
627 +
628 + /// Site 1: a newer remote change beats our pending edit, so our edit is the
629 + /// one that vanishes from the row. It must be recoverable.
630 + #[test]
631 + fn stash_keeps_our_own_edit_when_the_remote_wins() {
632 + let (mut conn, n) = device(1);
633 + let peer = node(2);
634 +
635 + // Our pending edit, stamped early so it loses.
636 + conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
637 + .unwrap();
638 + stamp_pending(&conn, n, 1_000).unwrap();
639 +
640 + let newer = Hlc {
641 + wall_ms: 9_000,
642 + counter: 0,
643 + node: peer,
644 + };
645 + let resolved = resolve_pull(
646 + &conn,
647 + &schema(),
648 + n,
649 + vec![remote_change(peer, "n1", "theirs", newer, 1)],
650 + Utc::now(),
651 + "",
652 + )
653 + .unwrap();
654 + apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
655 +
656 + assert_eq!(note_name(&conn, "n1").as_deref(), Some("theirs"));
657 + let rows = stash_rows(&conn);
658 + assert_eq!(
659 + rows.len(),
660 + 1,
661 + "our discarded edit must be stashed: {rows:?}"
662 + );
663 + assert_eq!(rows[0].losing_side, "local");
664 + assert_eq!(rows[0].row_id, "n1");
665 + assert!(
666 + rows[0].losing_payload.as_deref().unwrap().contains("mine"),
667 + "the stash must hold the discarded value, not a placeholder: {rows:?}"
668 + );
669 + }
670 +
671 + /// Site 2: our pending edit wins, so the other writer's change is discarded.
672 + /// Their bytes are the ones that need keeping.
673 + #[test]
674 + fn stash_keeps_the_remote_edit_when_ours_wins() {
675 + let (conn, n) = device(1);
676 + let peer = node(2);
677 +
678 + conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
679 + .unwrap();
680 + stamp_pending(&conn, n, 9_000).unwrap();
681 +
682 + let older = Hlc {
683 + wall_ms: 1_000,
684 + counter: 0,
685 + node: peer,
686 + };
687 + let resolved = resolve_pull(
688 + &conn,
689 + &schema(),
690 + n,
691 + vec![remote_change(peer, "n1", "theirs", older, 1)],
692 + Utc::now(),
693 + "",
694 + )
695 + .unwrap();
696 + assert!(
697 + resolved.is_empty(),
698 + "the older remote change must not apply"
699 + );
700 +
701 + let rows = stash_rows(&conn);
702 + assert_eq!(
703 + rows.len(),
704 + 1,
705 + "their discarded edit must be stashed: {rows:?}"
706 + );
707 + assert_eq!(rows[0].losing_side, "remote");
708 + assert!(
709 + rows[0]
710 + .losing_payload
711 + .as_deref()
712 + .unwrap()
713 + .contains("theirs")
714 + );
715 + }
716 +
717 + /// Site 3, the quiet one: no local pending edit contests the row, so no
718 + /// ConflictPair is ever built. The change is dropped by the committed-HLC
719 + /// gate alone, which is why a stash wired only into the match arms misses it.
720 + #[test]
721 + fn stash_keeps_a_change_superseded_by_the_committed_clock() {
722 + let (mut conn, n) = device(1);
723 + let peer = node(2);
724 +
725 + // Apply and commit a newer value, with nothing left pending afterwards.
726 + let newer = Hlc {
727 + wall_ms: 9_000,
728 + counter: 0,
729 + node: peer,
730 + };
731 + let resolved = resolve_pull(
732 + &conn,
733 + &schema(),
734 + n,
735 + vec![remote_change(peer, "n1", "current", newer, 1)],
736 + Utc::now(),
737 + "",
738 + )
739 + .unwrap();
740 + apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
741 + record_committed(&conn, &resolved).unwrap();
742 + assert!(stash_rows(&conn).is_empty(), "nothing lost yet");
743 +
744 + // Now pull an older change for the same row. No pending edit contests it.
745 + let older = Hlc {
746 + wall_ms: 1_000,
747 + counter: 0,
748 + node: peer,
749 + };
750 + let resolved = resolve_pull(
751 + &conn,
752 + &schema(),
753 + n,
754 + vec![remote_change(peer, "n1", "stale", older, 2)],
755 + Utc::now(),
756 + "",
757 + )
758 + .unwrap();
759 + assert!(resolved.is_empty(), "the superseded change must not apply");
760 + assert_eq!(note_name(&conn, "n1").as_deref(), Some("current"));
761 +
762 + let rows = stash_rows(&conn);
763 + assert_eq!(
764 + rows.len(),
765 + 1,
766 + "the superseded change must be stashed: {rows:?}"
767 + );
768 + assert_eq!(rows[0].losing_side, "remote");
769 + assert!(rows[0].losing_payload.as_deref().unwrap().contains("stale"));
770 + }
771 +
772 + /// An echo carries the same bytes as the winner, so nothing was lost. Without
773 + /// this the stash fills with no-ops and stops being worth reading.
774 + #[test]
775 + fn stash_ignores_a_conflict_whose_payloads_are_identical() {
776 + let (conn, n) = device(1);
777 + let peer = node(2);
778 +
779 + conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'same')", [])
780 + .unwrap();
781 + stamp_pending(&conn, n, 1_000).unwrap();
782 + let local = load_local_pending(&conn, n).unwrap().pop().unwrap();
783 +
784 + // Byte-identical payload, newer clock: the remote wins and our identical
785 + // value goes away, which costs nothing.
786 + let mut remote = remote_change(
787 + peer,
788 + "n1",
789 + "x",
790 + Hlc {
791 + wall_ms: 9_000,
792 + counter: 0,
793 + node: peer,
794 + },
795 + 1,
796 + );
797 + remote.entry.data = local.data.clone();
798 +
799 + resolve_pull(&conn, &schema(), n, vec![remote], Utc::now(), "").unwrap();
800 + assert!(
801 + stash_rows(&conn).is_empty(),
802 + "an identical payload is not a lost edit: {:?}",
803 + stash_rows(&conn)
804 + );
805 + }
806 +
807 + /// A clock-poisoned entry is refused as hostile, not outvoted. Stashing it
808 + /// would hand whoever sent it a way to fill the stash.
809 + #[test]
810 + fn stash_ignores_a_clock_poisoned_drop() {
811 + let (conn, n) = device(1);
812 + let peer = node(2);
813 + let now = Utc::now();
814 + let poisoned = Hlc {
815 + wall_ms: now.timestamp_millis() + crate::conflict::MAX_HLC_DRIFT_MS * 10,
816 + counter: 0,
817 + node: peer,
818 + };
819 +
820 + let resolved = resolve_pull(
821 + &conn,
822 + &schema(),
823 + n,
824 + vec![remote_change(peer, "n1", "from the future", poisoned, 1)],
825 + now,
826 + "",
827 + )
828 + .unwrap();
829 + assert!(resolved.is_empty(), "a poisoned entry must not apply");
830 + assert!(
831 + stash_rows(&conn).is_empty(),
832 + "a poisoned entry is refused, not outvoted: {:?}",
833 + stash_rows(&conn)
834 + );
835 + }
836 +
837 + /// The scope a loss happened in is recorded, so a consuming app can tell a
838 + /// personal conflict from one inside a shared group.
839 + #[test]
840 + fn stash_records_the_scope() {
841 + let (conn, n) = device(1);
842 + let peer = node(2);
843 +
844 + conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
845 + .unwrap();
846 + stamp_pending(&conn, n, 9_000).unwrap();
847 +
848 + let older = Hlc {
849 + wall_ms: 1_000,
850 + counter: 0,
851 + node: peer,
852 + };
853 + resolve_pull(
854 + &conn,
855 + &schema(),
856 + n,
857 + vec![remote_change(peer, "n1", "theirs", older, 1)],
858 + Utc::now(),
859 + "group-7",
860 + )
861 + .unwrap();
862 +
863 + assert_eq!(stash_rows(&conn)[0].scope, "group-7");
864 + }
865 +
866 + /// ServerOrder does not compare versions, so it has no loser to name.
867 + #[test]
868 + fn server_order_stashes_nothing() {
869 + let (mut conn, n) = device(1);
870 + let peer = node(2);
871 + let s = server_order_schema();
872 +
873 + conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
874 + .unwrap();
875 + stamp_pending(&conn, n, 9_000).unwrap();
876 +
877 + let older = Hlc {
878 + wall_ms: 1_000,
879 + counter: 0,
880 + node: peer,
881 + };
882 + let resolved = resolve_pull(
883 + &conn,
884 + &s,
885 + n,
886 + vec![remote_change(peer, "n1", "theirs", older, 1)],
887 + Utc::now(),
888 + "",
889 + )
890 + .unwrap();
891 + apply_remote_changes(&mut conn, &s, &resolved, "").unwrap();
892 +
893 + assert!(stash_rows(&conn).is_empty());
894 + }
895 +
896 + /// The stash is bounded. Unbounded, a pathological sync loop grows it without
897 + /// limit.
898 + #[test]
899 + fn stash_is_trimmed_to_its_ceiling() {
900 + let (conn, _) = device(1);
901 + for i in 0..(super::stash::MAX_STASH_ROWS + 50) {
902 + conn.execute(
903 + "INSERT INTO sync_conflict_stash
904 + (table_name, row_id, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
905 + VALUES ('note', ?1, 'remote', '{}', '1:0:x', 'dev', '2:0:y')",
906 + [i.to_string()],
907 + )
908 + .unwrap();
909 + }
910 + super::stash::trim_stash(&conn).unwrap();
911 +
912 + let kept: i64 = conn
913 + .query_row("SELECT COUNT(*) FROM sync_conflict_stash", [], |r| r.get(0))
914 + .unwrap();
915 + assert_eq!(kept, super::stash::MAX_STASH_ROWS);
916 +
917 + // The newest survive: the oldest losses are the least likely to be acted on.
918 + let oldest_kept: String = conn
919 + .query_row(
920 + "SELECT row_id FROM sync_conflict_stash ORDER BY id ASC LIMIT 1",
921 + [],
922 + |r| r.get(0),
923 + )
924 + .unwrap();
925 + assert_eq!(oldest_kept, "50");
926 + }
510 927 }
M synckit-client/src/store/migrate.rs +74
@@ -68,6 +68,38 @@
68 68 hlc_node TEXT NOT NULL,
69 69 PRIMARY KEY (table_name, row_id)
70 70 ) WITHOUT ROWID;
71 +
72 + -- Every version last-write-wins threw away, kept instead of dropped.
73 + --
74 + -- LWW picks a winner and the loser's bytes are gone: the local user's edit leaves
75 + -- the row, or the other writer's does. Multi-user editing is not a first-class
76 + -- feature here and is not becoming one, so this is the safety net under it rather
77 + -- than a merge mechanism. A consuming app decides how (and whether) to surface a
78 + -- row; nothing in the engine reads them back.
79 + --
80 + -- Local-only by construction: absent from every sync manifest, so it is never
81 + -- group-scoped, never pushed, and never rides a shared changelog. A stash is
82 + -- per-device evidence about a decision this device made, not shared state.
83 + CREATE TABLE IF NOT EXISTS sync_conflict_stash (
84 + id INTEGER PRIMARY KEY AUTOINCREMENT,
85 + table_name TEXT NOT NULL,
86 + row_id TEXT NOT NULL,
87 + -- '' = personal, otherwise the group id, matching sync_changelog.scope.
88 + scope TEXT NOT NULL DEFAULT '',
89 + -- 'local' = this device's edit lost, 'remote' = the other writer's did.
90 + losing_side TEXT NOT NULL CHECK (losing_side IN ('local', 'remote')),
91 + -- The discarded payload, decrypted, as it would have been applied.
92 + losing_payload TEXT,
93 + losing_hlc TEXT NOT NULL,
94 + losing_device TEXT NOT NULL,
95 + winning_hlc TEXT NOT NULL,
96 + detected_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
97 + -- Set by a consuming app once a human has looked at this. Never set by the
98 + -- engine.
99 + reviewed_at TEXT
100 + );
101 + CREATE INDEX IF NOT EXISTS idx_sync_conflict_stash_row
102 + ON sync_conflict_stash(table_name, row_id);
71 103 ";
72 104
73 105 /// Provisioned once, never synced. Only emitted when a table hashes its row-id.
@@ -605,4 +637,46 @@
605 637 let plain = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
606 638 assert!(!plain.migration_sql().contains("row_id_salt"));
607 639 }
640 +
641 + /// The conflict stash is bookkeeping, not synced data. It must get the table
642 + /// and no triggers: a trigger would put a device's discarded plaintext into
643 + /// the changelog, which is exactly what "local-only by construction" rules
644 + /// out. Consumers are additionally responsible for keeping the name out of
645 + /// their own manifests.
646 + #[test]
647 + fn conflict_stash_is_created_but_never_synced() {
648 + let sql = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]).migration_sql();
649 +
650 + assert!(sql.contains("CREATE TABLE IF NOT EXISTS sync_conflict_stash"));
651 + assert!(
652 + !sql.contains("sync_conflict_stash_ai")
653 + && !sql.contains("AFTER INSERT ON sync_conflict_stash"),
654 + "a trigger on the stash would push discarded plaintext to the server"
655 + );
656 +
657 + // And it is reachable: the DDL runs and the shape is what the writer binds.
658 + let conn = rusqlite::Connection::open_in_memory().unwrap();
659 + conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
660 + .unwrap();
661 + conn.execute_batch(&sql).unwrap();
662 + conn.execute(
663 + "INSERT INTO sync_conflict_stash
664 + (table_name, row_id, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
665 + VALUES ('note', 'r1', 'remote', '{}', '1:0:a', 'dev', '2:0:b')",
666 + [],
667 + )
668 + .unwrap();
669 +
670 + // losing_side is constrained: a typo must fail loudly rather than produce
671 + // a row nothing can classify.
672 + assert!(
673 + conn.execute(
674 + "INSERT INTO sync_conflict_stash
675 + (table_name, row_id, losing_side, losing_hlc, losing_device, winning_hlc)
676 + VALUES ('note', 'r2', 'neither', '1:0:a', 'dev', '2:0:b')",
677 + [],
678 + )
679 + .is_err()
680 + );
681 + }
608 682 }
M synckit-client/src/store/mod.rs +1
@@ -27,6 +27,7 @@
27 27 pub mod migrate;
28 28 pub mod scheduler;
29 29 pub mod schema;
30 + pub(crate) mod stash;
30 31 pub mod sync;
31 32
32 33 pub use apply::{ApplyOutcome, Unapplied, apply_remote_changes};
M synckit-client/src/store/sync.rs +4 -6
@@ -199,18 +199,16 @@
199 199 cursor: i64,
200 200 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
201 201 async move {
202 - let gck = self.group_content_key(group_id, gck_version).await?;
203 202 match self
204 - .group_pull_rich(group_id, &gck, device_id, cursor)
203 + .group_pull_rich(group_id, gck_version, device_id, cursor)
205 204 .await
206 205 {
207 206 // A decrypt failure means this client is holding a stale GCK (it
208 - // missed a rotation): drop it, re-fetch the current grant, retry
207 + // missed a rotation): drop every cached generation, re-fetch, retry
209 208 // once. A second failure surfaces.
210 209 Err(SyncKitError::DecryptionFailed) => {
211 210 self.invalidate_gck(group_id);
212 - let gck = self.group_content_key(group_id, gck_version).await?;
213 - self.group_pull_rich(group_id, &gck, device_id, cursor)
211 + self.group_pull_rich(group_id, gck_version, device_id, cursor)
214 212 .await
215 213 }
216 214 other => other,
@@ -536,7 +534,7 @@
536 534 let mut all = retry;
537 535 all.extend(pulled);
538 536
539 - let resolved = resolve_pull(conn, schema, device_id, all, Utc::now())?;
537 + let resolved = resolve_pull(conn, schema, device_id, all, Utc::now(), scope_key)?;
540 538 let outcome = apply_remote_changes(conn, schema, &resolved, scope_key)?;
541 539
542 540 // Only what actually landed advances the committed ledger. Recording an
A synckit-client/src/store/stash.rs +166
@@ -1,0 +1,166 @@
1 + //! The conflict stash: every version last-write-wins threw away, kept.
2 + //!
3 + //! LWW always discards one side. Which side is a detail of clock order, so from
4 + //! the user's seat the outcome is the same either way: an edit somebody made is
5 + //! gone, with no record that it existed. Multi-user editing is not a first-class
6 + //! feature in SyncKit and is not becoming one; this is the safety net under it.
7 + //!
8 + //! Local-only by construction. `sync_conflict_stash` is absent from every sync
9 + //! manifest, so it is never group-scoped, never pushed, and never rides a shared
10 + //! changelog. A stash row is per-device evidence about a decision this device
11 + //! made, not shared state, and pushing it would leak one member's discarded
12 + //! plaintext into a group log.
13 + //!
14 + //! Nothing in the engine reads these rows back. A consuming app decides whether
15 + //! and how to surface them (a conflicts view, a badge on the row, an annotation
16 + //! in context) and when to mark one reviewed.
17 + //!
18 + //! Design: wiki synckit-groups-design.
19 +
20 + use rusqlite::Connection;
21 +
22 + use crate::conflict::canonical_payload;
23 + use crate::error::Result;
24 + use crate::types::{ChangeEntry, Hlc, PulledChange};
25 +
26 + /// Which side of the contest lost, as stored in `losing_side`.
27 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
28 + pub(crate) enum LosingSide {
29 + /// This device's own edit was discarded: a remote change won.
30 + Local,
31 + /// The other writer's edit was discarded: our value stood.
32 + Remote,
33 + }
34 +
35 + impl LosingSide {
36 + fn as_str(self) -> &'static str {
37 + match self {
38 + LosingSide::Local => "local",
39 + LosingSide::Remote => "remote",
40 + }
41 + }
42 + }
43 +
44 + /// How many stash rows to keep per device before the oldest are trimmed.
45 + ///
46 + /// The stash is evidence a human might read, not an audit log. Unbounded, a
47 + /// pathological sync loop between two devices would grow it without limit; at
48 + /// this size it stays small next to the changelog and still holds far more than
49 + /// anyone will review.
50 + pub(crate) const MAX_STASH_ROWS: i64 = 1_000;
51 +
52 + /// Format an HLC for storage. Sortable and human-legible, so a stash row can be
53 + /// ordered and read without decoding.
54 + fn hlc_text(hlc: &Hlc) -> String {
55 + format!("{}:{}:{}", hlc.wall_ms, hlc.counter, hlc.node)
56 + }
57 +
58 + /// Record one discarded version.
59 + ///
60 + /// Returns `Ok(false)` without writing when the two payloads are byte-identical
61 + /// under [`canonical_payload`], the comparison [`crate::conflict::resolve_lww`]
62 + /// already uses for its exact-HLC tiebreak. Every echo and every unchanged
63 + /// re-save would otherwise stash, and a table full of no-ops is one nobody reads.
64 + pub(crate) fn stash_loser(
65 + conn: &Connection,
66 + scope: &str,
67 + side: LosingSide,
68 + losing: &ChangeEntry,
69 + losing_device: crate::ids::DeviceId,
70 + winning: &ChangeEntry,
71 + ) -> Result<bool> {
72 + if canonical_payload(losing.data.as_ref()) == canonical_payload(winning.data.as_ref()) {
73 + return Ok(false);
74 + }
75 +
76 + let payload = losing
77 + .data
78 + .as_ref()
79 + .map(serde_json::to_string)
80 + .transpose()
81 + .map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?;
82 +
83 + conn.execute(
84 + "INSERT INTO sync_conflict_stash
85 + (table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
86 + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
87 + rusqlite::params![
88 + losing.table,
89 + losing.row_id,
90 + scope,
91 + side.as_str(),
92 + payload,
93 + hlc_text(&losing.hlc),
94 + losing_device.to_string(),
95 + hlc_text(&winning.hlc),
96 + ],
97 + )?;
98 +
99 + tracing::debug!(
100 + table = %losing.table,
101 + row_id = %losing.row_id,
102 + side = side.as_str(),
103 + "stashed the losing side of a conflict"
104 + );
105 + Ok(true)
106 + }
107 +
108 + /// Record a remote change the committed-HLC gate discarded.
109 + ///
110 + /// This is the quiet loss: no [`crate::conflict::ConflictPair`] is ever built for
111 + /// it, because no local *pending* edit contests the row. It happens when this
112 + /// device already applied and pushed a newer edit and then pulls an older remote
113 + /// one, which means the other writer's edit is dropped without anything looking
114 + /// like a conflict. There is no losing `ChangeEntry` to compare against, only the
115 + /// committed clock, so the payload-identity check cannot apply here.
116 + pub(crate) fn stash_superseded(
117 + conn: &Connection,
118 + scope: &str,
119 + dropped: &PulledChange,
120 + committed: &Hlc,
121 + ) -> Result<()> {
122 + let payload = dropped
123 + .entry
124 + .data
125 + .as_ref()
126 + .map(serde_json::to_string)
127 + .transpose()
128 + .map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?;
129 +
130 + conn.execute(
131 + "INSERT INTO sync_conflict_stash
132 + (table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
133 + VALUES (?1, ?2, ?3, 'remote', ?4, ?5, ?6, ?7)",
134 + rusqlite::params![
135 + dropped.entry.table,
136 + dropped.entry.row_id,
137 + scope,
138 + payload,
139 + hlc_text(&dropped.entry.hlc),
140 + dropped.device_id.to_string(),
141 + hlc_text(committed),
142 + ],
143 + )?;
144 +
145 + tracing::debug!(
146 + table = %dropped.entry.table,
147 + row_id = %dropped.entry.row_id,
148 + "stashed a remote change superseded by the committed HLC"
149 + );
150 + Ok(())
151 + }
152 +
153 + /// Trim the stash to [`MAX_STASH_ROWS`], oldest first. Reviewed rows are trimmed
154 + /// like any other: marking one reviewed says a human saw it, not that it must be
155 + /// kept forever.
156 + pub(crate) fn trim_stash(conn: &Connection) -> Result<usize> {
157 + let removed = conn.execute(
158 + "DELETE FROM sync_conflict_stash WHERE id NOT IN
159 + (SELECT id FROM sync_conflict_stash ORDER BY id DESC LIMIT ?1)",
160 + rusqlite::params![MAX_STASH_ROWS],
161 + )?;
162 + if removed > 0 {
163 + tracing::debug!(removed, "trimmed the conflict stash");
164 + }
165 + Ok(removed)
166 + }