fix: Prevent web deadlocks in transactional membership reads (#161)

.github/workflows/ci.yaml

@@ -98,6 +98,21 @@ jobs:
       #     file-path: coverage-results.txt
       #     comment-tag: coverage-results
 
+  browser_tests:
+    name: dart test (Chrome)
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup
+
+      - name: Prepare SQLite web assets
+        run: bash scripts/setup_sqlite_web_assets.sh
+
+      - name: Run browser regressions
+        working-directory: test/serverpod_offline_sync_test_server
+        run: dart test test/browser --platform chrome --concurrency=1
+
   simulation:
     name: deterministic simulation (${{ matrix.topology }}, ${{ matrix.delivery }})
     runs-on: ubuntu-latest

.gitignore

@@ -22,3 +22,7 @@ coverage-results.txt
 
 # Deterministic simulation hunt results
 .dst-soak/
+
+# Downloaded by scripts/setup_sqlite_web_assets.sh.
+test/serverpod_offline_sync_test_server/test/browser/sqlite3.wasm
+test/serverpod_offline_sync_test_server/test/browser/db_worker.js

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## Unreleased
 
+- fix: Keep membership reads inside the caller's transaction, preventing web
+  deadlocks and reflecting uncommitted membership grants and revocations.
 - fix: Include the persisted local logical clock in resume checkpoints so
   reconnecting within the same millisecond or after clock rollback settles.
 - fix: Keep relay checkpoints tagged with their actual author and recover

analysis_options.yaml

@@ -107,7 +107,6 @@ linter:
     - use_colored_box
     - use_decorated_box
     - use_enums
-    - use_if_null_to_convert_nulls_to_bools
     - use_is_even_rather_than_modulo
     - use_named_constants
     - use_raw_strings

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

@@ -1059,15 +1059,18 @@ class OfflineSyncDatabase implements Database {
       OfflineSyncSpace.db.find(
         _delegate.session,
         where: (t) => t.uuidSpaceId.equals(userId),
+        transaction: transaction,
       ),
       OfflineSyncSpaceMember.db
           .find(
             _delegate.session,
             where: (t) => t.userUuid.equals(userId),
             include: OfflineSyncSpaceMember.include(space: OfflineSyncSpace.include()),
+            transaction: transaction,
           )
           .then((memberships) => [for (final member in memberships) member.space!]),
     ]);
+
     return {
       for (final spaces in spaceGroups)
         for (final space in spaces) space.id!,

scripts/setup_sqlite_web_assets.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Run from the repository root after dart pub get. The browser test runner
+# serves assets relative to the test file, matching sqlite_async's defaults.
+test_directory=test/serverpod_offline_sync_test_server/test/browser
+
+resolved_version() {
+  awk -v dependency="  $1:" '
+    $0 == dependency { found = 1 }
+    found && /^    version:/ {
+      gsub(/"/, "", $2)
+      print $2
+      exit
+    }
+  ' pubspec.lock
+}
+
+sqlite_version="$(resolved_version sqlite3)"
+worker_version="$(resolved_version sqlite_async)"
+
+if [[ -z "$sqlite_version" || -z "$worker_version" ]]; then
+  echo 'Run dart pub get before provisioning SQLite browser assets.' >&2
+  exit 1
+fi
+
+curl --fail --location --silent --show-error --retry 3 \
+  "https://github.com/simolus3/sqlite3.dart/releases/download/sqlite3-$sqlite_version/sqlite3.wasm" \
+  --output "$test_directory/sqlite3.wasm"
+
+curl --fail --location --silent --show-error --retry 3 \
+  "https://github.com/powersync-ja/sqlite_async.dart/releases/download/sqlite_async-v$worker_version/db_worker.js" \
+  --output "$test_directory/db_worker.js"

test/serverpod_offline_sync_test_server/pubspec.yaml

@@ -21,6 +21,7 @@ dev_dependencies:
   clock: ^1.1.2
   lints: '>=3.0.0 <7.0.0'
   path: ^1.9.1
+  serverpod_cli: '>=4.0.4 <4.1.0'
   serverpod_database: '>=4.0.4 <4.1.0'
   serverpod_offline_sync_client:
     path: ../../packages/serverpod_offline_sync_client

test/serverpod_offline_sync_test_server/test/browser/transaction_membership_test.dart

@@ -0,0 +1,88 @@
+@TestOn('browser')
+library;
+
+import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+const _readTimeout = Duration(seconds: 5);
+
+void main() {
+  group('Given a persistent user with a personal space on web,', () {
+    late UuidValue userId;
+    late OfflineSyncDatabaseSession session;
+
+    setUpAll(() async {
+      userId = const Uuid().v7obj();
+      session = await Client('http://localhost:8081/').createSyncSession(
+        'transaction-membership-$userId.db',
+        persistentUserId: userId,
+      );
+      await session.db.currentNodeId();
+    });
+
+    tearDownAll(() async {
+      await session.close();
+    });
+
+    group('when inserting and reading within transactionForUser,', () {
+      late Person inserted;
+      late List<Person> found;
+      late Person? foundById;
+      late Person? first;
+      late int count;
+      late List<Person> committed;
+
+      setUpAll(() async {
+        await session.db.transactionForUser(userId, (tx) async {
+          inserted = await Person.db.insertRow(
+            session,
+            Person(name: 'personal'),
+            transaction: tx,
+          );
+
+          // Bound the read inside the callback so a regression throws and
+          // rolls back the transaction instead of keeping its lock forever.
+          found = await Person.db.find(session, transaction: tx).timeout(_readTimeout);
+          foundById = await Person.db
+              .findById(
+                session,
+                inserted.id!,
+                transaction: tx,
+              )
+              .timeout(_readTimeout);
+          first = await Person.db
+              .findFirstRow(
+                session,
+                where: (t) => t.name.equals('personal'),
+                transaction: tx,
+              )
+              .timeout(_readTimeout);
+          count = await Person.db.count(session, transaction: tx).timeout(_readTimeout);
+        });
+
+        committed = await Person.db.find(session);
+      });
+
+      test('then find returns the uncommitted row without deadlocking.', () {
+        expect(found.map((row) => row.id), [inserted.id]);
+      });
+
+      test('then findById returns the uncommitted row without deadlocking.', () {
+        expect(foundById?.id, inserted.id);
+      });
+
+      test('then findFirstRow returns the uncommitted row without deadlocking.', () {
+        expect(first?.id, inserted.id);
+      });
+
+      test('then count includes the uncommitted row without deadlocking.', () {
+        expect(count, 1);
+      });
+
+      test('then the transaction commits the inserted row.', () {
+        expect(committed.map((row) => row.id), [inserted.id]);
+      });
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/crud/transaction_membership_test.dart

@@ -0,0 +1,155 @@
+import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+
+void main() {
+  initTestClientSession(withPersistentUser: true);
+
+  group('Given a persistent user without membership in a populated shared space,', () {
+    late Person sharedPerson;
+    late OfflineSyncSpaceMember membership;
+
+    setUp(() async {
+      final ownerId = const Uuid().v7obj();
+      sharedPerson = await session.db.transactionForUser(
+        ownerId,
+        (tx) => Person.db.insertRow(
+          session,
+          Person(name: 'shared'),
+          transaction: tx,
+        ),
+      );
+      final sharedSpace = await OfflineSyncSpace.db.findFirstRow(
+        testSession,
+        where: (t) => t.uuidSpaceId.equals(ownerId),
+      );
+      membership = OfflineSyncSpaceMember(
+        spaceId: sharedSpace!.id!,
+        userUuid: testCrdtUserId,
+        role: OfflineSyncSpaceRole.readWrite,
+      );
+    });
+
+    test(
+      'when granting, revoking, and restoring membership within one transaction, '
+      'then each read reflects the preceding uncommitted change.',
+      () async {
+        late List<Person> beforeGrant;
+        late Person? afterGrant;
+        late Person? afterRevocation;
+        late int afterRestoration;
+
+        await session.db.transactionForUser(testCrdtUserId, (tx) async {
+          beforeGrant = await Person.db.find(session, transaction: tx);
+          final insertedMembership = await OfflineSyncSpaceMember.db.insertRow(
+            session,
+            membership,
+            transaction: tx,
+          );
+          afterGrant = await Person.db.findById(
+            session,
+            sharedPerson.id!,
+            transaction: tx,
+          );
+
+          await OfflineSyncSpaceMember.db.deleteRow(
+            session,
+            insertedMembership,
+            transaction: tx,
+          );
+          afterRevocation = await Person.db.findFirstRow(session, transaction: tx);
+
+          await OfflineSyncSpaceMember.db.insertRow(
+            session,
+            membership,
+            transaction: tx,
+          );
+          afterRestoration = await Person.db.count(session, transaction: tx);
+        });
+
+        expect(beforeGrant, isEmpty);
+        expect(afterGrant?.id, sharedPerson.id);
+        expect(afterRevocation, isNull);
+        expect(afterRestoration, 1);
+      },
+    );
+
+    group('when a membership grant is read and then rolled back,', () {
+      late List<Person> insideTransaction;
+      late List<Person> afterRollback;
+      late List<OfflineSyncSpaceMember> persistedMemberships;
+      Object? failure;
+
+      setUp(() async {
+        failure = null;
+
+        try {
+          await session.db.transactionForUser<void>(testCrdtUserId, (tx) async {
+            await OfflineSyncSpaceMember.db.insertRow(
+              session,
+              membership,
+              transaction: tx,
+            );
+            insideTransaction = await Person.db.find(session, transaction: tx);
+
+            throw StateError('Roll back the membership grant');
+          });
+        } on Object catch (error) {
+          failure = error;
+        }
+
+        afterRollback = await Person.db.find(session);
+        persistedMemberships = await OfflineSyncSpaceMember.db.find(testSession);
+      });
+
+      test('then the row was visible inside the transaction.', () {
+        expect(insideTransaction.map((row) => row.id), [sharedPerson.id]);
+      });
+
+      test('then the rolled-back grant leaves no membership or visible row.', () {
+        expect(
+          failure,
+          isA<StateError>().having(
+            (error) => error.message,
+            'message',
+            'Roll back the membership grant',
+          ),
+        );
+        expect(persistedMemberships, isEmpty);
+        expect(afterRollback, isEmpty);
+      });
+    });
+
+    test(
+      'when membership is committed and revoked between transactions, '
+      'then later transactions pick up each change.',
+      () async {
+        final beforeGrant = await session.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.find(session, transaction: tx),
+        );
+
+        final insertedMembership = await OfflineSyncSpaceMember.db.insertRow(
+          testSession,
+          membership,
+        );
+        final afterGrant = await session.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.find(session, transaction: tx),
+        );
+
+        await OfflineSyncSpaceMember.db.deleteRow(testSession, insertedMembership);
+        final afterRevocation = await session.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.find(session, transaction: tx),
+        );
+
+        expect(beforeGrant, isEmpty);
+        expect(afterGrant.map((row) => row.id), [sharedPerson.id]);
+        expect(afterRevocation, isEmpty);
+      },
+    );
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/server/transaction_membership_test.dart

@@ -0,0 +1,153 @@
+import 'dart:io';
+
+import 'package:serverpod/serverpod.dart';
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_server/src/generated/protocol.dart'
+    as server;
+import 'package:test/test.dart';
+
+import '../test_tools/postgres_migrations.dart';
+import '../test_tools/serverpod_test_tools.dart';
+
+void main() {
+  final serverDirectory = Directory(
+    '${Directory.systemTemp.path}/offline_sync_membership_${const Uuid().v4()}',
+  );
+
+  setUpAll(() => preparePostgresMigrations(serverDirectory));
+
+  tearDownAll(() async {
+    if (serverDirectory.existsSync()) await serverDirectory.delete(recursive: true);
+  });
+
+  withServerpod(
+    'PostgreSQL membership visibility',
+    rollbackDatabase: RollbackDatabase.disabled,
+    serverDirectory: serverDirectory,
+    configOverride: (config) => config.copyWith(
+      database: PostgresDatabaseConfig.embedded(
+        dataPath: '${serverDirectory.path}/postgres',
+        name: 'serverpod_test',
+        maxConnectionCount: 6,
+      ),
+    ),
+    (sessionBuilder, _) {
+      late Session otherSession;
+      late OfflineSyncDatabaseSession session;
+
+      setUp(() async {
+        otherSession = sessionBuilder.build();
+        session = OfflineSyncDatabaseSession.wraps(
+          sessionBuilder.build(),
+          syncTables: server.syncTables,
+        );
+        await session.db.initialize();
+      });
+
+      group('Given a PostgreSQL user with membership in a populated shared space,', () {
+        late UuidValue userId;
+        late server.Person person;
+        late OfflineSyncSpaceMember membership;
+
+        setUp(() async {
+          userId = const Uuid().v7obj();
+          final ownerId = const Uuid().v7obj();
+          person = await session.db.transactionForUser(
+            ownerId,
+            (tx) => server.Person.db.insertRow(
+              session,
+              server.Person(name: 'shared'),
+              transaction: tx,
+            ),
+          );
+          final sharedSpace = await OfflineSyncSpace.db.findFirstRow(
+            otherSession,
+            where: (t) => t.uuidSpaceId.equals(ownerId),
+          );
+          membership = await OfflineSyncSpaceMember.db.insertRow(
+            otherSession,
+            OfflineSyncSpaceMember(
+              spaceId: sharedSpace!.id!,
+              userUuid: userId,
+              role: OfflineSyncSpaceRole.readWrite,
+            ),
+          );
+        });
+
+        group(
+          'when another session revokes and restores membership during READ COMMITTED,',
+          () {
+            late List<server.Person> before;
+            late List<server.Person> revoked;
+            late List<server.Person> restored;
+
+            setUp(() async {
+              await session.db.transactionForUser(
+                userId,
+                (tx) async {
+                  before = await server.Person.db.find(session, transaction: tx);
+
+                  await OfflineSyncSpaceMember.db.deleteRow(otherSession, membership);
+                  revoked = await server.Person.db.find(session, transaction: tx);
+
+                  await OfflineSyncSpaceMember.db.insertRow(otherSession, membership);
+                  restored = await server.Person.db.find(session, transaction: tx);
+                },
+                settings: const TransactionSettings(
+                  isolationLevel: IsolationLevel.readCommitted,
+                ),
+              );
+            });
+
+            test('then successive reads observe both committed changes.', () {
+              expect(before.map((row) => row.id), [person.id]);
+              expect(revoked, isEmpty);
+              expect(restored.map((row) => row.id), [person.id]);
+            });
+          },
+        );
+
+        group('when another session revokes membership during REPEATABLE READ,', () {
+          late List<server.Person> before;
+          late List<server.Person> insideTransaction;
+          late List<server.Person> nextTransaction;
+          late OfflineSyncSpaceMember? committedMembership;
+
+          setUp(() async {
+            await session.db.transactionForUser(
+              userId,
+              (tx) async {
+                before = await server.Person.db.find(session, transaction: tx);
+
+                await OfflineSyncSpaceMember.db.deleteRow(otherSession, membership);
+                committedMembership = await OfflineSyncSpaceMember.db.findById(
+                  otherSession,
+                  membership.id!,
+                );
+                insideTransaction = await server.Person.db.find(
+                  session,
+                  transaction: tx,
+                );
+              },
+              settings: const TransactionSettings(
+                isolationLevel: IsolationLevel.repeatableRead,
+              ),
+            );
+
+            nextTransaction = await session.db.transactionForUser(
+              userId,
+              (tx) => server.Person.db.find(session, transaction: tx),
+            );
+          });
+
+          test('then membership follows the snapshot until the next transaction.', () {
+            expect(before.map((row) => row.id), [person.id]);
+            expect(committedMembership, isNull);
+            expect(insideTransaction.map((row) => row.id), [person.id]);
+            expect(nextTransaction, isEmpty);
+          });
+        });
+      });
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/test_tools/postgres_migrations.dart

@@ -0,0 +1,38 @@
+import 'dart:convert';
+import 'dart:io';
+
+// 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;
+
+// # TODO Remove this workaround once the server uses Postgres properly.
+
+/// Renders the generated schema for an isolated PostgreSQL test database.
+/// The committed SQL targets SQLite; use Serverpod's generator for PostgreSQL.
+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');
+}