Fix tx_id mismatch for insert-only tables

crates/core/src/crud_vtab.rs

@@ -50,6 +50,15 @@ struct VirtualTable {
 struct ActiveCrudTransaction {
     tx_id: i64,
     mode: CrudTransactionMode,
+    /// Whether the virtual table owning this transaction observed the begin, or whether we re-used
+    /// a transaction we already knew existed.
+    ///
+    /// We have two virtual tables to record local mutations as crud entries. They both need a begin
+    /// hook to increment `ps_tx` when a transaction operates on that table. For transactions
+    /// operating on both tables, only one table must increment the counter though. We store whether
+    /// we're in a transaction in [DatabaseState::current_transaction_id], if that is true when
+    /// `xBegin` is called then we skip incrementing the counter.
+    observed_begin: bool,
 }
 
 enum CrudTransactionMode {
@@ -184,17 +193,28 @@ impl VirtualTable {
     fn begin(&mut self) -> Result<()> {
         let db = self.db;
 
-        // language=SQLite
-        let statement =
-            db.prepare_v2("UPDATE ps_tx SET next_tx = next_tx + 1 WHERE id = 1 RETURNING next_tx")?;
-        let tx_id = if statement.step()? {
-            statement.column_int64(0) - 1
-        } else {
-            return Err(PowerSyncError::unknown_internal());
+        let (tx_id, observed_begin) = {
+            if let Some(existing_tx) = self.state.current_transaction_id.get() {
+                // Re-use existing transaction, the other table is responsible for clearing that
+                // field on commit.
+                (existing_tx, false)
+            } else {
+                let statement = db.prepare_v2(
+                    "UPDATE ps_tx SET next_tx = next_tx + 1 WHERE id = 1 RETURNING next_tx",
+                )?;
+                let tx_id = if statement.step()? {
+                    statement.column_int64(0) - 1
+                } else {
+                    return Err(PowerSyncError::unknown_internal());
+                };
+                self.state.current_transaction_id.set(Some(tx_id));
+                (tx_id, true)
+            }
         };
 
         self.current_tx = Some(ActiveCrudTransaction {
             tx_id,
+            observed_begin,
             mode: if self.is_simple {
                 CrudTransactionMode::Simple(Default::default())
             } else {
@@ -206,7 +226,11 @@ impl VirtualTable {
     }
 
     fn end_transaction(&mut self) {
-        self.current_tx = None;
+        if let Some(tx) = self.current_tx.take() {
+            if tx.observed_begin {
+                self.state.current_transaction_id.set(None);
+            }
+        }
     }
 }
 

crates/core/src/state.rs

@@ -36,6 +36,7 @@ pub struct DatabaseState {
     /// Cached put and delete statements for raw tables, used by the `sync_local` step of the sync
     /// client.
     pub inferred_schema_cache: InferredSchemaCache,
+    pub current_transaction_id: Cell<Option<i64>>,
 }
 
 impl DatabaseState {

dart/test/crud_test.dart

@@ -980,5 +980,68 @@ INSERT INTO ps_kv(key, value) VALUES
         expect(db.select('SELECT * FROM ps_crud'), isEmpty);
       });
     });
+
+    group('transaction ids', () {
+      setUp(() {
+        db.executeInTx('select powersync_replace_schema(?)', [
+          json.encode({
+            'tables': [
+              {
+                'name': 'regular',
+                'columns': [
+                  {'name': 'a', 'type': 'integer'}
+                ]
+              },
+              {
+                'name': 'insertonly',
+                'insert_only': true,
+                'columns': [
+                  {'name': 'a', 'type': 'integer'}
+                ]
+              }
+            ],
+          })
+        ]);
+      });
+
+      for (final table in ['regular', 'insertonly']) {
+        test('for write into $table table', () {
+          for (var tx = 1; tx < 10; tx++) {
+            db.execute('BEGIN');
+            final numWrites = tx * 2;
+            for (var i = 0; i < numWrites; i++) {
+              db.execute('INSERT INTO $table (id, a) VALUES (uuid(), 1234)');
+            }
+            db.execute('COMMIT');
+
+            expect(
+              db.select('SELECT * FROM ps_crud WHERE tx_id = ?', [tx]),
+              hasLength(numWrites),
+            );
+          }
+        });
+      }
+
+      for (final (first, second) in [
+        ('regular', 'insertonly'),
+        ('insertonly', 'regular')
+      ]) {
+        test('write $first then $second', () {
+          for (var tx = 1; tx < 10; tx++) {
+            db.execute('BEGIN');
+            for (var i = 0; i < tx; i++) {
+              db.execute('INSERT INTO $first (id, a) VALUES (uuid(), 1234)');
+              db.execute('INSERT INTO $second (id, a) VALUES (uuid(), 1234)');
+            }
+            db.execute('COMMIT');
+
+            expect(
+              db.select('SELECT * FROM ps_crud WHERE tx_id = ?', [tx]),
+              hasLength(tx * 2),
+            );
+          }
+        });
+      }
+    });
   });
 }