Skip to main content

max / makenotwork

synckit: multi-scope transport seam + scope-aware push (Groups phase 3, slice 4a) The push half of multi-scope sync, plus the SyncTransport group seam. - SyncScope enum (Personal | Group{id, gck_version}); .key() maps to the sync_changelog.scope / sync_scope_cursor.scope string ('' = personal). - SyncTransport gains list_group_scopes, group_scope_push, group_scope_pull with default bodies (a transport that doesn't support groups just never syncs one), so existing test doubles compile unchanged. SyncKitClient overrides all three, resolving the GCK internally (group_content_key) so the store never handles keys; group_scope_pull retries once on a decrypt failure (missed rotation: invalidate + re-fetch the grant). This wires up slice 3's resolver. - read_push_batch filters by scope; push_changes becomes a thin wrapper over the new scope-aware push_scope, which drains only the scope's changelog rows and dispatches to personal push or group_scope_push. A group's changes never leak into the personal push and vice versa. - test: push_scope drains only its own scope -- a personal task (group_id NULL) goes only to the personal log, a group task only to that group's log. Pull + apply-side group_id stamping + sync_now iteration + the two-member convergence test land in slice 4b. Design: wiki synckit-multiscope-design. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 20:02 UTC
Commit: f1945175376e6b57d9012de9f7943356e6ec6232
Parent: f1050c5
2 files changed, +251 insertions, -13 deletions
M shared/synckit-client/src/store/mod.rs +3 -2
@@ -43,6 +43,7 @@ pub use hlc::{
43 43 pub use scheduler::{NoopObserver, SyncObserver, SyncState};
44 44 pub use schema::{ConflictStrategy, DeleteMode, RowIdScheme, SyncMode, SyncSchema, SyncTable};
45 45 pub use sync::{
46 - PUSH_BATCH_LIMIT, PullOutcome, SyncTransport, cleanup_changelog, create_initial_snapshot,
47 - enforce_changelog_retention, ensure_device_registered, pull_changes, push_changes,
46 + PUSH_BATCH_LIMIT, PullOutcome, SyncScope, SyncTransport, cleanup_changelog,
47 + create_initial_snapshot, enforce_changelog_retention, ensure_device_registered, pull_changes,
48 + push_changes, push_scope,
48 49 };
M shared/synckit-client/src/store/sync.rs +248 -11
@@ -25,9 +25,32 @@ use super::migrate::{json_object, row_id_expr};
25 25 use super::schema::{SyncMode, SyncSchema};
26 26 use crate::client::SyncKitClient;
27 27 use crate::error::{Result, SyncKitError};
28 - use crate::ids::DeviceId;
28 + use crate::ids::{DeviceId, GroupId};
29 29 use crate::types::{ChangeEntry, ChangeOp, Device, Hlc, PulledChange};
30 30
31 + /// A sync scope: the personal changelog, or one group's shared changelog. Each
32 + /// scope has its own changelog partition (`sync_changelog.scope`), pull cursor
33 + /// (`sync_scope_cursor`), key, and endpoint. See the wiki design note
34 + /// `synckit-multiscope-design`.
35 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
36 + pub enum SyncScope {
37 + /// The user's personal data, encrypted under the master key.
38 + Personal,
39 + /// A group's shared data, encrypted under the group's GCK at `gck_version`.
40 + Group { id: GroupId, gck_version: i32 },
41 + }
42 +
43 + impl SyncScope {
44 + /// The `sync_changelog.scope` / `sync_scope_cursor.scope` string for this
45 + /// scope: `""` for personal, the group id otherwise.
46 + pub fn key(&self) -> String {
47 + match self {
48 + SyncScope::Personal => String::new(),
49 + SyncScope::Group { id, .. } => id.to_string(),
50 + }
51 + }
52 + }
53 +
31 54 /// Maximum changes sent in one push batch.
32 55 pub const PUSH_BATCH_LIMIT: usize = 500;
33 56
@@ -58,8 +81,58 @@ pub trait SyncTransport: Sync {
58 81 device_id: DeviceId,
59 82 cursor: i64,
60 83 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send;
84 +
85 + /// The groups this user belongs to, as `(group_id, current_gck_version)` —
86 + /// the scopes to sync beyond personal. Defaults to none, so a transport that
87 + /// does not support groups (a test double, a future minimal SDK) simply never
88 + /// syncs any group scope.
89 + // `-> impl Future + Send` (not `async fn`) keeps the explicit Send bound the
90 + // store's spawned tasks need; that is the whole trait's convention.
91 + #[allow(clippy::manual_async_fn)]
92 + fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
93 + async move { Ok(Vec::new()) }
94 + }
95 +
96 + /// Push encrypted changes to a group's shared changelog. The implementation
97 + /// resolves the group's GCK internally; the caller never handles keys. The
98 + /// default errors — it is only reached if [`list_group_scopes`] returns a
99 + /// group, which the default never does.
100 + #[allow(clippy::manual_async_fn)]
101 + fn group_scope_push(
102 + &self,
103 + _group_id: GroupId,
104 + _gck_version: i32,
105 + _device_id: DeviceId,
106 + _changes: Vec<ChangeEntry>,
107 + ) -> impl Future<Output = Result<i64>> + Send {
108 + async move {
109 + Err(SyncKitError::Internal(
110 + "group sync not supported by this transport".into(),
111 + ))
112 + }
113 + }
114 +
115 + /// Pull a group's changes since `cursor`, decrypted under its GCK. Default
116 + /// errors; see [`group_scope_push`](Self::group_scope_push).
117 + #[allow(clippy::manual_async_fn)]
118 + fn group_scope_pull(
119 + &self,
120 + _group_id: GroupId,
121 + _gck_version: i32,
122 + _device_id: DeviceId,
123 + _cursor: i64,
124 + ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
125 + async move {
126 + Err(SyncKitError::Internal(
127 + "group sync not supported by this transport".into(),
128 + ))
129 + }
130 + }
61 131 }
62 132
133 + // The group methods below return `-> impl Future + Send` with an `async move`
134 + // body (not `async fn`) to keep the explicit Send bound the trait requires.
135 + #[allow(clippy::manual_async_fn)]
63 136 impl SyncTransport for SyncKitClient {
64 137 fn register_device(
65 138 &self,
@@ -85,6 +158,57 @@ impl SyncTransport for SyncKitClient {
85 158 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
86 159 SyncKitClient::pull_rich(self, device_id, cursor)
87 160 }
161 +
162 + fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
163 + async move {
164 + Ok(self
165 + .list_groups()
166 + .await?
167 + .into_iter()
168 + .map(|g| (g.id, g.gck_version))
169 + .collect())
170 + }
171 + }
172 +
173 + fn group_scope_push(
174 + &self,
175 + group_id: GroupId,
176 + gck_version: i32,
177 + device_id: DeviceId,
178 + changes: Vec<ChangeEntry>,
179 + ) -> impl Future<Output = Result<i64>> + Send {
180 + async move {
181 + let gck = self.group_content_key(group_id, gck_version).await?;
182 + self.group_push(group_id, &gck, device_id, changes).await
183 + }
184 + }
185 +
186 + fn group_scope_pull(
187 + &self,
188 + group_id: GroupId,
189 + gck_version: i32,
190 + device_id: DeviceId,
191 + cursor: i64,
192 + ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
193 + async move {
194 + let gck = self.group_content_key(group_id, gck_version).await?;
195 + match self
196 + .group_pull_rich(group_id, &gck, device_id, cursor)
197 + .await
198 + {
199 + // A decrypt failure means this client is holding a stale GCK (it
200 + // missed a rotation): drop it, re-fetch the current grant, retry
201 + // once. A second failure surfaces.
202 + Err(SyncKitError::DecryptionFailed) => {
203 + self.invalidate_gck(group_id);
204 + let gck = self.group_content_key(group_id, gck_version).await?;
205 + self.group_pull_rich(group_id, &gck, device_id, cursor)
206 + .await
207 + }
208 + other => other,
209 + }
210 + }
211 + }
88 212 }
89 213
90 214 /// Ensure this install has a registered device, returning its id (which doubles
@@ -135,21 +259,37 @@ struct PushBatch {
135 259 total_read: usize,
136 260 }
137 261
138 - /// Drain all unpushed local changes to the server, batch by batch.
262 + /// Drain all unpushed local changes for the **personal** scope. Thin wrapper over
263 + /// [`push_scope`] preserved for the personal call sites.
264 + pub async fn push_changes<T: SyncTransport>(
265 + db: &DbSource,
266 + client: &T,
267 + device_id: DeviceId,
268 + ) -> Result<u64> {
269 + push_scope(db, client, device_id, SyncScope::Personal).await
270 + }
271 +
272 + /// Drain all unpushed local changes for one `scope` to the server, batch by batch.
139 273 ///
140 - /// Each batch is stamped and read on a blocking thread, sent, then marked pushed
274 + /// Each batch is stamped and read on a blocking thread, sent to the scope's
275 + /// endpoint (personal push, or a group push under its GCK), then marked pushed
141 276 /// with its HLC recorded as the committed clock (so a peer's later stale re-pull
142 - /// of this row is gated out). `device_id` doubles as the HLC node.
143 - pub async fn push_changes<T: SyncTransport>(
277 + /// of this row is gated out). `device_id` doubles as the HLC node. The drain
278 + /// reads only rows whose `sync_changelog.scope` matches, so a group's changes
279 + /// never leak into the personal push and vice versa.
280 + pub async fn push_scope<T: SyncTransport>(
144 281 db: &DbSource,
145 282 client: &T,
146 283 device_id: DeviceId,
284 + scope: SyncScope,
147 285 ) -> Result<u64> {
286 + let scope_key = scope.key();
148 287 let mut pushed_total = 0u64;
149 288 loop {
150 289 let db_read = db.clone();
290 + let sk = scope_key.clone();
151 291 let batch =
152 - tokio::task::spawn_blocking(move || read_push_batch(&db_read.open()?, device_id))
292 + tokio::task::spawn_blocking(move || read_push_batch(&db_read.open()?, device_id, &sk))
153 293 .await
154 294 .map_err(join_err)??;
155 295
@@ -158,7 +298,16 @@ pub async fn push_changes<T: SyncTransport>(
158 298 }
159 299
160 300 if !batch.wire.is_empty() {
161 - client.push(device_id, batch.wire.clone()).await?;
301 + match scope {
302 + SyncScope::Personal => {
303 + client.push(device_id, batch.wire.clone()).await?;
304 + }
305 + SyncScope::Group { id, gck_version } => {
306 + client
307 + .group_scope_push(id, gck_version, device_id, batch.wire.clone())
308 + .await?;
309 + }
310 + }
162 311 }
163 312
164 313 let db_write = db.clone();
@@ -183,13 +332,13 @@ pub async fn push_changes<T: SyncTransport>(
183 332 Ok(pushed_total)
184 333 }
185 334
186 - fn read_push_batch(conn: &Connection, device_id: DeviceId) -> Result<PushBatch> {
335 + fn read_push_batch(conn: &Connection, device_id: DeviceId, scope: &str) -> Result<PushBatch> {
187 336 stamp_pending(conn, device_id, Utc::now().timestamp_millis())?;
188 337 let mut stmt = conn.prepare(
189 338 "SELECT id, table_name, op, row_id, data, hlc_wall, hlc_counter \
190 - FROM sync_changelog WHERE pushed = 0 ORDER BY id LIMIT ?1",
339 + FROM sync_changelog WHERE pushed = 0 AND scope = ?2 ORDER BY id LIMIT ?1",
191 340 )?;
192 - let rows = stmt.query_map([PUSH_BATCH_LIMIT as i64], |r| {
341 + let rows = stmt.query_map(rusqlite::params![PUSH_BATCH_LIMIT as i64, scope], |r| {
193 342 Ok((
194 343 r.get::<_, i64>(0)?,
195 344 r.get::<_, String>(1)?,
@@ -400,13 +549,32 @@ mod tests {
400 549 use super::*;
401 550 use std::sync::{Arc, Mutex};
402 551
403 - /// A shared in-memory "server": an append-only log of (origin device, entry).
552 + /// A shared in-memory "server": an append-only personal log of (origin
553 + /// device, entry), plus a group log of (group, origin device, entry).
404 554 #[derive(Clone, Default)]
405 555 struct FakeServer {
406 556 log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
557 + group_log: Arc<Mutex<Vec<(GroupId, DeviceId, ChangeEntry)>>>,
407 558 }
408 559
409 560 impl SyncTransport for FakeServer {
561 + fn group_scope_push(
562 + &self,
563 + group_id: GroupId,
564 + _gck_version: i32,
565 + device_id: DeviceId,
566 + changes: Vec<ChangeEntry>,
567 + ) -> impl Future<Output = Result<i64>> + Send {
568 + let group_log = self.group_log.clone();
569 + async move {
570 + let mut l = group_log.lock().unwrap();
571 + for c in changes {
572 + l.push((group_id, device_id, c));
573 + }
574 + Ok(l.len() as i64)
575 + }
576 + }
577 +
410 578 async fn register_device(&self, _name: &str, _platform: &str) -> Result<Device> {
411 579 Ok(Device {
412 580 id: DeviceId::new(uuid::Uuid::from_u128(0xDE)),
@@ -461,6 +629,75 @@ mod tests {
461 629 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
462 630 }
463 631
632 + /// A schema with a group-scoped table: `task` carries a local `group_id`
633 + /// provenance column (not a synced column), declared via `group_scoped`.
634 + fn group_schema() -> SyncSchema {
635 + SyncSchema::new(vec![
636 + SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
637 + ])
638 + }
639 +
640 + fn group_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
641 + let db = DbSource::path(path);
642 + let conn = db.open().unwrap();
643 + conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
644 + .unwrap();
645 + conn.execute_batch(&group_schema().migration_sql()).unwrap();
646 + (db, DeviceId::new(uuid::Uuid::from_u128(n)))
647 + }
648 +
649 + #[tokio::test]
650 + async fn push_scope_drains_only_its_own_scope() {
651 + let dir = tempdir();
652 + let (db, node) = group_device(&dir.join("g.db"), 7);
653 + let server = FakeServer::default();
654 + let gid = GroupId::new(uuid::Uuid::from_u128(0x6971));
655 +
656 + // A personal task (group_id NULL) and a group task (group_id = gid).
657 + {
658 + let c = db.open().unwrap();
659 + c.execute(
660 + "INSERT INTO task (id, name, group_id) VALUES ('p', 'personal', NULL)",
661 + [],
662 + )
663 + .unwrap();
664 + c.execute(
665 + "INSERT INTO task (id, name, group_id) VALUES ('g', 'grouped', ?1)",
666 + [gid.to_string()],
667 + )
668 + .unwrap();
669 + }
670 +
671 + // Personal push drains only the personal row.
672 + let pushed = push_scope(&db, &server, node, SyncScope::Personal)
673 + .await
674 + .unwrap();
675 + assert_eq!(pushed, 1);
676 + assert_eq!(server.log.lock().unwrap().len(), 1);
677 + assert_eq!(server.log.lock().unwrap()[0].1.row_id, "p");
678 + assert!(server.group_log.lock().unwrap().is_empty());
679 +
680 + // Group push drains only the group row, to that group.
681 + let pushed = push_scope(
682 + &db,
683 + &server,
684 + node,
685 + SyncScope::Group {
686 + id: gid,
687 + gck_version: 1,
688 + },
689 + )
690 + .await
691 + .unwrap();
692 + assert_eq!(pushed, 1);
693 + let gl = server.group_log.lock().unwrap();
694 + assert_eq!(gl.len(), 1);
695 + assert_eq!(gl[0].0, gid);
696 + assert_eq!(gl[0].2.row_id, "g");
697 + // The personal log did not grow.
698 + assert_eq!(server.log.lock().unwrap().len(), 1);
699 + }
700 +
464 701 fn device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
465 702 let db = DbSource::path(path);
466 703 let conn = db.open().unwrap();