Fix duplicate entries in ps_sync_state

Cargo.toml

@@ -29,7 +29,7 @@ inherits = "release"
 inherits = "wasm"
 
 [workspace.package]
-version = "0.3.11"
+version = "0.3.12"
 edition = "2021"
 authors = ["JourneyApps"]
 keywords = ["sqlite", "powersync"]

android/build.gradle.kts

@@ -6,7 +6,7 @@ plugins {
 }
 
 group = "co.powersync"
-version = "0.3.11"
+version = "0.3.12"
 description = "PowerSync Core SQLite Extension"
 
 repositories {

android/src/prefab/prefab.json

@@ -2,5 +2,5 @@
   "name": "powersync_sqlite_core",
   "schema_version": 2,
   "dependencies": [],
-  "version": "0.3.11"
+  "version": "0.3.12"
 }

crates/core/src/migrations.rs

@@ -333,5 +333,26 @@ json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7')
         local_db.exec_safe(&stmt).into_db_result(local_db)?;
     }
 
+    if current_version < 8 && target_version >= 8 {
+        let stmt = "\
+ALTER TABLE ps_sync_state RENAME TO ps_sync_state_old;
+CREATE TABLE ps_sync_state (
+  priority INTEGER NOT NULL PRIMARY KEY,
+  last_synced_at TEXT NOT NULL
+) STRICT;
+INSERT INTO ps_sync_state (priority, last_synced_at)
+  SELECT priority, MAX(last_synced_at) FROM ps_sync_state_old GROUP BY priority;
+DROP TABLE ps_sync_state_old;
+INSERT INTO ps_migration(id, down_migrations) VALUES(8, json_array(
+json_object('sql', 'ALTER TABLE ps_sync_state RENAME TO ps_sync_state_new'),
+json_object('sql', 'CREATE TABLE ps_sync_state (\n  priority INTEGER NOT NULL,\n  last_synced_at TEXT NOT NULL\n) STRICT'),
+json_object('sql', 'INSERT INTO ps_sync_state SELECT * FROM ps_sync_state_new'),
+json_object('sql', 'DROP TABLE ps_sync_state_new'),
+json_object('sql', 'DELETE FROM ps_migration WHERE id >= 8')
+));
+";
+        local_db.exec_safe(&stmt).into_db_result(local_db)?;
+    }
+
     Ok(())
 }

crates/core/src/view_admin.rs

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

dart/pubspec.lock

@@ -5,23 +5,18 @@ packages:
     dependency: transitive
     description:
       name: _fe_analyzer_shared
-      sha256: "03f6da266a27a4538a69295ec142cb5717d7d4e5727b84658b63e1e1509bac9c"
+      sha256: dc27559385e905ad30838356c5f5d574014ba39872d732111cd07ac0beff4c57
       url: "https://pub.dev"
     source: hosted
-    version: "79.0.0"
-  _macros:
-    dependency: transitive
-    description: dart
-    source: sdk
-    version: "0.3.3"
+    version: "80.0.0"
   analyzer:
     dependency: transitive
     description:
       name: analyzer
-      sha256: c9040fc56483c22a5e04a9f6a251313118b1a3c42423770623128fa484115643
+      sha256: "192d1c5b944e7e53b24b5586db760db934b177d4147c42fbca8c8c5f1eb8d11e"
       url: "https://pub.dev"
     source: hosted
-    version: "7.2.0"
+    version: "7.3.0"
   args:
     dependency: transitive
     description:
@@ -34,10 +29,10 @@ packages:
     dependency: transitive
     description:
       name: async
-      sha256: d2872f9c19731c2e5f10444b14686eb7cc85c76274bd6c16e1816bff9a3bab63
+      sha256: "758e6d74e971c3e5aceb4110bfd6698efc7f501675bcfe0c775459a8140750eb"
       url: "https://pub.dev"
     source: hosted
-    version: "2.12.0"
+    version: "2.13.0"
   boolean_selector:
     dependency: transitive
     description:
@@ -46,6 +41,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.1.2"
+  clock:
+    dependency: transitive
+    description:
+      name: clock
+      sha256: fddb70d9b5277016c77a80201021d40a2247104d9f4aa7bab7157b7e3f05b84b
+      url: "https://pub.dev"
+    source: hosted
+    version: "1.1.2"
   collection:
     dependency: transitive
     description:
@@ -78,16 +81,24 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "3.0.6"
+  fake_async:
+    dependency: "direct dev"
+    description:
+      name: fake_async
+      sha256: "5368f224a74523e8d2e7399ea1638b37aecfca824a3cc4dfdf77bf1fa905ac44"
+      url: "https://pub.dev"
+    source: hosted
+    version: "1.3.3"
   ffi:
     dependency: transitive
     description:
       name: ffi
-      sha256: "16ed7b077ef01ad6170a3d0c57caa4a112a38d7a2ed5602e0aca9ca6f3d98da6"
+      sha256: "289279317b4b16eb2bb7e271abccd4bf84ec9bdcbe999e278a94b804f5630418"
       url: "https://pub.dev"
     source: hosted
-    version: "2.1.3"
+    version: "2.1.4"
   file:
-    dependency: transitive
+    dependency: "direct dev"
     description:
       name: file
       sha256: a3b4f84adafef897088c160faf7dfffb7696046cb13ae90b508c2cbc95d3b8d4
@@ -138,10 +149,10 @@ packages:
     dependency: transitive
     description:
       name: js
-      sha256: c1b2e9b5ea78c45e1a0788d29606ba27dc5f71f019f32ca5140f61ef071838cf
+      sha256: "53385261521cc4a0c4658fd0ad07a7d14591cf8fc33abbceae306ddb974888dc"
       url: "https://pub.dev"
     source: hosted
-    version: "0.7.1"
+    version: "0.7.2"
   logging:
     dependency: transitive
     description:
@@ -150,14 +161,6 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.3.0"
-  macros:
-    dependency: transitive
-    description:
-      name: macros
-      sha256: "1d9e801cd66f7ea3663c45fc708450db1fa57f988142c64289142c9b7ee80656"
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.1.3-main.0"
   matcher:
     dependency: transitive
     description:
@@ -250,10 +253,10 @@ packages:
     dependency: transitive
     description:
       name: shelf_web_socket
-      sha256: cc36c297b52866d203dbf9332263c94becc2fe0ceaa9681d07b6ef9807023b67
+      sha256: "3632775c8e90d6c9712f883e633716432a27758216dfb61bd86a8321c0580925"
       url: "https://pub.dev"
     source: hosted
-    version: "2.0.1"
+    version: "3.0.0"
   source_map_stack_trace:
     dependency: transitive
     description:
@@ -282,10 +285,18 @@ packages:
     dependency: "direct main"
     description:
       name: sqlite3
-      sha256: "35d3726fe18ab1463403a5cc8d97dbc81f2a0b08082e8173851363fcc97b6627"
+      sha256: "32b632dda27d664f85520093ed6f735ae5c49b5b75345afb8b19411bc59bb53d"
       url: "https://pub.dev"
     source: hosted
-    version: "2.7.2"
+    version: "2.7.4"
+  sqlite3_test:
+    dependency: "direct dev"
+    description:
+      name: sqlite3_test
+      sha256: "0b6f76541385cbe0cebf9454854f78dc9aa18b8cb512d8e597e63385e61d4f45"
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.1.1"
   stack_trace:
     dependency: transitive
     description:
@@ -322,10 +333,10 @@ packages:
     dependency: "direct dev"
     description:
       name: test
-      sha256: "8391fbe68d520daf2314121764d38e37f934c02fd7301ad18307bd93bd6b725d"
+      sha256: "301b213cd241ca982e9ba50266bd3f5bd1ea33f1455554c5abb85d1be0e2d87e"
       url: "https://pub.dev"
     source: hosted
-    version: "1.25.14"
+    version: "1.25.15"
   test_api:
     dependency: transitive
     description:
@@ -370,10 +381,10 @@ packages:
     dependency: transitive
     description:
       name: web
-      sha256: cd3543bd5798f6ad290ea73d210f423502e71900302dde696f8bff84bf89a1cb
+      sha256: "868d88a33d8a87b18ffc05f9f030ba328ffefba92d6c127917a2ba740f9cfe4a"
       url: "https://pub.dev"
     source: hosted
-    version: "1.1.0"
+    version: "1.1.1"
   web_socket:
     dependency: transitive
     description:
@@ -407,4 +418,4 @@ packages:
     source: hosted
     version: "3.1.3"
 sdks:
-  dart: ">=3.5.0 <4.0.0"
+  dart: ">=3.7.0 <4.0.0"

dart/pubspec.yaml

@@ -8,3 +8,6 @@ dependencies:
   sqlite3: ^2.4.5
 dev_dependencies:
   test: ^1.25.0
+  file: ^7.0.1
+  sqlite3_test: ^0.1.1
+  fake_async: ^1.3.3

dart/test/migration_test.dart

@@ -246,5 +246,34 @@ ${fixtures.schema5.trim()}
 end''';
       expect(schema, equals(expected));
     });
+
+    test('schema 7 -> 8 migrates last_synced_at data', () {
+      db.execute(fixtures.expectedState[7]!);
+
+      for (var i = 0; i < 10; i++) {
+        db.execute(
+          'INSERT OR REPLACE INTO ps_sync_state (priority, last_synced_at) VALUES (?, ?);',
+          [2147483647, '2025-03-05 14:58:${i.toString().padLeft(2, '0')}'],
+        );
+
+        db.execute(
+          'INSERT OR REPLACE INTO ps_sync_state (priority, last_synced_at) VALUES (?, ?);',
+          [3, '2025-03-05 13:58:${i.toString().padLeft(2, '0')}'],
+        );
+      }
+
+      db.execute('SELECT powersync_test_migration(8);');
+
+      expect(db.select('SELECT * FROM ps_sync_state ORDER BY priority'), [
+        {
+          'priority': 3,
+          'last_synced_at': '2025-03-05 13:58:09',
+        },
+        {
+          'priority': 2147483647,
+          'last_synced_at': '2025-03-05 14:58:09',
+        }
+      ]);
+    });
   });
 }

dart/test/sync_test.dart

@@ -1,16 +1,28 @@
 import 'dart:convert';
 
+import 'package:fake_async/fake_async.dart';
+import 'package:file/local.dart';
 import 'package:sqlite3/common.dart';
+import 'package:sqlite3/sqlite3.dart';
+import 'package:sqlite3_test/sqlite3_test.dart';
 import 'package:test/test.dart';
 
 import 'utils/native_test_utils.dart';
 
 void main() {
+  final vfs = TestSqliteFileSystem(fs: const LocalFileSystem());
+
+  setUpAll(() {
+    loadExtension();
+    sqlite3.registerVirtualFileSystem(vfs, makeDefault: false);
+  });
+  tearDownAll(() => sqlite3.unregisterVirtualFileSystem(vfs));
+
   group('sync tests', () {
     late CommonDatabase db;
 
     setUp(() async {
-      db = openTestDatabase()
+      db = openTestDatabase(vfs)
         ..select('select powersync_init();')
         ..select('select powersync_replace_schema(?)', [json.encode(_schema)]);
     });
@@ -187,6 +199,38 @@ void main() {
       }
     });
 
+    test('can sync multiple times', () {
+      fakeAsync((controller) {
+        for (var i = 0; i < 10; i++) {
+          for (var prio in const [1, 2, 3, null]) {
+            pushCheckpointComplete('1', null, [], priority: prio);
+
+            // Make sure there's only a single row in last_synced_at
+            expect(
+              db.select(
+                  "SELECT datetime(last_synced_at, 'localtime') AS last_synced_at FROM ps_sync_state WHERE priority = ?",
+                  [prio ?? 2147483647]),
+              [
+                {'last_synced_at': '2025-03-01 ${10 + i}:00:00'}
+              ],
+            );
+
+            if (prio == null) {
+              expect(
+                db.select(
+                    "SELECT datetime(powersync_last_synced_at(), 'localtime') AS last_synced_at"),
+                [
+                  {'last_synced_at': '2025-03-01 ${10 + i}:00:00'}
+                ],
+              );
+            }
+          }
+
+          controller.elapse(const Duration(hours: 1));
+        }
+      }, initialTime: DateTime(2025, 3, 1, 10));
+    });
+
     test('clearing database clears sync status', () {
       pushSyncData('prio1', '1', 'row-0', 'PUT', {'col': 'hi'});