Fix test

Author: simolus3

Cargo.toml

@@ -9,7 +9,6 @@ default-members = ["crates/shell", "crates/sqlite"]
 
 [profile.dev]
 panic = "abort"
-strip = true
 
 [profile.release]
 panic = "abort"

crates/core/src/migrations.rs

@@ -314,9 +314,9 @@ json_array(
         local_db
             .exec_safe(
                 "\
-ALTER TABLE ps_buckets ADD COLUMN priority NOT NULL DEFAULT 1;
+ALTER TABLE ps_buckets ADD COLUMN priority INTEGER NOT NULL DEFAULT 1;
 INSERT INTO ps_migration(id, down_migrations)
-VALUES(6,
+VALUES(7,
 json_array(
   json_object('sql', 'ALTER TABLE ps_buckets DROP COLUMN priority'),
   json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7')

crates/core/src/operations.rs

@@ -79,7 +79,7 @@ FROM json_each(?) e",
                         RETURNING id, last_applied_op",
     )?;
     bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
-    bucket_statement.bind_int(2, priority.into());
+    bucket_statement.bind_int(2, priority.into())?;
     bucket_statement.step()?;
 
     let bucket_id = bucket_statement.column_int64(0)?;

crates/core/src/sync_local.rs

@@ -59,10 +59,6 @@ pub fn sync_local<V: Value>(db: *mut sqlite::sqlite3, data: &V) -> Result<i64, S
         return Ok(0);
     }
 
-    if priority >= BucketPriority::LOWEST {
-        todo!("Only consider changes from certain bucket priorities")
-    }
-
     // language=SQLite
     let statement = db
         .prepare_v2("SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'")
@@ -84,8 +80,8 @@ pub fn sync_local<V: Value>(db: *mut sqlite::sqlite3, data: &V) -> Result<i64, S
 --    SELECT DISTINCT / UNION is important for cases with many duplicate ids.
 WITH updated_rows AS (
   SELECT DISTINCT b.row_type, b.row_id FROM ps_buckets AS buckets
-    CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id
-  AND (b.op_id > buckets.last_applied_op)
+    CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id AND (b.op_id > buckets.last_applied_op)
+    WHERE buckets.priority <= ?
   UNION SELECT row_type, row_id FROM ps_updated_rows
 )
 
@@ -108,6 +104,7 @@ GROUP BY b.row_type, b.row_id",
 
     // TODO: cache statements
 
+    statement.bind_int(1, priority.into())?;
     while statement.step().into_db_result(db)? == ResultCode::ROW {
         let type_name = statement.column_text(0)?;
         let id = statement.column_text(1)?;

crates/sqlite/build.rs

@@ -1,4 +1,3 @@
-
 fn main() {
     let mut cfg = cc::Build::new();
 

dart/test/utils/fix_035_fixtures.dart

@@ -18,9 +18,9 @@ const dataBroken = '''
 
 /// Data after applying the migration fix, but before sync_local
 const dataMigrated = '''
-;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, 0, 3, 0)
+;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, priority) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0, 1),
+  (2, 'b2', 0, 0, 0, 0, 3, 0, 1)
 ;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),
@@ -39,9 +39,9 @@ const dataMigrated = '''
 
 /// Data after applying the migration fix and sync_local
 const dataFixed = '''
-;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, 0, 3, 0)
+;INSERT INTO ps_buckets(id, name, last_applied_op, last_op, target_op, add_checksum, op_checksum, pending_delete, priority) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0, 1),
+  (2, 'b2', 0, 0, 0, 0, 3, 0, 1)
 ;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),

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,6 +172,46 @@ 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
+, priority INTEGER NOT NULL DEFAULT 1) 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_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":"ALTER TABLE ps_buckets DROP COLUMN priority"},{"sql":"DELETE FROM ps_migration WHERE id >= 7"}]')
 '''
 };
 
@@ -230,6 +270,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, priority) VALUES
+  (1, 'b1', 0, 0, 0, 0, 120, 0, 1),
+  (2, 'b2', 0, 0, 0, 1005, 3, 0, 1)
+;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 +321,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]!;

dart/test/utils/native_test_utils.dart

@@ -3,6 +3,7 @@ import 'dart:ffi';
 import 'package:sqlite3/common.dart';
 import 'package:sqlite3/open.dart' as sqlite_open;
 import 'package:sqlite3/sqlite3.dart';
+import 'package:path/path.dart' as p;
 
 const defaultSqlitePath = 'libsqlite3.so.0';
 
@@ -13,7 +14,8 @@ CommonDatabase openTestDatabase() {
     return DynamicLibrary.open('libsqlite3.so.0');
   });
   sqlite_open.open.overrideFor(sqlite_open.OperatingSystem.macOS, () {
-    return DynamicLibrary.open('libsqlite3.dylib');
+    return DynamicLibrary.open(
+        '/opt/homebrew/Cellar/sqlite/3.48.0/lib/libsqlite3.dylib');
   });
   var lib = DynamicLibrary.open(getLibraryForPlatform(path: libPath));
   var extension = SqliteExtension.inLibrary(lib, 'sqlite3_powersync_init');
@@ -22,29 +24,22 @@ CommonDatabase openTestDatabase() {
 }
 
 String getLibraryForPlatform({String? path = "."}) {
-  switch (Abi.current()) {
-    case Abi.androidArm:
-    case Abi.androidArm64:
-    case Abi.androidX64:
-      return '$path/libpowersync.so';
-    case Abi.macosArm64:
-    case Abi.macosX64:
-      return '$path/libpowersync.dylib';
-    case Abi.linuxX64:
-    case Abi.linuxArm64:
-      return '$path/libpowersync.so';
-    case Abi.windowsX64:
-      return '$path/powersync.dll';
-    case Abi.androidIA32:
-      throw ArgumentError(
+  return p.normalize(p.absolute(switch (Abi.current()) {
+    Abi.androidArm ||
+    Abi.androidArm64 ||
+    Abi.androidX64 =>
+      '$path/libpowersync.so',
+    Abi.macosArm64 || Abi.macosX64 => '$path/libpowersync.dylib',
+    Abi.linuxX64 || Abi.linuxArm64 => '$path/libpowersync.so',
+    Abi.windowsX64 => '$path/powersync.dll',
+    Abi.androidIA32 => throw ArgumentError(
         'Unsupported processor architecture. X86 Android emulators are not '
         'supported. Please use an x86_64 emulator instead. All physical '
         'Android devices are supported including 32bit ARM.',
-      );
-    default:
-      throw ArgumentError(
+      ),
+    _ => throw ArgumentError(
         'Unsupported processor architecture "${Abi.current()}". '
         'Please open an issue on GitHub to request it.',
-      );
-  }
+      )
+  }));
 }