Release transactions held by web clients that are gone

drift/CHANGELOG.md

@@ -1,5 +1,9 @@
 ## 2.35.1-dev
 
+- Release transactions held by web clients that are gone: a tab closed
+  while a transaction was open would block database access in every other
+  tab connected to the same worker.
+
 - Web: Fix writes made in transactions or through `RETURNING` statements not being persisted to IndexedDB (#3864).
 
 ## 2.35.0

drift/lib/src/remote/server_impl.dart

@@ -22,7 +22,17 @@ class ServerImplementation implements DriftServer {
   /// Whether this server should close the executor after shutting down.
   final bool closeExecutorWhenShutdown;
 
-  final Map<int, QueryExecutor> _managedExecutors = {};
+  final Map<int, _ManagedExecutor> _managedExecutors = {};
+
+  /// Statements already running against a managed executor.
+  ///
+  /// A handler resolves its executor and then yields (`_runQuery` waits a
+  /// microtask for a cancellation to arrive), so a rollback starting in
+  /// between would release the executor while a statement was still about to
+  /// use it. That statement would then run outside the transaction it was
+  /// issued in: the assertion that catches a closed executor is stripped in
+  /// release builds.
+  final Map<int, List<Future<void>>> _inFlight = {};
   int _currentExecutorId = 0;
   int _knownSchemaVersion = 0;
 
@@ -78,7 +88,10 @@ class ServerImplementation implements DriftServer {
     comm.notify(ServerInfo(connection.dialect));
 
     _activeChannels.add(comm);
-    return comm.closed.then((_) => _activeChannels.remove(comm));
+    return comm.closed.then((_) {
+      _activeChannels.remove(comm);
+      return _abandonExecutorsOf(comm);
+    });
   }
 
   @override
@@ -127,11 +140,17 @@ class ServerImplementation implements DriftServer {
         ),
       );
       _cancellableOperations[request.id] = token;
-      return token.result.whenComplete(
-        () => _cancellableOperations.remove(request.id),
+      return _tracked(
+        payload.executorId,
+        token.result.whenComplete(
+          () => _cancellableOperations.remove(request.id),
+        ),
       );
     } else if (payload is ExecuteBatchedStatement) {
-      return _runBatched(payload.stmts, payload.executorId);
+      return _tracked(
+        payload.executorId,
+        _runBatched(payload.stmts, payload.executorId),
+      );
     } else if (payload is NotifyTablesUpdated) {
       _tableUpdateNotifications.add(payload);
       dispatchTableUpdateNotification(payload, comms);
@@ -198,7 +217,7 @@ class ServerImplementation implements DriftServer {
   Future<QueryExecutor> _loadExecutor(int? transactionId) async {
     await _waitForTurn(transactionId);
     return transactionId != null
-        ? _managedExecutors[transactionId]!
+        ? _managedExecutors[transactionId]!.executor
         : connection;
   }
 
@@ -207,18 +226,22 @@ class ServerImplementation implements DriftServer {
     await transaction.ensureOpen(
       _ServerDbUser(this, comm, _knownSchemaVersion),
     );
-    return _putExecutor(transaction, beforeCurrent: true);
+    return _putOwnedExecutor(transaction, comm);
   }
 
   Future<int> _spawnExclusive(DriftCommunication comm, int? executor) async {
     final exclusive = (await _loadExecutor(executor)).beginExclusive();
     await exclusive.ensureOpen(_ServerDbUser(this, comm, _knownSchemaVersion));
-    return _putExecutor(exclusive, beforeCurrent: true);
+    return _putOwnedExecutor(exclusive, comm);
   }
 
-  int _putExecutor(QueryExecutor executor, {bool beforeCurrent = false}) {
+  int _putExecutor(
+    QueryExecutor executor, {
+    bool beforeCurrent = false,
+    DriftCommunication? owner,
+  }) {
     final id = _currentExecutorId++;
-    _managedExecutors[id] = executor;
+    _managedExecutors[id] = _ManagedExecutor(executor, owner);
 
     if (beforeCurrent && _executorBacklog.isNotEmpty) {
       _executorBacklog.insert(0, id);
@@ -229,6 +252,16 @@ class ServerImplementation implements DriftServer {
     return id;
   }
 
+  int _putOwnedExecutor(QueryExecutor executor, DriftCommunication owner) {
+    final id = _putExecutor(executor, beforeCurrent: true, owner: owner);
+    if (owner.isClosed) {
+      // The client went away while this executor was waiting for its turn,
+      // so nothing will ever commit it or roll it back.
+      unawaited(_abandonExecutor(id));
+    }
+    return id;
+  }
+
   Future<ResponsePayload?> _transactionControl(
     DriftCommunication comm,
     NestedExecutorControl action,
@@ -284,8 +317,64 @@ class ServerImplementation implements DriftServer {
     return null;
   }
 
+  /// Rolls back the transactions and closes the exclusive executors [comm]
+  /// opened and can no longer finish because it closed.
+  ///
+  /// A client that disappears mid-transaction, like a browser tab that is
+  /// closed, otherwise leaves its executor at the head of the backlog. Every
+  /// other client of this server then waits behind it indefinitely.
+  Future<void> _abandonExecutorsOf(DriftCommunication comm) async {
+    // Concurrently: each waits for its own turn, and a parent's turn only
+    // comes once its nested children have ended.
+    await Future.wait([
+      for (final MapEntry(:key, :value) in _managedExecutors.entries)
+        if (value.owner == comm) _abandonExecutor(key),
+    ]);
+  }
+
+  /// Remembers [work] while it runs, so an abandoned executor can wait for it.
+  Future<T> _tracked<T>(int? executorId, Future<T> work) {
+    if (executorId == null) return work;
+
+    final running = _inFlight.putIfAbsent(executorId, () => []);
+    final tracked = work.then<void>((_) {}, onError: (_, _) {});
+    running.add(tracked);
+    tracked.whenComplete(() {
+      running.remove(tracked);
+      if (running.isEmpty) _inFlight.remove(executorId);
+    });
+    return work;
+  }
+
+  Future<void> _abandonExecutor(int id) async {
+    if (!_managedExecutors.containsKey(id)) return;
+    await _waitForTurn(id);
+    // Statements that already hold this executor finish first: rolling back
+    // underneath one would let it run outside its transaction.
+    while (_inFlight[id]?.isNotEmpty ?? false) {
+      await Future.wait(List.of(_inFlight[id]!));
+    }
+    final executor = _managedExecutors[id]?.executor;
+    if (executor == null) return;
+
+    try {
+      if (executor is TransactionExecutor) {
+        await executor.rollback();
+      } else {
+        await executor.close();
+      }
+    } catch (_) {
+      // There is no client left to report this to, and the executor is
+      // released either way - drift already considers a transaction over
+      // after a rollback, successful or not.
+    } finally {
+      _releaseExecutor(id);
+    }
+  }
+
   void _releaseExecutor(int id) {
     _managedExecutors.remove(id);
+    _inFlight.remove(id);
     _executorBacklog.remove(id);
     _notifyActiveExecutorUpdated();
   }
@@ -325,6 +414,20 @@ class ServerImplementation implements DriftServer {
   }
 }
 
+/// A managed executor and the channel that opened it.
+///
+/// A client can disappear without closing its transactions - a closed browser
+/// tab sends no rollback and no close message - so the server has to know who
+/// opened an executor to end it when that client is gone. [owner] is null for
+/// the executors the server itself puts up, such as the one a `beforeOpen`
+/// callback runs on.
+class _ManagedExecutor {
+  final QueryExecutor executor;
+  final DriftCommunication? owner;
+
+  _ManagedExecutor(this.executor, this.owner);
+}
+
 class _ServerDbUser implements QueryExecutorUser {
   final ServerImplementation _server;
   final DriftCommunication connection;

drift/lib/src/web/channel_new.dart

@@ -28,6 +28,10 @@ extension WebPortToChannel on web.MessagePort {
   /// Both "ends" of a JS channel calling [channel] on their part must use the
   /// value for [explicitClose].
   ///
+  /// [peerGone], when set, completes once the other end is known to be gone
+  /// despite never having sent a close message - a closed browser tab, for
+  /// instance. The returned channel is closed when it completes.
+  ///
   /// When [webNativeSerialization] is enabled, the [StreamChannel] can only be
   /// used for drift databases using the `package:drift/remote.dart` protocol
   /// and is not suitable for any other message.
@@ -38,6 +42,7 @@ extension WebPortToChannel on web.MessagePort {
     bool explicitClose = false,
     bool webNativeSerialization = false,
     int nativeSerializionVersion = 0,
+    Future<void>? peerGone,
   }) {
     final controller = StreamChannelController<Object?>();
     final protocol = WebProtocol(
@@ -76,6 +81,10 @@ extension WebPortToChannel on web.MessagePort {
       },
     );
 
+    // Ending the stream is what tells a drift server that this client is
+    // gone, so that it can release the executors the client still held.
+    peerGone?.then((_) => controller.local.sink.close());
+
     return controller.foreign;
   }
 }

drift/lib/src/web/wasm_setup.dart

@@ -259,6 +259,7 @@ final class _ProbeResult implements WasmProbeResult {
           enableMigrations: enableMigrations,
           protocolVersion: sharedWorker!.version,
           newSerialization: sharedWorker.version >= ProtocolVersion.v3,
+          clientLock: await holdClientLock(),
         );
 
         message.sendTo(sharedWorker.send);
@@ -274,6 +275,7 @@ final class _ProbeResult implements WasmProbeResult {
             enableMigrations: enableMigrations,
             protocolVersion: dedicatedWorker.version,
             newSerialization: dedicatedWorker.version >= ProtocolVersion.v3,
+            clientLock: await holdClientLock(),
           );
 
           message.sendTo(dedicatedWorker.send);

drift/lib/src/web/wasm_setup/protocol.dart

@@ -270,6 +270,13 @@ final class ServeDriftDatabase extends WasmInitializationMessage {
   final bool enableMigrations;
   final bool newSerialization;
 
+  /// The name of a Web Lock held by the client for as long as it lives.
+  ///
+  /// Workers request this lock, which is granted to them once the client is
+  /// gone. Sent unconditionally: the field is nullable, so a worker that
+  /// predates it ignores it and a newer worker simply finds it absent.
+  final String? clientLock;
+
   ServeDriftDatabase({
     required this.sqlite3WasmUri,
     required this.port,
@@ -279,6 +286,7 @@ final class ServeDriftDatabase extends WasmInitializationMessage {
     required this.protocolVersion,
     required this.enableMigrations,
     required this.newSerialization,
+    this.clientLock,
   });
 
   factory ServeDriftDatabase.fromJsPayload(JSObject payload) {
@@ -299,6 +307,7 @@ final class ServeDriftDatabase extends WasmInitializationMessage {
           ? (payload['new_serialization'] as JSBoolean).toDart
           : true,
       protocolVersion: version,
+      clientLock: (payload['client_lock'] as JSString?)?.toDart,
     );
   }
 
@@ -312,6 +321,9 @@ final class ServeDriftDatabase extends WasmInitializationMessage {
       ..['initPort'] = initializationPort
       ..['migrations'] = enableMigrations.toJS
       ..['new_serialization'] = newSerialization.toJS;
+    if (clientLock case final lock?) {
+      object['client_lock'] = lock.toJS;
+    }
 
     protocolVersion.writeToJs(object);
 

drift/lib/src/web/wasm_setup/shared.dart

@@ -11,6 +11,7 @@ import 'package:web/web.dart'
         Worker,
         Navigator,
         LockManager,
+        Crypto,
         StorageManager,
         IDBFactory,
         IDBRequest,
@@ -323,6 +324,7 @@ class DriftServerController {
         explicitClose: message.protocolVersion >= ProtocolVersion.v1,
         webNativeSerialization: message.newSerialization,
         nativeSerializionVersion: message.protocolVersion.versionCode,
+        peerGone: clientGone(message.clientLock),
       ),
       // With the new serialization mode, instruct the drift server not to apply
       // its internal serialization logic.
@@ -544,3 +546,41 @@ extension AcquireLock on LockManager {
     return hasLock.future;
   }
 }
+
+// Detecting clients that are gone without having closed their databases.
+//
+// A closed tab sends its worker nothing, so a transaction it had open stays at
+// the head of the drift server's backlog and every other tab's database work
+// waits behind it. `MessagePort`s have no close event for a context going away,
+// but a Web Lock is released when the context holding it is destroyed. Clients
+// therefore hold a uniquely named lock for their entire lifetime and tell the
+// worker its name; the worker requests the same lock, which it is granted only
+// once the client is gone.
+
+@JS('crypto')
+external Crypto get _crypto;
+
+/// Takes a lock that is held until this JavaScript context is gone and returns
+/// its name, or null where Web Locks are unavailable.
+Future<String?> holdClientLock() async {
+  final manager = locks;
+  if (manager == null) return null;
+
+  final name = 'drift-client-${_crypto.randomUUID()}';
+  // The completer returning the lock is never completed, so the browser
+  // releases it only when this context is destroyed.
+  await manager.acquire(name, Completer<void>());
+  return name;
+}
+
+/// Completes once the client holding [lockName] is gone, or null when that
+/// cannot be observed.
+Future<void>? clientGone(String? lockName) {
+  final manager = locks;
+  if (lockName == null || manager == null) return null;
+
+  // Granted once the client has released the lock, which happens when the
+  // context holding it goes away.
+  final release = Completer<void>();
+  return manager.acquire(lockName, release).then((_) => release.complete());
+}