Skip to main content

max / makenotwork

synckit: scope-aware pull + apply stamping + sync_now iteration (Groups phase 3, slice 4b) Completes the multi-scope SyncStore: the client now syncs a personal scope plus every group the user belongs to, each under its own key and cursor. Two members converge on a shared task through a group scope. - apply: apply_remote_changes takes the pull scope and stamps a group-scoped table's group_id column from it (never from the payload) on upsert -- provenance comes from the channel, not the wire. Personal scope leaves group_id NULL; deletes rely on globally-unique row-ids. - sync: pull_changes becomes a thin wrapper over the new scope-aware pull_scope, which reads the scope's own sync_scope_cursor, dispatches to personal or group pull, and applies with the scope stamped. Personal pull now uses the '' scope cursor (migrated from the old pull_cursor in slice 1). - facade: sync_now pushes + pulls personal, then iterates list_group_scopes, syncing each group scope. A groupless transport reports none, so personal-only apps are unaffected. - test: two_members_converge_through_a_group_scope -- member A creates a group task, both sync_now, member B receives it with group_id stamped from the channel, and B's personal changelog scope stays empty (no leak). This closes p3-multiscope (slices 1-4). Remaining in phase 3: p3-rotation (server group-GCK-rotation routes + member-pubkey storage). 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:16 UTC
Commit: cb94e5e479e81275ed1dd4677ed8ee654b8318d5
Parent: f194517
5 files changed, +242 insertions, -35 deletions
M shared/synckit-client/src/store/apply.rs +26 -4
@@ -51,6 +51,7 @@ pub fn apply_remote_changes(
51 51 conn: &mut Connection,
52 52 schema: &SyncSchema,
53 53 changes: &[ChangeEntry],
54 + scope: &str,
54 55 ) -> Result<ApplyOutcome> {
55 56 let by_name: HashMap<&str, &SyncTable> = schema.tables.iter().map(|t| (t.name, t)).collect();
56 57
@@ -63,7 +64,8 @@ pub fn apply_remote_changes(
63 64 conn.execute_batch("PRAGMA foreign_keys=OFF")?;
64 65 }
65 66
66 - let outcome = with_applying_remote(conn, |tx| apply_inner(tx, schema, &by_name, changes));
67 + let outcome =
68 + with_applying_remote(conn, |tx| apply_inner(tx, schema, &by_name, changes, scope));
67 69
68 70 if fk_off {
69 71 // Restore enforcement regardless of the apply result. A failed restore is
@@ -82,6 +84,7 @@ fn apply_inner(
82 84 schema: &SyncSchema,
83 85 by_name: &HashMap<&str, &SyncTable>,
84 86 changes: &[ChangeEntry],
87 + scope: &str,
85 88 ) -> Result<ApplyOutcome> {
86 89 let mut out = ApplyOutcome::default();
87 90
@@ -109,7 +112,11 @@ fn apply_inner(
109 112 for change in changes.iter().filter(|c| {
110 113 c.table == table.name && matches!(c.op, ChangeOp::Insert | ChangeOp::Update)
111 114 }) {
112 - step(&mut out, table, apply_upsert(tx, table, change, &not_null))?;
115 + step(
116 + &mut out,
117 + table,
118 + apply_upsert(tx, table, change, &not_null, scope),
119 + )?;
113 120 }
114 121 }
115 122
@@ -209,6 +216,7 @@ fn apply_upsert(
209 216 table: &SyncTable,
210 217 change: &ChangeEntry,
211 218 not_null: &HashMap<&str, HashSet<String>>,
219 + scope: &str,
212 220 ) -> rusqlite::Result<bool> {
213 221 let Some(data) = change.data.as_ref().filter(|v| v.is_object()) else {
214 222 return Ok(false); // an upsert with no object payload cannot be applied
@@ -222,7 +230,7 @@ fn apply_upsert(
222 230 let nn = not_null
223 231 .get(table.name)
224 232 .expect("Full table not_null precomputed");
225 - apply_full_upsert(tx, table, data, nn)
233 + apply_full_upsert(tx, table, data, nn, scope)
226 234 }
227 235 }
228 236 }
@@ -232,6 +240,7 @@ fn apply_full_upsert(
232 240 table: &SyncTable,
233 241 data: &Value,
234 242 not_null: &HashSet<String>,
243 + scope: &str,
235 244 ) -> rusqlite::Result<bool> {
236 245 // A remote null for a NOT NULL column is invalid input the changelog never
237 246 // emits; omit such columns so a new row takes the schema default and an
@@ -258,6 +267,16 @@ fn apply_full_upsert(
258 267 return Ok(false);
259 268 }
260 269
270 + // Group provenance: stamp a group-scoped table's `group_id` column from the
271 + // pull scope (never from the payload) so the pulled row is tagged with the
272 + // group it came from. Personal scope ("") leaves group_id at its default
273 + // (NULL). Injected into the INSERT columns only — never into the ON CONFLICT
274 + // SET, so an existing row's group never changes.
275 + let group_stamp: Option<&str> = table.group_scope.filter(|_| !scope.is_empty());
276 + if let Some(gcol) = group_stamp {
277 + insert_cols.push(gcol);
278 + }
279 +
261 280 let pk: HashSet<&str> = table.pk.iter().copied().collect();
262 281 let preserve: HashSet<&str> = table.preserve_local.iter().copied().collect();
263 282 // ON CONFLICT update touches present columns that are neither PK nor a
@@ -307,6 +326,9 @@ fn apply_full_upsert(
307 326 .iter()
308 327 .map(|v| Box::new((*v).to_string()) as Box<dyn ToSql>),
309 328 );
329 + if group_stamp.is_some() {
330 + params.push(Box::new(scope.to_string()) as Box<dyn ToSql>);
331 + }
310 332
311 333 exec(tx, &sql, &params)?;
312 334 Ok(true)
@@ -570,7 +592,7 @@ mod tests {
570 592 }
571 593
572 594 fn apply(conn: &mut Connection, changes: &[ChangeEntry]) -> ApplyOutcome {
573 - apply_remote_changes(conn, &schema(), changes).unwrap()
595 + apply_remote_changes(conn, &schema(), changes, "").unwrap()
574 596 }
575 597
576 598 #[test]
M shared/synckit-client/src/store/facade.rs +164 -5
@@ -19,8 +19,9 @@ use super::scheduler::{
19 19 };
20 20 use super::schema::SyncSchema;
21 21 use super::sync::{
22 - SyncTransport, cleanup_changelog, create_initial_snapshot, enforce_changelog_retention,
23 - ensure_device_registered, pull_changes, push_changes,
22 + SyncScope, SyncTransport, cleanup_changelog, create_initial_snapshot,
23 + enforce_changelog_retention, ensure_device_registered, pull_changes, pull_scope, push_changes,
24 + push_scope,
24 25 };
25 26 use crate::client::SyncKitClient;
26 27 use crate::error::{Result, SyncKitError};
@@ -154,8 +155,22 @@ impl<C: SyncTransport + BlobTransport> SyncStore<C> {
154 155 .await?;
155 156 self.maybe_snapshot().await?;
156 157
157 - let pushed = push_changes(&self.db, &*self.client, device_id).await?;
158 + let mut pushed = push_changes(&self.db, &*self.client, device_id).await?;
158 159 let pull = pull_changes(&self.db, &*self.client, &self.schema, device_id).await?;
160 + let mut pulled = pull.applied;
161 + let mut changed_tables = pull.changed_tables;
162 +
163 + // Group scopes: sync each group the user belongs to, under its own key and
164 + // cursor. A transport without groups reports none (the default), so this
165 + // loop is a no-op for personal-only apps.
166 + for (id, gck_version) in self.client.list_group_scopes().await? {
167 + let scope = SyncScope::Group { id, gck_version };
168 + pushed += push_scope(&self.db, &*self.client, device_id, scope).await?;
169 + let gp = pull_scope(&self.db, &*self.client, &self.schema, device_id, scope).await?;
170 + pulled += gp.applied;
171 + changed_tables.extend(gp.changed_tables);
172 + }
173 +
159 174 let blobs = match &self.blob_policy {
160 175 Some(policy) => sync_blobs(&self.db, &*self.client, policy).await?,
161 176 None => BlobOutcome::default(),
@@ -164,8 +179,8 @@ impl<C: SyncTransport + BlobTransport> SyncStore<C> {
164 179 self.finalize().await?;
165 180 Ok(SyncOutcome {
166 181 pushed,
167 - pulled: pull.applied,
168 - changed_tables: pull.changed_tables,
182 + pulled,
183 + changed_tables,
169 184 blobs,
170 185 })
171 186 }
@@ -354,11 +369,18 @@ mod tests {
354 369
355 370 /// A combined fake implementing both transports over a shared changelog log
356 371 /// and blob store, tagged with this device's registered id.
372 + type GroupLog = Arc<Mutex<Vec<(crate::ids::GroupId, DeviceId, ChangeEntry)>>>;
373 +
357 374 #[derive(Clone)]
358 375 struct Fake {
359 376 device_id: DeviceId,
360 377 log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
361 378 blobs: Arc<Mutex<HashMap<String, Vec<u8>>>>,
379 + /// Groups this member belongs to, reported by `list_group_scopes`.
380 + groups: Vec<(crate::ids::GroupId, i32)>,
381 + /// Shared group changelog (plaintext — the fake bypasses encryption, which
382 + /// is covered by the `client::groups` tests).
383 + group_log: GroupLog,
362 384 }
363 385 impl Fake {
364 386 fn new(device_id: u128, log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>) -> Self {
@@ -366,8 +388,17 @@ mod tests {
366 388 device_id: DeviceId::new(uuid::Uuid::from_u128(device_id)),
367 389 log,
368 390 blobs: Arc::new(Mutex::new(HashMap::new())),
391 + groups: Vec::new(),
392 + group_log: Arc::new(Mutex::new(Vec::new())),
369 393 }
370 394 }
395 +
396 + /// Enroll this member in a group sharing `group_log`.
397 + fn in_group(mut self, id: crate::ids::GroupId, version: i32, group_log: GroupLog) -> Self {
398 + self.groups.push((id, version));
399 + self.group_log = group_log;
400 + self
401 + }
371 402 }
372 403
373 404 impl SyncTransport for Fake {
@@ -424,6 +455,56 @@ mod tests {
424 455 Ok((out, l.len() as i64, false))
425 456 }
426 457 }
458 +
459 + fn list_group_scopes(
460 + &self,
461 + ) -> impl Future<Output = Result<Vec<(crate::ids::GroupId, i32)>>> + Send {
462 + let groups = self.groups.clone();
463 + async move { Ok(groups) }
464 + }
465 +
466 + fn group_scope_push(
467 + &self,
468 + group_id: crate::ids::GroupId,
469 + _gck_version: i32,
470 + device_id: DeviceId,
471 + changes: Vec<ChangeEntry>,
472 + ) -> impl Future<Output = Result<i64>> + Send {
473 + let group_log = self.group_log.clone();
474 + async move {
475 + let mut l = group_log.lock().unwrap();
476 + for c in changes {
477 + l.push((group_id, device_id, c));
478 + }
479 + Ok(l.len() as i64)
480 + }
481 + }
482 +
483 + fn group_scope_pull(
484 + &self,
485 + group_id: crate::ids::GroupId,
486 + _gck_version: i32,
487 + _device_id: DeviceId,
488 + cursor: i64,
489 + ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
490 + let group_log = self.group_log.clone();
491 + async move {
492 + let l = group_log.lock().unwrap();
493 + // Filter to this group, then page by global position (mirrors the
494 + // personal fake's seq scheme).
495 + let out: Vec<PulledChange> = l
496 + .iter()
497 + .enumerate()
498 + .filter(|(i, (gid, _, _))| *gid == group_id && (*i as i64 + 1) > cursor)
499 + .map(|(i, (_, dev, entry))| PulledChange {
500 + entry: entry.clone(),
501 + device_id: *dev,
502 + seq: i as i64 + 1,
503 + })
504 + .collect();
505 + Ok((out, l.len() as i64, false))
506 + }
507 + }
427 508 }
428 509
429 510 impl BlobTransport for Fake {
@@ -488,6 +569,84 @@ mod tests {
488 569 .build()
489 570 }
490 571
572 + // ── Group-scope sync ──
573 +
574 + fn group_schema() -> SyncSchema {
575 + SyncSchema::new(vec![
576 + SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
577 + ])
578 + }
579 +
580 + fn group_make_db(path: &std::path::Path) -> DbSource {
581 + let db = DbSource::path(path);
582 + let conn = db.open().unwrap();
583 + conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
584 + .unwrap();
585 + conn.execute_batch(&group_schema().migration_sql()).unwrap();
586 + db
587 + }
588 +
589 + fn group_store(db: DbSource, fake: Fake) -> SyncStore<Fake> {
590 + SyncStore::builder(db, fake, group_schema())
591 + .device("dev", "test")
592 + .build()
593 + }
594 +
595 + #[tokio::test]
596 + async fn two_members_converge_through_a_group_scope() {
597 + let dir = tempdir();
598 + let gid = crate::ids::GroupId::new(uuid::Uuid::from_u128(0x9));
599 + let group_log = Arc::new(Mutex::new(Vec::new()));
600 + let plog = Arc::new(Mutex::new(Vec::new()));
601 +
602 + let da = group_make_db(&dir.join("a.db"));
603 + let dbb = group_make_db(&dir.join("b.db"));
604 + let sa = group_store(
605 + da.clone(),
606 + Fake::new(0xA, plog.clone()).in_group(gid, 1, group_log.clone()),
607 + );
608 + let sb = group_store(
609 + dbb.clone(),
610 + Fake::new(0xB, plog.clone()).in_group(gid, 1, group_log.clone()),
611 + );
612 +
613 + // Member A creates a task in the group.
614 + da.open()
615 + .unwrap()
616 + .execute(
617 + "INSERT INTO task (id, name, group_id) VALUES ('t', 'from-A', ?1)",
618 + [gid.to_string()],
619 + )
620 + .unwrap();
621 +
622 + sa.sync_now().await.unwrap(); // pushes the group task to the shared log
623 + sb.sync_now().await.unwrap(); // pulls it and applies, stamping group_id
624 +
625 + // Member B now has the task, tagged with the group it came from (stamped
626 + // from the channel, not the payload — the payload carried only id + name).
627 + let (name, group_id): (String, String) = dbb
628 + .open()
629 + .unwrap()
630 + .query_row("SELECT name, group_id FROM task WHERE id = 't'", [], |r| {
631 + Ok((r.get(0)?, r.get(1)?))
632 + })
633 + .unwrap();
634 + assert_eq!(name, "from-A");
635 + assert_eq!(group_id, gid.to_string());
636 +
637 + // The group write never touched B's personal changelog scope.
638 + let personal_rows: i64 = dbb
639 + .open()
640 + .unwrap()
641 + .query_row(
642 + "SELECT COUNT(*) FROM sync_changelog WHERE scope = ''",
643 + [],
644 + |r| r.get(0),
645 + )
646 + .unwrap();
647 + assert_eq!(personal_rows, 0);
648 + }
649 +
491 650 /// Pre-stamp a device's pending edit at a controlled time so cross-device HLC
492 651 /// ordering is deterministic (sync_now's internal stamp is then a no-op).
493 652 fn stamp_pending(db: &DbSource, node: DeviceId, now_ms: i64) {
M shared/synckit-client/src/store/hlc.rs +2 -2
@@ -373,7 +373,7 @@ mod tests {
373 373 ) {
374 374 let now = Utc::now();
375 375 let resolved = resolve_pull(conn, s, node, pulled, now).unwrap();
376 - apply_remote_changes(conn, s, &resolved).unwrap();
376 + apply_remote_changes(conn, s, &resolved, "").unwrap();
377 377 record_committed(conn, &resolved).unwrap();
378 378 }
379 379
@@ -500,7 +500,7 @@ mod tests {
500 500 let (src2, sn2) = device(3);
501 501 let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
502 502 let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now()).unwrap();
503 - apply_remote_changes(&mut a, &s, &resolved).unwrap();
503 + apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
504 504 assert_eq!(
505 505 note_name(&a, "r").as_deref(),
506 506 Some("second"),
M shared/synckit-client/src/store/mod.rs +1 -1
@@ -45,5 +45,5 @@ pub use schema::{ConflictStrategy, DeleteMode, RowIdScheme, SyncMode, SyncSchema
45 45 pub use sync::{
46 46 PUSH_BATCH_LIMIT, PullOutcome, SyncScope, SyncTransport, cleanup_changelog,
47 47 create_initial_snapshot, enforce_changelog_retention, ensure_device_registered, pull_changes,
48 - push_changes, push_scope,
48 + pull_scope, push_changes, push_scope,
49 49 };
M shared/synckit-client/src/store/sync.rs +49 -23
@@ -19,7 +19,9 @@ use chrono::Utc;
19 19 use rusqlite::Connection;
20 20
21 21 use super::apply::{ApplyOutcome, apply_remote_changes};
22 - use super::db::{DbSource, device_id, get_sync_state_or, set_device_id, set_sync_state};
22 + use super::db::{
23 + DbSource, device_id, get_scope_cursor, set_device_id, set_scope_cursor, set_sync_state,
24 + };
23 25 use super::hlc::{record_committed, resolve_pull, set_committed, stamp_pending};
24 26 use super::migrate::{json_object, row_id_expr};
25 27 use super::schema::{SyncMode, SyncSchema};
@@ -411,42 +413,62 @@ fn mark_pushed_committed(
411 413
412 414 // -- pull --
413 415
414 - /// Pull, resolve, and apply all remote changes, page by page.
415 - ///
416 - /// Follows the SDK contract: the cursor is advanced only *after* the batch is
417 - /// applied and committed, in the same blocking step, so a crash re-pulls a batch
418 - /// the idempotent apply/gate absorbs.
416 + /// Pull, resolve, and apply all remote changes for the **personal** scope. Thin
417 + /// wrapper over [`pull_scope`] preserved for the personal call sites.
419 418 pub async fn pull_changes<T: SyncTransport>(
420 419 db: &DbSource,
421 420 client: &T,
422 421 schema: &SyncSchema,
423 422 device_id: DeviceId,
424 423 ) -> Result<PullOutcome> {
424 + pull_scope(db, client, schema, device_id, SyncScope::Personal).await
425 + }
426 +
427 + /// Pull, resolve, and apply all remote changes for one `scope`, page by page.
428 + ///
429 + /// Reads from the scope's own cursor (`sync_scope_cursor`), pulls from the
430 + /// scope's endpoint (personal pull, or a group pull under its GCK), and applies
431 + /// with the scope stamped so group rows are tagged with their group. Follows the
432 + /// SDK contract: the cursor advances only *after* the batch is applied and
433 + /// committed, in the same blocking step, so a crash re-pulls a batch the
434 + /// idempotent apply/gate absorbs.
435 + pub async fn pull_scope<T: SyncTransport>(
436 + db: &DbSource,
437 + client: &T,
438 + schema: &SyncSchema,
439 + device_id: DeviceId,
440 + scope: SyncScope,
441 + ) -> Result<PullOutcome> {
442 + let scope_key = scope.key();
425 443 let mut out = PullOutcome::default();
426 444 loop {
427 445 let db_cursor = db.clone();
428 - let cursor = tokio::task::spawn_blocking(move || -> Result<i64> {
429 - Ok(get_sync_state_or(&db_cursor.open()?, "pull_cursor", "0")?
430 - .parse()
431 - .unwrap_or(0))
432 - })
433 - .await
434 - .map_err(join_err)??;
446 + let sk = scope_key.clone();
447 + let cursor = tokio::task::spawn_blocking(move || get_scope_cursor(&db_cursor.open()?, &sk))
448 + .await
449 + .map_err(join_err)??;
435 450
436 - let (pulled, new_cursor, has_more) = client.pull_rich(device_id, cursor).await?;
451 + let (pulled, new_cursor, has_more) = match scope {
452 + SyncScope::Personal => client.pull_rich(device_id, cursor).await?,
453 + SyncScope::Group { id, gck_version } => {
454 + client
455 + .group_scope_pull(id, gck_version, device_id, cursor)
456 + .await?
457 + }
458 + };
437 459
438 460 if pulled.is_empty() {
439 461 let db_c = db.clone();
440 - tokio::task::spawn_blocking(move || {
441 - set_sync_state(&db_c.open()?, "pull_cursor", &new_cursor.to_string())
442 - })
443 - .await
444 - .map_err(join_err)??;
462 + let sk = scope_key.clone();
463 + tokio::task::spawn_blocking(move || set_scope_cursor(&db_c.open()?, &sk, new_cursor))
464 + .await
465 + .map_err(join_err)??;
445 466 break;
446 467 }
447 468
448 469 let db_apply = db.clone();
449 470 let schema = schema.clone();
471 + let sk = scope_key.clone();
450 472 let outcome = tokio::task::spawn_blocking(move || {
451 473 apply_pull(
452 474 &mut db_apply.open()?,
@@ -454,6 +476,7 @@ pub async fn pull_changes<T: SyncTransport>(
454 476 device_id,
455 477 pulled,
456 478 new_cursor,
479 + &sk,
457 480 )
458 481 })
459 482 .await
@@ -474,11 +497,12 @@ fn apply_pull(
474 497 device_id: DeviceId,
475 498 pulled: Vec<PulledChange>,
476 499 new_cursor: i64,
500 + scope_key: &str,
477 501 ) -> Result<ApplyOutcome> {
478 502 let resolved = resolve_pull(conn, schema, device_id, pulled, Utc::now())?;
479 - let outcome = apply_remote_changes(conn, schema, &resolved)?;
503 + let outcome = apply_remote_changes(conn, schema, &resolved, scope_key)?;
480 504 record_committed(conn, &resolved)?;
481 - set_sync_state(conn, "pull_cursor", &new_cursor.to_string())?;
505 + set_scope_cursor(conn, scope_key, new_cursor)?;
482 506 Ok(outcome)
483 507 }
484 508
@@ -545,6 +569,7 @@ pub fn enforce_changelog_retention(conn: &Connection, cap: i64) -> Result<u64> {
545 569
546 570 #[cfg(test)]
547 571 mod tests {
572 + use super::super::db::get_sync_state_or;
548 573 use super::super::schema::SyncTable;
549 574 use super::*;
550 575 use std::sync::{Arc, Mutex};
@@ -777,8 +802,9 @@ mod tests {
777 802 let out = pull_changes(&db_, &server, &schema(), nb).await.unwrap();
778 803 assert_eq!(out.applied, 2);
779 804 assert_eq!(note_name(&db_, "r1").as_deref(), Some("x"));
780 - let cursor = get_sync_state_or(&db_.open().unwrap(), "pull_cursor", "0").unwrap();
781 - assert_eq!(cursor, "2");
805 + // Personal pull advances the personal ('') scope cursor.
806 + let cursor = get_scope_cursor(&db_.open().unwrap(), "").unwrap();
807 + assert_eq!(cursor, 2);
782 808 }
783 809
784 810 #[test]