fix: Preserve harness transactions for non-synced row updates (#133)

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Unreleased
 
+- fix: Keep non-synced `updateRow` calls in the test harness transaction.
 - fix: Resolve shared-package models and enums during projection and sync.
 - fix: Preserve boolean types when syncing SQLite column updates.
 - fix: Preserve JSON and JSONB field types and values through inserts, updates,

packages/serverpod_offline_sync/lib/src/database/database.dart

@@ -657,6 +657,13 @@ class OfflineSyncDatabase implements Database {
     Transaction? transaction,
   }) async {
     await _ensureInitialized();
+    if (!_recorder.isCrdtTracked<T>(row.table)) {
+      return _delegate.updateRow<T>(
+        row,
+        columns: columns,
+        transaction: transaction,
+      );
+    }
     return DatabaseUtil.runInTransactionOrSavepoint(
       _delegate,
       transaction,

packages/serverpod_offline_sync_server/pubspec.yaml

@@ -16,5 +16,6 @@ dependencies:
 
 dev_dependencies:
   lints: '>=3.0.0 <7.0.0'
+  serverpod_cli: 4.0.0
   serverpod_test: 4.0.0
   test: ^1.25.5

packages/serverpod_offline_sync_server/test/integration/untracked_update_test.dart

@@ -0,0 +1,109 @@
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:serverpod/serverpod.dart';
+// Serverpod currently exposes its SQL generator only through this library.
+// ignore: implementation_imports
+import 'package:serverpod_cli/src/database/dialects/postgres.dart';
+import 'package:serverpod_database/serverpod_database.dart' show DatabaseDefinition;
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:test/test.dart';
+
+import 'test_tools/serverpod_test_tools.dart';
+
+void main() {
+  final serverDirectory = Directory(
+    '${Directory.systemTemp.path}/offline_sync_updates_${const Uuid().v4()}',
+  );
+  setUpAll(() async {
+    await _preparePostgresMigrations(serverDirectory);
+  });
+
+  tearDownAll(() async {
+    if (serverDirectory.existsSync()) await serverDirectory.delete(recursive: true);
+  });
+
+  withServerpod(
+    'PostgreSQL untracked updates with the database interceptor',
+    (sessionBuilder, _) {
+      late Session session;
+
+      setUp(() {
+        sessionBuilder.build().serverpod.initializeOfflineSync(syncTables: []);
+        session = sessionBuilder.build();
+      });
+
+      group('Given an ordinary row in the test harness transaction,', () {
+        late ServerHealthMetric metric;
+
+        setUp(() async {
+          metric = await ServerHealthMetric.db.insertRow(
+            session,
+            ServerHealthMetric(
+              name: 'original',
+              serverId: 'test server',
+              timestamp: DateTime.utc(2026),
+              isHealthy: true,
+              value: 1,
+              granularity: 1,
+            ),
+          );
+        });
+
+        group('when updateRow changes its name,', () {
+          setUp(() async {
+            await ServerHealthMetric.db.updateRow(
+              session,
+              metric.copyWith(name: 'updated'),
+            );
+          });
+
+          test('then the change is visible in the test transaction.', () async {
+            final stored = await ServerHealthMetric.db.findById(session, metric.id!);
+            expect(stored?.name, 'updated');
+          });
+        });
+      });
+    },
+    databaseInterceptor: offlineSyncDatabaseInterceptor,
+    serverDirectory: serverDirectory,
+    configOverride: (config) => config.copyWith(
+      apiServer: ServerConfig(
+        port: 0,
+        publicHost: 'localhost',
+        publicPort: 0,
+        publicScheme: 'http',
+      ),
+      database: PostgresDatabaseConfig.embedded(
+        dataPath: '${Directory.systemTemp.path}/offline_sync_updates_postgres_$pid',
+        name: 'serverpod_test',
+        maxConnectionCount: 5,
+      ),
+    ),
+  );
+}
+
+// The module's committed SQL targets SQLite. Render the current generated
+// definition with the same PostgreSQL generator used by serverpod create-migration.
+Future<void> _preparePostgresMigrations(Directory serverDirectory) async {
+  final versions = await File('migrations/migration_registry.txt').readAsLines();
+  final version = versions.lastWhere((line) => line.trim().isNotEmpty);
+  final source = Directory('migrations/$version');
+  final target = Directory('${serverDirectory.path}/migrations/$version');
+  await target.create(recursive: true);
+  await for (final file in source.list()) {
+    if (file is File) {
+      await file.copy('${target.path}/${file.uri.pathSegments.last}');
+    }
+  }
+  final definition = DatabaseDefinition.fromJson(
+    jsonDecode(await File('${target.path}/definition.json').readAsString())
+        as Map<String, dynamic>,
+  );
+  final sql = definition.toPgSql(installedModules: definition.installedModules);
+  await File('${target.path}/definition.sql').writeAsString(sql);
+  await File('${target.path}/migration.sql').writeAsString(sql);
+  await File(
+    '${serverDirectory.path}/migrations/migration_registry.txt',
+  ).writeAsString('$version\n');
+}

test/serverpod_offline_sync_test_server/test/integration/crud/delete_test.dart

@@ -785,4 +785,142 @@ void main() {
       });
     },
   );
+
+  group('Given two rows on a table not tracked by CRDT,', () {
+    late OfflineSyncIntegrityViolation violation;
+    late OfflineSyncIntegrityViolation otherViolation;
+
+    setUp(() async {
+      violation = await OfflineSyncIntegrityViolation.db.insertRow(
+        session,
+        OfflineSyncIntegrityViolation(
+          type: OfflineSyncViolationType.ownershipCollision,
+          domainTableName: Person.t.tableName,
+          uuidRowId: const Uuid().v7obj(),
+          incomingSpaceUuid: testCrdtUserId,
+          operation: OfflineSyncViolationOperation.mergeInsert,
+          firstSeenAt: DateTime.utc(2026),
+          lastSeenAt: DateTime.utc(2026),
+          occurrences: 1,
+        ),
+      );
+      otherViolation = await OfflineSyncIntegrityViolation.db.insertRow(
+        session,
+        violation.copyWith(id: null, uuidRowId: const Uuid().v7obj()),
+      );
+    });
+
+    group('when deleting one row with deleteRow,', () {
+      late OfflineSyncIntegrityViolation deleted;
+
+      setUp(() async {
+        deleted = await OfflineSyncIntegrityViolation.db.deleteRow(session, violation);
+      });
+
+      test('then the deleted row is returned.', () {
+        expect(deleted.id, violation.id);
+        expect(deleted.uuidRowId, violation.uuidRowId);
+      });
+
+      test('then only that row is physically removed.', () async {
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(testSession, violation.id!),
+          isNull,
+        );
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(
+            testSession,
+            otherViolation.id!,
+          ),
+          isNotNull,
+        );
+      });
+    });
+
+    group('when deleting one row with delete and noReturn,', () {
+      late List<OfflineSyncIntegrityViolation> deleted;
+
+      setUp(() async {
+        deleted = await OfflineSyncIntegrityViolation.db.delete(
+          session,
+          [violation],
+          noReturn: true,
+        );
+      });
+
+      test('then an empty list is returned.', () {
+        expect(deleted, isEmpty);
+      });
+
+      test('then only that row is physically removed.', () async {
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(testSession, violation.id!),
+          isNull,
+        );
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(
+            testSession,
+            otherViolation.id!,
+          ),
+          isNotNull,
+        );
+      });
+    });
+
+    group('when deleting one row with deleteWhere and noReturn,', () {
+      late List<OfflineSyncIntegrityViolation> deleted;
+
+      setUp(() async {
+        deleted = await OfflineSyncIntegrityViolation.db.deleteWhere(
+          session,
+          where: (t) => t.id.equals(violation.id),
+          noReturn: true,
+        );
+      });
+
+      test('then an empty list is returned.', () {
+        expect(deleted, isEmpty);
+      });
+
+      test('then only the matching row is physically removed.', () async {
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(testSession, violation.id!),
+          isNull,
+        );
+        expect(
+          await OfflineSyncIntegrityViolation.db.findById(
+            testSession,
+            otherViolation.id!,
+          ),
+          isNotNull,
+        );
+      });
+    });
+
+    group('when deleteRow runs in an explicit transaction that rolls back,', () {
+      setUp(() async {
+        final rollback = StateError('rollback');
+        await expectLater(
+          session.db.transaction((tx) async {
+            await OfflineSyncIntegrityViolation.db.deleteRow(
+              session,
+              violation,
+              transaction: tx,
+            );
+            throw rollback;
+          }),
+          throwsA(same(rollback)),
+        );
+      });
+
+      test('then the deleted row is restored.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          testSession,
+          violation.id!,
+        );
+        expect(stored?.uuidRowId, violation.uuidRowId);
+        expect(stored?.occurrences, 1);
+      });
+    });
+  });
 }

test/serverpod_offline_sync_test_server/test/integration/crud/insert_test.dart

@@ -545,4 +545,112 @@ void main() {
       });
     });
   });
+
+  group('Given an empty table not tracked by CRDT,', () {
+    late OfflineSyncIntegrityViolation violation;
+
+    setUp(() {
+      violation = OfflineSyncIntegrityViolation(
+        type: OfflineSyncViolationType.ownershipCollision,
+        domainTableName: Person.t.tableName,
+        uuidRowId: const Uuid().v7obj(),
+        incomingSpaceUuid: testCrdtUserId,
+        operation: OfflineSyncViolationOperation.mergeInsert,
+        firstSeenAt: DateTime.utc(2026),
+        lastSeenAt: DateTime.utc(2026),
+        occurrences: 1,
+      );
+    });
+
+    group('when inserting a row with insertRow,', () {
+      late OfflineSyncIntegrityViolation inserted;
+
+      setUp(() async {
+        inserted = await OfflineSyncIntegrityViolation.db.insertRow(session, violation);
+      });
+
+      test('then the returned row has a generated id and the supplied values.', () {
+        expect(inserted.id, isNotNull);
+        expect(inserted.uuidRowId, violation.uuidRowId);
+        expect(inserted.occurrences, 1);
+      });
+
+      test('then the row is stored in the database.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          inserted.id!,
+        );
+        expect(stored?.uuidRowId, violation.uuidRowId);
+        expect(stored?.occurrences, 1);
+      });
+    });
+
+    group('when inserting rows with noReturn,', () {
+      late List<OfflineSyncIntegrityViolation> inserted;
+
+      setUp(() async {
+        inserted = await OfflineSyncIntegrityViolation.db.insert(
+          session,
+          [violation],
+          noReturn: true,
+        );
+      });
+
+      test('then an empty list is returned.', () {
+        expect(inserted, isEmpty);
+      });
+
+      test('then the row is stored in the database.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findFirstRow(
+          session,
+          where: (t) => t.uuidRowId.equals(violation.uuidRowId),
+        );
+        expect(stored?.occurrences, 1);
+      });
+    });
+  });
+
+  group('Given an existing row on a table not tracked by CRDT,', () {
+    late OfflineSyncIntegrityViolation violation;
+
+    setUp(() async {
+      violation = await OfflineSyncIntegrityViolation.db.insertRow(
+        session,
+        OfflineSyncIntegrityViolation(
+          type: OfflineSyncViolationType.ownershipCollision,
+          domainTableName: Person.t.tableName,
+          uuidRowId: const Uuid().v7obj(),
+          incomingSpaceUuid: testCrdtUserId,
+          operation: OfflineSyncViolationOperation.mergeInsert,
+          firstSeenAt: DateTime.utc(2026),
+          lastSeenAt: DateTime.utc(2026),
+          occurrences: 1,
+        ),
+      );
+    });
+
+    group('when inserting the same primary key with ignoreConflicts,', () {
+      late List<OfflineSyncIntegrityViolation> inserted;
+
+      setUp(() async {
+        inserted = await OfflineSyncIntegrityViolation.db.insert(
+          session,
+          [violation.copyWith(occurrences: 2)],
+          ignoreConflicts: true,
+        );
+      });
+
+      test('then an empty list is returned.', () {
+        expect(inserted, isEmpty);
+      });
+
+      test('then the original row is preserved.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 1);
+      });
+    });
+  });
 }

test/serverpod_offline_sync_test_server/test/integration/crud/update_test.dart

@@ -728,5 +728,132 @@ void main() {
         expect(row?.occurrences, 3);
       });
     });
+
+    group('when updating the row with updateRow,', () {
+      late OfflineSyncIntegrityViolation updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.updateRow(
+          session,
+          violation.copyWith(occurrences: 2),
+        );
+      });
+
+      test('then the updated row is returned.', () {
+        expect(updated.id, violation.id);
+        expect(updated.occurrences, 2);
+      });
+
+      test('then the row reflects the new values.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+      });
+    });
+
+    group('when updateRow selects only the occurrences column,', () {
+      late OfflineSyncIntegrityViolation updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.updateRow(
+          session,
+          violation.copyWith(occurrences: 2, lastSeenAt: DateTime.utc(2027)),
+          columns: (t) => [t.occurrences],
+        );
+      });
+
+      test('then the returned row preserves the unselected timestamp.', () {
+        expect(updated.occurrences, 2);
+        expect(updated.lastSeenAt, violation.lastSeenAt);
+      });
+
+      test('then only the selected column changes in the database.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+        expect(stored?.lastSeenAt, violation.lastSeenAt);
+      });
+    });
+
+    group('when update selects only the occurrences column,', () {
+      late List<OfflineSyncIntegrityViolation> updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.update(
+          session,
+          [violation.copyWith(occurrences: 2, lastSeenAt: DateTime.utc(2027))],
+          columns: (t) => [t.occurrences],
+        );
+      });
+
+      test('then the returned row preserves the unselected timestamp.', () {
+        expect(updated.single.id, violation.id);
+        expect(updated.single.occurrences, 2);
+        expect(updated.single.lastSeenAt, violation.lastSeenAt);
+      });
+
+      test('then only the selected column changes in the database.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+        expect(stored?.lastSeenAt, violation.lastSeenAt);
+      });
+    });
+
+    group('when updateRow runs in an explicit transaction that rolls back,', () {
+      setUp(() async {
+        final rollback = StateError('rollback');
+        await expectLater(
+          session.db.transaction((tx) async {
+            await OfflineSyncIntegrityViolation.db.updateRow(
+              session,
+              violation.copyWith(occurrences: 2),
+              transaction: tx,
+            );
+            throw rollback;
+          }),
+          throwsA(same(rollback)),
+        );
+      });
+
+      test('then the original row is preserved.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 1);
+      });
+    });
+
+    group('when updateById changes the occurrences column,', () {
+      late OfflineSyncIntegrityViolation? updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.updateById(
+          session,
+          violation.id!,
+          columnValues: (t) => [t.occurrences(2)],
+        );
+      });
+
+      test('then the updated row is returned.', () {
+        expect(updated?.id, violation.id);
+        expect(updated?.occurrences, 2);
+      });
+
+      test('then the row reflects the new values.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+      });
+    });
   });
 }

test/serverpod_offline_sync_test_server/test/integration/crud/upsert_test.dart

@@ -638,4 +638,139 @@ void main() {
       });
     });
   });
+
+  group('Given an existing row on a table not tracked by CRDT,', () {
+    late OfflineSyncIntegrityViolation violation;
+
+    setUp(() async {
+      violation = await OfflineSyncIntegrityViolation.db.insertRow(
+        session,
+        OfflineSyncIntegrityViolation(
+          type: OfflineSyncViolationType.ownershipCollision,
+          domainTableName: Person.t.tableName,
+          uuidRowId: const Uuid().v7obj(),
+          incomingSpaceUuid: testCrdtUserId,
+          operation: OfflineSyncViolationOperation.mergeInsert,
+          firstSeenAt: DateTime.utc(2026),
+          lastSeenAt: DateTime.utc(2026),
+          occurrences: 1,
+        ),
+      );
+    });
+
+    group('when upserting a new row without an id,', () {
+      late UuidValue rowUuid;
+      late OfflineSyncIntegrityViolation? inserted;
+
+      setUp(() async {
+        rowUuid = const Uuid().v7obj();
+        inserted = await OfflineSyncIntegrityViolation.db.upsertRow(
+          session,
+          violation.copyWith(id: null, uuidRowId: rowUuid, occurrences: 2),
+          conflictColumns: (t) => [t.id],
+        );
+      });
+
+      test('then the returned row has a generated id and the supplied values.', () {
+        expect(inserted?.id, isNotNull);
+        expect(inserted?.id, isNot(violation.id));
+        expect(inserted?.uuidRowId, rowUuid);
+        expect(inserted?.occurrences, 2);
+      });
+
+      test('then both rows are stored with their respective values.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findFirstRow(
+          session,
+          where: (t) => t.uuidRowId.equals(rowUuid),
+        );
+        final original = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+        expect(original?.occurrences, 1);
+      });
+    });
+
+    group(
+      'when upserting a conflicting row with only occurrences selected for update,',
+      () {
+        late OfflineSyncIntegrityViolation? updated;
+
+        setUp(() async {
+          updated = await OfflineSyncIntegrityViolation.db.upsertRow(
+            session,
+            violation.copyWith(occurrences: 2, lastSeenAt: DateTime.utc(2027)),
+            conflictColumns: (t) => [t.id],
+            updateColumns: (t) => [t.occurrences],
+          );
+        });
+
+        test('then the returned row preserves its id and unselected timestamp.', () {
+          expect(updated?.id, violation.id);
+          expect(updated?.occurrences, 2);
+          expect(updated?.lastSeenAt, violation.lastSeenAt);
+        });
+
+        test('then only the selected column changes in the database.', () async {
+          final stored = await OfflineSyncIntegrityViolation.db.findById(
+            session,
+            violation.id!,
+          );
+          expect(stored?.occurrences, 2);
+          expect(stored?.lastSeenAt, violation.lastSeenAt);
+        });
+      },
+    );
+
+    group('when upserting a conflicting row with a non-matching updateWhere,', () {
+      late OfflineSyncIntegrityViolation? updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.upsertRow(
+          session,
+          violation.copyWith(occurrences: 2),
+          conflictColumns: (t) => [t.id],
+          updateWhere: (t) => t.occurrences.equals(99),
+        );
+      });
+
+      test('then no row is returned.', () {
+        expect(updated, isNull);
+      });
+
+      test('then the original row is preserved.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 1);
+      });
+    });
+
+    group('when upserting a conflicting row with noReturn,', () {
+      late List<OfflineSyncIntegrityViolation> updated;
+
+      setUp(() async {
+        updated = await OfflineSyncIntegrityViolation.db.upsert(
+          session,
+          [violation.copyWith(occurrences: 2)],
+          conflictColumns: (t) => [t.id],
+          noReturn: true,
+        );
+      });
+
+      test('then an empty list is returned.', () {
+        expect(updated, isEmpty);
+      });
+
+      test('then the row reflects the new values.', () async {
+        final stored = await OfflineSyncIntegrityViolation.db.findById(
+          session,
+          violation.id!,
+        );
+        expect(stored?.occurrences, 2);
+      });
+    });
+  });
 }