Introduce table for partial sync completions

Author: simolus3

crates/core/src/kv.rs

@@ -46,13 +46,14 @@ fn powersync_last_synced_at_impl(
     let db = ctx.db_handle();
 
     // language=SQLite
-    let statement = db.prepare_v2("select value from ps_kv where key = 'last_synced_at'")?;
+    let statement =
+        db.prepare_v2("select last_synced_at from ps_sync_state where priority = -1")?;
 
     if statement.step()? == ResultCode::ROW {
         let client_id = statement.column_text(0)?;
-        return Ok(Some(client_id.to_string()));
+        Ok(Some(client_id.to_string()))
     } else {
-        return Ok(None);
+        Ok(None)
     }
 }
 

crates/core/src/migrations.rs

@@ -310,5 +310,28 @@ json_array(
             .into_db_result(local_db)?;
     }
 
+    if current_version < 7 && target_version >= 7 {
+        local_db
+            .exec_safe(
+                "\
+CREATE TABLE ps_sync_state (
+  priority INTEGER NOT NULL,
+  last_synced_at TEXT NOT NULL
+) STRICT;
+INSERT OR IGNORE INTO ps_sync_state (priority, last_synced_at)
+  SELECT -1, value from ps_kv where key = 'last_synced_at';
+
+INSERT INTO ps_migration(id, down_migrations)
+VALUES(7,
+json_array(
+json_object('sql', 'INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = -1'),
+json_object('sql', 'DROP TABLE ps_sync_state'),
+json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7')
+));
+",
+            )
+            .into_db_result(local_db)?;
+    }
+
     Ok(())
 }

crates/core/src/sync_local.rs

@@ -1,4 +1,7 @@
+use core::ffi::c_void;
+
 use alloc::collections::BTreeSet;
+use alloc::fmt::format;
 use alloc::format;
 use alloc::string::String;
 use alloc::vec::Vec;
@@ -288,19 +291,35 @@ GROUP BY b.row_type, b.row_id",
     }
 
     fn mark_completed(&self) -> Result<(), SQLiteError> {
-        if self.partial.is_none() {
-            // language=SQLite
-            self.db
-                .exec_safe("DELETE FROM ps_updated_rows")
-                .into_db_result(self.db)?;
+        let priority_code = match &self.partial {
+            None => {
+                // language=SQLite
+                self.db
+                    .exec_safe("DELETE FROM ps_updated_rows")
+                    .into_db_result(self.db)?;
+                -1
+            }
+            Some(partial) => partial.priority.into(),
+        };
 
-            // language=SQLite
-            self.db
-                .exec_safe(
-                    "insert or replace into ps_kv(key, value) values('last_synced_at', datetime())",
-                )
-                .into_db_result(self.db)?;
-        }
+        // Higher-priority buckets are always part of lower-priority sync operations too, so we can
+        // delete information about higher-priority syncs (represented as lower priority numbers).
+        // A complete sync is represented as -1.
+        // language=SQLite
+        let stmt = self
+            .db
+            .prepare_v2("DELETE FROM ps_sync_state WHERE (priority < ?1) OR (?1 = -1);")
+            .into_db_result(self.db)?;
+        stmt.bind_int(1, priority_code)?;
+        stmt.exec()?;
+
+        // language=SQLite
+        let stmt = self
+            .db
+            .prepare_v2("INSERT OR REPLACE INTO ps_sync_state (priority, last_synced_at) VALUES (?, datetime());")
+            .into_db_result(self.db)?;
+        stmt.bind_int(1, priority_code)?;
+        stmt.exec()?;
 
         Ok(())
     }

crates/core/src/view_admin.rs

@@ -120,7 +120,7 @@ fn powersync_init_impl(
 
     setup_internal_views(local_db)?;
 
-    powersync_migrate(ctx, 6)?;
+    powersync_migrate(ctx, 7)?;
 
     Ok(String::from(""))
 }

dart/test/sync_test.dart

@@ -177,6 +177,9 @@ void main() {
         expect(fetchRows(), [
           for (var j = 0; j <= i; j++) {'id': 'row-$j', 'col': '$j'},
         ]);
+
+        expect(db.select('select 1 from ps_sync_state where priority = ?', [i]),
+            isNotEmpty);
       }
     });
   });

dart/test/utils/migration_fixtures.dart

@@ -1,5 +1,5 @@
 /// The current database version
-const databaseVersion = 6;
+const databaseVersion = 7;
 
 /// This is the base database state that we expect at various schema versions.
 /// Generated by loading the specific library version, and exporting the schema.
@@ -172,7 +172,51 @@ const expectedState = <int, String>{
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(4, '[{"sql":"DELETE FROM ps_migration WHERE id >= 4"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN op_checksum"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN remove_operations"}]')
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(5, '[{"sql":"SELECT powersync_drop_view(view.name)\n  FROM sqlite_master view\n  WHERE view.type = ''view''\n    AND view.sql GLOB  ''*-- powersync-auto-generated''"},{"sql":"ALTER TABLE ps_buckets RENAME TO ps_buckets_5"},{"sql":"ALTER TABLE ps_oplog RENAME TO ps_oplog_5"},{"sql":"CREATE TABLE ps_buckets(\n  name TEXT PRIMARY KEY,\n  last_applied_op INTEGER NOT NULL DEFAULT 0,\n  last_op INTEGER NOT NULL DEFAULT 0,\n  target_op INTEGER NOT NULL DEFAULT 0,\n  add_checksum INTEGER NOT NULL DEFAULT 0,\n  pending_delete INTEGER NOT NULL DEFAULT 0\n, op_checksum INTEGER NOT NULL DEFAULT 0, remove_operations INTEGER NOT NULL DEFAULT 0)"},{"sql":"INSERT INTO ps_buckets(name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete)\n    SELECT name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete FROM ps_buckets_5"},{"sql":"CREATE TABLE ps_oplog(\n  bucket TEXT NOT NULL,\n  op_id INTEGER NOT NULL,\n  op INTEGER NOT NULL,\n  row_type TEXT,\n  row_id TEXT,\n  key TEXT,\n  data TEXT,\n  hash INTEGER NOT NULL,\n  superseded INTEGER NOT NULL)"},{"sql":"CREATE INDEX ps_oplog_by_row ON ps_oplog (row_type, row_id) WHERE superseded = 0"},{"sql":"CREATE INDEX ps_oplog_by_opid ON ps_oplog (bucket, op_id)"},{"sql":"CREATE INDEX ps_oplog_by_key ON ps_oplog (bucket, key) WHERE superseded = 0"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, key, data, hash, superseded)\n    SELECT ps_buckets_5.name, oplog.op_id, 3, oplog.row_type, oplog.row_id, oplog.key, oplog.data, oplog.hash, 0\n    FROM ps_oplog_5 oplog\n    JOIN ps_buckets_5\n        ON ps_buckets_5.id = oplog.bucket"},{"sql":"DROP TABLE ps_oplog_5"},{"sql":"DROP TABLE ps_buckets_5"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, hash, superseded)\n    SELECT ''$local'', 1, 4, r.row_type, r.row_id, 0, 0\n    FROM ps_updated_rows r"},{"sql":"INSERT OR REPLACE INTO ps_buckets(name, pending_delete, last_op, target_op) VALUES(''$local'', 1, 0, 9223372036854775807)"},{"sql":"DROP TABLE ps_updated_rows"},{"sql":"DELETE FROM ps_migration WHERE id >= 5"}]')
 ;INSERT INTO ps_migration(id, down_migrations) VALUES(6, '[{"sql":"DELETE FROM ps_migration WHERE id >= 6"}]')
-'''
+''',
+  7: r'''
+;CREATE TABLE ps_buckets(
+  id INTEGER PRIMARY KEY,
+  name TEXT NOT NULL,
+  last_applied_op INTEGER NOT NULL DEFAULT 0,
+  last_op INTEGER NOT NULL DEFAULT 0,
+  target_op INTEGER NOT NULL DEFAULT 0,
+  add_checksum INTEGER NOT NULL DEFAULT 0,
+  op_checksum INTEGER NOT NULL DEFAULT 0,
+  pending_delete INTEGER NOT NULL DEFAULT 0
+) STRICT
+;CREATE TABLE ps_crud (id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT, tx_id INTEGER)
+;CREATE TABLE ps_kv(key TEXT PRIMARY KEY NOT NULL, value BLOB)
+;CREATE TABLE ps_migration(id INTEGER PRIMARY KEY, down_migrations TEXT)
+;CREATE TABLE ps_oplog(
+  bucket INTEGER NOT NULL,
+  op_id INTEGER NOT NULL,
+  row_type TEXT,
+  row_id TEXT,
+  key TEXT,
+  data TEXT,
+  hash INTEGER NOT NULL) STRICT
+;CREATE TABLE ps_sync_state (
+  priority INTEGER NOT NULL,
+  last_synced_at TEXT NOT NULL
+) STRICT
+;CREATE TABLE ps_tx(id INTEGER PRIMARY KEY NOT NULL, current_tx INTEGER, next_tx INTEGER)
+;CREATE TABLE ps_untyped(type TEXT NOT NULL, id TEXT NOT NULL, data TEXT, PRIMARY KEY (type, id))
+;CREATE TABLE ps_updated_rows(
+  row_type TEXT,
+  row_id TEXT,
+  PRIMARY KEY(row_type, row_id)) STRICT, WITHOUT ROWID
+;CREATE UNIQUE INDEX ps_buckets_name ON ps_buckets (name)
+;CREATE INDEX ps_oplog_key ON ps_oplog (bucket, key)
+;CREATE INDEX ps_oplog_opid ON ps_oplog (bucket, op_id)
+;CREATE INDEX ps_oplog_row ON ps_oplog (row_type, row_id)
+;INSERT INTO ps_migration(id, down_migrations) VALUES(1, null)
+;INSERT INTO ps_migration(id, down_migrations) VALUES(2, '[{"sql":"DELETE FROM ps_migration WHERE id >= 2","params":[]},{"sql":"DROP TABLE ps_tx","params":[]},{"sql":"ALTER TABLE ps_crud DROP COLUMN tx_id","params":[]}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(3, '[{"sql":"DELETE FROM ps_migration WHERE id >= 3"},{"sql":"DROP TABLE ps_kv"}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(4, '[{"sql":"DELETE FROM ps_migration WHERE id >= 4"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN op_checksum"},{"sql":"ALTER TABLE ps_buckets DROP COLUMN remove_operations"}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(5, '[{"sql":"SELECT powersync_drop_view(view.name)\n  FROM sqlite_master view\n  WHERE view.type = ''view''\n    AND view.sql GLOB  ''*-- powersync-auto-generated''"},{"sql":"ALTER TABLE ps_buckets RENAME TO ps_buckets_5"},{"sql":"ALTER TABLE ps_oplog RENAME TO ps_oplog_5"},{"sql":"CREATE TABLE ps_buckets(\n  name TEXT PRIMARY KEY,\n  last_applied_op INTEGER NOT NULL DEFAULT 0,\n  last_op INTEGER NOT NULL DEFAULT 0,\n  target_op INTEGER NOT NULL DEFAULT 0,\n  add_checksum INTEGER NOT NULL DEFAULT 0,\n  pending_delete INTEGER NOT NULL DEFAULT 0\n, op_checksum INTEGER NOT NULL DEFAULT 0, remove_operations INTEGER NOT NULL DEFAULT 0)"},{"sql":"INSERT INTO ps_buckets(name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete)\n    SELECT name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete FROM ps_buckets_5"},{"sql":"CREATE TABLE ps_oplog(\n  bucket TEXT NOT NULL,\n  op_id INTEGER NOT NULL,\n  op INTEGER NOT NULL,\n  row_type TEXT,\n  row_id TEXT,\n  key TEXT,\n  data TEXT,\n  hash INTEGER NOT NULL,\n  superseded INTEGER NOT NULL)"},{"sql":"CREATE INDEX ps_oplog_by_row ON ps_oplog (row_type, row_id) WHERE superseded = 0"},{"sql":"CREATE INDEX ps_oplog_by_opid ON ps_oplog (bucket, op_id)"},{"sql":"CREATE INDEX ps_oplog_by_key ON ps_oplog (bucket, key) WHERE superseded = 0"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, key, data, hash, superseded)\n    SELECT ps_buckets_5.name, oplog.op_id, 3, oplog.row_type, oplog.row_id, oplog.key, oplog.data, oplog.hash, 0\n    FROM ps_oplog_5 oplog\n    JOIN ps_buckets_5\n        ON ps_buckets_5.id = oplog.bucket"},{"sql":"DROP TABLE ps_oplog_5"},{"sql":"DROP TABLE ps_buckets_5"},{"sql":"INSERT INTO ps_oplog(bucket, op_id, op, row_type, row_id, hash, superseded)\n    SELECT ''$local'', 1, 4, r.row_type, r.row_id, 0, 0\n    FROM ps_updated_rows r"},{"sql":"INSERT OR REPLACE INTO ps_buckets(name, pending_delete, last_op, target_op) VALUES(''$local'', 1, 0, 9223372036854775807)"},{"sql":"DROP TABLE ps_updated_rows"},{"sql":"DELETE FROM ps_migration WHERE id >= 5"}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(6, '[{"sql":"DELETE FROM ps_migration WHERE id >= 6"}]')
+;INSERT INTO ps_migration(id, down_migrations) VALUES(7, '[{"sql":"INSERT OR REPLACE INTO ps_kv(key, value) SELECT ''last_synced_at'', last_synced_at FROM ps_sync_state WHERE priority = -1"},{"sql":"DROP TABLE ps_sync_state"},{"sql":"DELETE FROM ps_migration WHERE id >= 7"}]')
+''',
 };
 
 final finalState = expectedState[databaseVersion]!;
@@ -230,6 +274,17 @@ const data1 = <int, String>{
   (2, 3, 'lists', 'l1', '', '{}', 3)
 ;INSERT INTO ps_updated_rows(row_type, row_id) VALUES
   ('lists', 'l2')
+''',
+  7: r'''
+;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0),
+  (2, 'b2', 0, 0, 0, 1005, 3, 0)
+;INSERT INTO ps_oplog(bucket, op_id, row_type, row_id, key, data, hash) VALUES
+  (1, 1, 'todos', 't1', '', '{}', 100),
+  (1, 2, 'todos', 't2', '', '{}', 20),
+  (2, 3, 'lists', 'l1', '', '{}', 3)
+;INSERT INTO ps_updated_rows(row_type, row_id) VALUES
+  ('lists', 'l2')
 '''
 };
 
@@ -270,7 +325,8 @@ final dataDown1 = <int, String>{
   ('b1', 2, 3, 'todos', 't2', '', '{}', 20, 0),
   ('b2', 3, 3, 'lists', 'l1', '', '{}', 3, 0)
 ''',
-  5: data1[5]!
+  5: data1[5]!,
+  6: data1[5]!
 };
 
 final finalData1 = data1[databaseVersion]!;