Merge pull request #46 from powersync-ja/fix-existing-data

Fix existing data (remove dangling rows)

Author: rkistner

Cargo.lock

@@ -24,7 +24,7 @@ inherits = "release"
 inherits = "wasm"
 
 [workspace.package]
-version = "0.3.5"
+version = "0.3.6"
 edition = "2021"
 authors = ["JourneyApps"]
 keywords = ["sqlite", "powersync"]

Cargo.toml

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

android/build.gradle.kts

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

android/src/prefab/prefab.json

@@ -0,0 +1,47 @@
+use alloc::format;
+
+use crate::error::{PSResult, SQLiteError};
+use sqlite_nostd as sqlite;
+use sqlite_nostd::{Connection, ResultCode};
+
+use crate::ext::SafeManagedStmt;
+use crate::util::quote_identifier;
+
+// Apply a data migration to fix any existing data affected by the issue
+// fixed in v0.3.5.
+//
+// The issue was that the `ps_updated_rows` table was not being populated
+// with remove operations in some cases. This causes the rows to be removed
+// from ps_oplog, but not from the ps_data__tables, resulting in dangling rows.
+//
+// The fix here is to find these dangling rows, and add them to ps_updated_rows.
+// The next time the sync_local operation is run, these rows will be removed.
+pub fn apply_v035_fix(db: *mut sqlite::sqlite3) -> Result<i64, SQLiteError> {
+    // language=SQLite
+    let statement = db
+      .prepare_v2("SELECT name, powersync_external_table_name(name) FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data__*'")
+      .into_db_result(db)?;
+
+    while statement.step()? == ResultCode::ROW {
+        let full_name = statement.column_text(0)?;
+        let short_name = statement.column_text(1)?;
+        let quoted = quote_identifier(full_name);
+
+        // language=SQLite
+        let statement = db.prepare_v2(&format!(
+            "
+INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
+SELECT ?1, id FROM {}
+  WHERE NOT EXISTS (
+      SELECT 1 FROM ps_oplog
+      WHERE row_type = ?1 AND row_id = {}.id
+  );",
+            quoted, quoted
+        ))?;
+        statement.bind_text(1, short_name, sqlite::Destructor::STATIC)?;
+
+        statement.exec()?;
+    }
+
+    Ok(1)
+}

crates/core/src/fix035.rs

@@ -17,6 +17,7 @@ mod crud_vtab;
 mod diff;
 mod error;
 mod ext;
+mod fix035;
 mod kv;
 mod macros;
 mod migrations;

crates/core/src/lib.rs

@@ -9,6 +9,7 @@ use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context};
 
 use crate::error::{PSResult, SQLiteError};
+use crate::fix035::apply_v035_fix;
 
 pub fn powersync_migrate(
     ctx: *mut sqlite::context,
@@ -283,5 +284,24 @@ VALUES(5,
           .into_db_result(local_db)?;
     }
 
+    if current_version < 6 && target_version >= 6 {
+        if current_version != 0 {
+            // Remove dangling rows, but skip if the database is created from scratch.
+            apply_v035_fix(local_db)?;
+        }
+
+        local_db
+            .exec_safe(
+                "\
+INSERT INTO ps_migration(id, down_migrations)
+VALUES(6,
+json_array(
+  json_object('sql', 'DELETE FROM ps_migration WHERE id >= 6')
+));
+",
+            )
+            .into_db_result(local_db)?;
+    }
+
     Ok(())
 }

crates/core/src/migrations.rs

@@ -129,7 +129,6 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
 
             while supersede_statement.step()? == ResultCode::ROW {
                 // Superseded (deleted) a previous operation, add the checksum
-                let superseded_op = supersede_statement.column_int64(0)?;
                 let supersede_checksum = supersede_statement.column_int(1)?;
                 add_checksum = add_checksum.wrapping_add(supersede_checksum);
                 op_checksum = op_checksum.wrapping_sub(supersede_checksum);

crates/core/src/operations.rs

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

crates/core/src/view_admin.rs

@@ -5,6 +5,7 @@ import 'package:test/test.dart';
 
 import 'utils/native_test_utils.dart';
 import 'utils/migration_fixtures.dart' as fixtures;
+import 'utils/fix_035_fixtures.dart' as fix035;
 import 'utils/schema.dart';
 
 void main() {
@@ -175,5 +176,40 @@ void main() {
           '${fixtures.expectedState[3]!.replaceAll(RegExp(r';INSERT INTO ps_migration.*'), '').trim()}\n${fixtures.schemaDown3.trim()}';
       expect(schema, equals(expected));
     });
+
+    test('migrate from 5 with broken data', () async {
+      var tableSchema = {
+        'tables': [
+          {
+            'name': 'lists',
+            'columns': [
+              {'name': 'description', 'type': 'TEXT'}
+            ]
+          },
+          {
+            'name': 'todos',
+            'columns': [
+              {'name': 'description', 'type': 'TEXT'}
+            ]
+          }
+        ]
+      };
+      db.select('select powersync_init()');
+      db.select(
+          'select powersync_replace_schema(?)', [jsonEncode(tableSchema)]);
+
+      db.select('select powersync_test_migration(5)');
+      db.execute(fix035.dataBroken);
+
+      db.select('select powersync_init()');
+      final data = getData(db);
+      expect(data, equals(fix035.dataMigrated.trim()));
+
+      db.select('insert into powersync_operations(op, data) values(?, ?)',
+          ['sync_local', '']);
+
+      final data2 = getData(db);
+      expect(data2, equals(fix035.dataFixed.trim()));
+    });
   });
 }