Skip to main content

max / makenotwork

synckit: multi-scope changelog foundation (Groups phase 3, slice 1) The manifest + DDL layer for syncing a personal scope plus N group scopes. Additive: an app that never calls group_scoped behaves byte-identically to today. Design approved in wiki synckit-multiscope-design. - schema: SyncTable::group_scoped(col) declares a table's rows carry a local group_id provenance column and may belong to group scopes. Un-annotated tables stay personal-only (config/credentials can't leak into a group). - migrate: sync_changelog gains a `scope` column ('' = personal, else the group id), set by the triggers from the row's group_scoped column via COALESCE(alias.col, ''); personal-only tables always emit ''. The single pull_cursor becomes a per-scope sync_scope_cursor table, personal seeded from the legacy value. - db: ensure_scope_schema idempotently upgrades a pre-Groups database (guarded ALTER to add the scope column, since CREATE IF NOT EXISTS cannot and SQLite has no ADD COLUMN IF NOT EXISTS), run from configure_connection. Per-scope cursor accessors get_scope_cursor / set_scope_cursor. - tests: group_scoped table stamps scope from group_id (insert/update/delete), personal-only table always '', legacy changelog upgrade + cursor migration. Scope is a transport-channel property mirrored by a local provenance column; it never rides in the wire payload. Next slices: transport group methods, GCK resolver, sync_now scope iteration, apply-side stamping. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 19:16 UTC
Commit: 8297ee2f22ba272e36ebd1f8e065bccdaf31c6fa
Parent: bf1be38
4 files changed, +284 insertions, -7 deletions
M shared/synckit-client/src/store/db.rs +124
@@ -76,6 +76,77 @@ pub(crate) fn configure_connection(conn: &Connection) -> Result<()> {
76 76 // the default the apply path relaxes only for `references_unsynced` tables.
77 77 conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
78 78 register_hash_row_id(conn)?;
79 + ensure_scope_schema(conn)?;
80 + Ok(())
81 + }
82 +
83 + /// Idempotently bring a pre-Groups bookkeeping schema up to the multi-scope
84 + /// shape: add `sync_changelog.scope` if the table exists without it, and ensure
85 + /// `sync_scope_cursor` exists (seeded from the old single `pull_cursor`).
86 + ///
87 + /// `CREATE TABLE IF NOT EXISTS` in [`SyncSchema::migration_sql`] handles a fresh
88 + /// DB, but it cannot *add a column* to an existing changelog and SQLite has no
89 + /// `ADD COLUMN IF NOT EXISTS`, so the ALTER is guarded on `pragma_table_info`
90 + /// here. On a fresh DB (no bookkeeping yet) this is a no-op — `migration_sql`
91 + /// creates everything with the column. Runs on every connection open; the guards
92 + /// make it a cheap no-op once applied.
93 + pub(crate) fn ensure_scope_schema(conn: &Connection) -> Result<()> {
94 + let bookkeeping_exists = conn
95 + .query_row(
96 + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'sync_changelog'",
97 + [],
98 + |_| Ok(()),
99 + )
100 + .optional()?
101 + .is_some();
102 + if !bookkeeping_exists {
103 + return Ok(());
104 + }
105 +
106 + let has_scope = conn
107 + .query_row(
108 + "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
109 + [],
110 + |_| Ok(()),
111 + )
112 + .optional()?
113 + .is_some();
114 + if !has_scope {
115 + conn.execute_batch(
116 + "ALTER TABLE sync_changelog ADD COLUMN scope TEXT NOT NULL DEFAULT '';",
117 + )?;
118 + }
119 +
120 + conn.execute_batch(
121 + "CREATE TABLE IF NOT EXISTS sync_scope_cursor (
122 + scope TEXT PRIMARY KEY NOT NULL,
123 + cursor INTEGER NOT NULL
124 + );
125 + INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
126 + SELECT '', CAST(value AS INTEGER) FROM sync_state WHERE key = 'pull_cursor';",
127 + )?;
128 + Ok(())
129 + }
130 +
131 + /// Read a scope's pull cursor (`''` = personal), defaulting to `0` when absent.
132 + pub fn get_scope_cursor(conn: &Connection, scope: &str) -> Result<i64> {
133 + conn.query_row(
134 + "SELECT cursor FROM sync_scope_cursor WHERE scope = ?1",
135 + [scope],
136 + |r| r.get(0),
137 + )
138 + .optional()
139 + .map(|o| o.unwrap_or(0))
140 + .map_err(Into::into)
141 + }
142 +
143 + /// Persist a scope's pull cursor.
144 + pub fn set_scope_cursor(conn: &Connection, scope: &str, cursor: i64) -> Result<()> {
145 + conn.execute(
146 + "INSERT INTO sync_scope_cursor (scope, cursor) VALUES (?1, ?2) \
147 + ON CONFLICT(scope) DO UPDATE SET cursor = excluded.cursor",
148 + (scope, cursor),
149 + )?;
79 150 Ok(())
80 151 }
81 152
@@ -340,6 +411,59 @@ mod tests {
340 411 }
341 412
342 413 #[test]
414 + fn ensure_scope_schema_upgrades_a_legacy_changelog() {
415 + // Simulate a pre-Groups database: a sync_changelog WITHOUT the scope
416 + // column, and a pull_cursor mid-stream at 42.
417 + let conn = Connection::open_in_memory().unwrap();
418 + conn.execute_batch(
419 + "CREATE TABLE sync_state (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL);
420 + INSERT INTO sync_state (key, value) VALUES ('pull_cursor', '42');
421 + CREATE TABLE sync_changelog (
422 + id INTEGER PRIMARY KEY AUTOINCREMENT,
423 + table_name TEXT NOT NULL, op TEXT NOT NULL, row_id TEXT NOT NULL,
424 + data TEXT, pushed INTEGER NOT NULL DEFAULT 0
425 + );",
426 + )
427 + .unwrap();
428 + // No scope column yet.
429 + let has_scope_before: bool = conn
430 + .query_row(
431 + "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
432 + [],
433 + |_| Ok(true),
434 + )
435 + .optional()
436 + .unwrap()
437 + .unwrap_or(false);
438 + assert!(!has_scope_before);
439 +
440 + // Idempotent: run it twice.
441 + ensure_scope_schema(&conn).unwrap();
442 + ensure_scope_schema(&conn).unwrap();
443 +
444 + // The scope column now exists and defaults to personal.
445 + conn.execute(
446 + "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('t','INSERT','r')",
447 + [],
448 + )
449 + .unwrap();
450 + let scope: String = conn
451 + .query_row(
452 + "SELECT scope FROM sync_changelog WHERE row_id = 'r'",
453 + [],
454 + |r| r.get(0),
455 + )
456 + .unwrap();
457 + assert_eq!(scope, "");
458 +
459 + // The personal cursor was migrated from the legacy pull_cursor value.
460 + assert_eq!(get_scope_cursor(&conn, "").unwrap(), 42);
461 + assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 0); // absent -> 0
462 + set_scope_cursor(&conn, "grp-1", 7).unwrap();
463 + assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 7);
464 + }
465 +
466 + #[test]
343 467 fn salt_is_provisioned_once_and_stable_across_reruns() {
344 468 let conn = Connection::open_in_memory().unwrap();
345 469 configure_connection(&conn).unwrap();
M shared/synckit-client/src/store/migrate.rs +143 -5
@@ -40,11 +40,25 @@ CREATE TABLE IF NOT EXISTS sync_changelog (
40 40 data TEXT,
41 41 pushed INTEGER NOT NULL DEFAULT 0,
42 42 hlc_wall INTEGER,
43 - hlc_counter INTEGER
43 + hlc_counter INTEGER,
44 + -- SyncKit Groups scope: '' = personal, otherwise the group id this change
45 + -- belongs to. Set by the triggers from the row's group_scoped column. An
46 + -- existing (pre-Groups) changelog is migrated to add this column by
47 + -- `ensure_scope_schema` (db.rs), since CREATE IF NOT EXISTS cannot alter it.
48 + scope TEXT NOT NULL DEFAULT ''
44 49 );
45 50 CREATE INDEX IF NOT EXISTS idx_sync_changelog_unpushed
46 51 ON sync_changelog(pushed) WHERE pushed = 0;
47 52
53 + -- Per-scope pull cursor. '' holds the personal cursor, migrated once from the
54 + -- pre-Groups single `pull_cursor` sync_state value; each group id gets its own.
55 + CREATE TABLE IF NOT EXISTS sync_scope_cursor (
56 + scope TEXT PRIMARY KEY NOT NULL,
57 + cursor INTEGER NOT NULL
58 + );
59 + INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
60 + VALUES ('', CAST((SELECT value FROM sync_state WHERE key = 'pull_cursor') AS INTEGER));
61 +
48 62 -- Committed HLC per row (conflict gating). Schema may be refined in B4.
49 63 CREATE TABLE IF NOT EXISTS sync_committed_hlc (
50 64 table_name TEXT NOT NULL,
@@ -126,12 +140,22 @@ fn trigger(name: &str, timing: &str, table: &str, when: &str, values: &str) -> S
126 140 AFTER {timing} ON {table}\n\
127 141 WHEN {when}\n\
128 142 BEGIN\n\
129 - INSERT INTO sync_changelog (table_name, op, row_id, data)\n\
143 + INSERT INTO sync_changelog (table_name, op, row_id, data, scope)\n\
130 144 VALUES {values};\n\
131 145 END;\n"
132 146 )
133 147 }
134 148
149 + /// The changelog `scope` expression for a table under the given row alias. A
150 + /// `group_scoped` table reads its provenance column (`NULL` -> personal `''`);
151 + /// every other table is personal-only, so its scope is the literal `''`.
152 + fn scope_expr(table: &SyncTable, alias: &str) -> String {
153 + match table.group_scope {
154 + Some(col) => format!("COALESCE({alias}.{col}, '')"),
155 + None => "''".to_string(),
156 + }
157 + }
158 +
135 159 /// The INSERT/UPDATE/DELETE trigger set for one table.
136 160 fn triggers_for(table: &SyncTable) -> String {
137 161 let name = table.name;
@@ -142,9 +166,10 @@ fn triggers_for(table: &SyncTable) -> String {
142 166 // (it only reflects state changes on rows that already exist locally).
143 167 if matches!(table.mode, SyncMode::Full) {
144 168 let values = format!(
145 - "('{name}', 'INSERT', {}, {})",
169 + "('{name}', 'INSERT', {}, {}, {})",
146 170 row_id_expr(table, "NEW"),
147 171 json_object("NEW", &emitted),
172 + scope_expr(table, "NEW"),
148 173 );
149 174 out.push_str(&trigger(
150 175 &format!("sync_{name}_insert"),
@@ -169,9 +194,10 @@ fn triggers_for(table: &SyncTable) -> String {
169 194 }
170 195 };
171 196 let update_values = format!(
172 - "('{name}', 'UPDATE', {}, {})",
197 + "('{name}', 'UPDATE', {}, {}, {})",
173 198 row_id_expr(table, "NEW"),
174 199 json_object("NEW", &emitted),
200 + scope_expr(table, "NEW"),
175 201 );
176 202 out.push_str(&trigger(
177 203 &format!("sync_{name}_update"),
@@ -187,9 +213,10 @@ fn triggers_for(table: &SyncTable) -> String {
187 213 // reconstructs the WHERE clause without the (possibly hashed) row-id.
188 214 if !matches!(table.deletes, DeleteMode::Ignore) {
189 215 let delete_values = format!(
190 - "('{name}', 'DELETE', {}, {})",
216 + "('{name}', 'DELETE', {}, {}, {})",
191 217 row_id_expr(table, "OLD"),
192 218 json_object("OLD", table.pk),
219 + scope_expr(table, "OLD"),
193 220 );
194 221 out.push_str(&trigger(
195 222 &format!("sync_{name}_delete"),
@@ -449,6 +476,117 @@ mod tests {
449 476 assert_eq!(log[0].row_id, "theme");
450 477 }
451 478
479 + /// A row observed in `sync_changelog` including its scope.
480 + #[derive(Debug, PartialEq, Eq)]
481 + struct ScopedEntry {
482 + op: String,
483 + row_id: String,
484 + scope: String,
485 + }
486 +
487 + fn scoped_changelog(conn: &Connection) -> Vec<ScopedEntry> {
488 + let mut stmt = conn
489 + .prepare("SELECT op, row_id, scope FROM sync_changelog ORDER BY id")
490 + .unwrap();
491 + let rows = stmt
492 + .query_map([], |r| {
493 + Ok(ScopedEntry {
494 + op: r.get(0)?,
495 + row_id: r.get(1)?,
496 + scope: r.get(2)?,
497 + })
498 + })
499 + .unwrap();
500 + rows.map(|r| r.unwrap()).collect()
501 + }
502 +
503 + #[test]
504 + fn group_scoped_table_stamps_scope_from_group_id() {
505 + let conn = Connection::open_in_memory().unwrap();
506 + conn.execute_batch(
507 + "CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);
508 + CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
509 + )
510 + .unwrap();
511 + let schema = SyncSchema::new(vec![
512 + SyncTable::full("task", &["id", "name", "group_id"]).group_scoped("group_id"),
513 + SyncTable::full("cfg", &["key", "value"]).pk(&["key"]),
514 + ]);
515 + conn.execute_batch(&schema.migration_sql()).unwrap();
516 +
517 + // A personal task row (group_id NULL) -> scope ''.
518 + conn.execute(
519 + "INSERT INTO task (id, name, group_id) VALUES ('t1', 'a', NULL)",
520 + [],
521 + )
522 + .unwrap();
523 + // A group task row -> scope = the group id.
524 + conn.execute(
525 + "INSERT INTO task (id, name, group_id) VALUES ('t2', 'b', 'grp-9')",
526 + [],
527 + )
528 + .unwrap();
529 + // Moving a task into a group is an UPDATE that re-scopes it.
530 + conn.execute("UPDATE task SET group_id = 'grp-9' WHERE id = 't1'", [])
531 + .unwrap();
532 + // A personal-only (non-group_scoped) table is always scope ''.
533 + conn.execute("INSERT INTO cfg (key, value) VALUES ('theme', 'dark')", [])
534 + .unwrap();
535 + // A delete carries the group scope of the deleted row (OLD).
536 + conn.execute("DELETE FROM task WHERE id = 't2'", [])
537 + .unwrap();
538 +
539 + let log = scoped_changelog(&conn);
540 + assert_eq!(
541 + log,
542 + vec![
543 + ScopedEntry {
544 + op: "INSERT".into(),
545 + row_id: "t1".into(),
546 + scope: "".into()
547 + },
548 + ScopedEntry {
549 + op: "INSERT".into(),
550 + row_id: "t2".into(),
551 + scope: "grp-9".into()
552 + },
553 + ScopedEntry {
554 + op: "UPDATE".into(),
555 + row_id: "t1".into(),
556 + scope: "grp-9".into()
557 + },
558 + ScopedEntry {
559 + op: "INSERT".into(),
560 + row_id: "theme".into(),
561 + scope: "".into()
562 + },
563 + ScopedEntry {
564 + op: "DELETE".into(),
565 + row_id: "t2".into(),
566 + scope: "grp-9".into()
567 + },
568 + ]
569 + );
570 + }
571 +
572 + #[test]
573 + fn scope_cursor_seeded_from_legacy_pull_cursor() {
574 + let conn = Connection::open_in_memory().unwrap();
575 + let schema = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
576 + conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
577 + .unwrap();
578 + conn.execute_batch(&schema.migration_sql()).unwrap();
579 + // Personal scope cursor exists, seeded from pull_cursor's default '0'.
580 + let personal: i64 = conn
581 + .query_row(
582 + "SELECT cursor FROM sync_scope_cursor WHERE scope = ''",
583 + [],
584 + |r| r.get(0),
585 + )
586 + .unwrap();
587 + assert_eq!(personal, 0);
588 + }
589 +
452 590 #[test]
453 591 fn salt_seeded_only_when_a_table_is_hashed() {
454 592 // schema() has hashed tables → salt present.
M shared/synckit-client/src/store/mod.rs +2 -2
@@ -32,8 +32,8 @@ pub use blob::{
32 32 upload_blobs,
33 33 };
34 34 pub use db::{
35 - DbSource, clear_applying_remote, device_id, get_sync_state, get_sync_state_or, set_device_id,
36 - set_sync_state, with_applying_remote,
35 + DbSource, clear_applying_remote, device_id, get_scope_cursor, get_sync_state,
36 + get_sync_state_or, set_device_id, set_scope_cursor, set_sync_state, with_applying_remote,
37 37 };
38 38 pub use facade::{SchedulerHandle, SyncConfig, SyncOutcome, SyncStore, SyncStoreBuilder};
39 39 pub use hlc::{
M shared/synckit-client/src/store/schema.rs +15
@@ -83,6 +83,7 @@ pub struct SyncTable {
83 83 pub(crate) insert_defaults: &'static [(&'static str, &'static str)],
84 84 pub(crate) references_unsynced: bool,
85 85 pub(crate) exclude_where: Option<&'static str>,
86 + pub(crate) group_scope: Option<&'static str>,
86 87 }
87 88
88 89 impl SyncTable {
@@ -101,6 +102,7 @@ impl SyncTable {
101 102 insert_defaults: &[],
102 103 references_unsynced: false,
103 104 exclude_where: None,
105 + group_scope: None,
104 106 }
105 107 }
106 108
@@ -167,6 +169,19 @@ impl SyncTable {
167 169 self
168 170 }
169 171
172 + /// Declare that this table's rows carry a `group_id` provenance column and may
173 + /// belong to group scopes (SyncKit Groups). `col` is the local column name
174 + /// (`NULL` = personal, a group id = that group's shared changelog). The trigger
175 + /// reads it to route a change to the right scope; the apply side stamps it from
176 + /// the scope being pulled. The value is local provenance only — it never rides
177 + /// in the encrypted wire payload. Un-annotated tables are personal-only, so a
178 + /// config/credential table can never leak into a group. See the wiki design
179 + /// note `synckit-multiscope-design`.
180 + pub fn group_scoped(mut self, col: &'static str) -> Self {
181 + self.group_scope = Some(col);
182 + self
183 + }
184 +
170 185 /// The columns emitted into `sync_changelog.data` for an INSERT/UPDATE:
171 186 /// every whitelisted column for `Full`, or the primary key plus the `set`
172 187 /// columns (deduped, PK first) for `PartialUpdate`.