refactor(database): serialize durable mutations through one queue, keep only guarded watch-action mutators

Every identity/pending mutation nested a per-instance SerialFutureQueue inside the static tvOS-recovery queue, so the per-instance layer added no serialization domain (reentrancy is already prevented by the durability Zone check). The unguarded updateSyncAttempt/deleteWatchAction pair had zero production callers — the sync service deliberately uses the revision-guarded variants.

Tests migrated to the IfUnchanged pair by passing the row's current revision.

lib/database/app_database.dart

@@ -81,7 +81,6 @@ class AppDatabase extends _$AppDatabase {
   AppDatabase._withRecovery(super.e, this._recoveryStore);
 
   final TvosDatabaseRecoveryStore? _recoveryStore;
-  final SerialFutureQueue _durabilityQueue = SerialFutureQueue();
   static final Object _durabilityZoneKey = Object();
   static final SerialFutureQueue _tvosRecoveryQueue = SerialFutureQueue();
 
@@ -156,12 +155,10 @@ class AppDatabase extends _$AppDatabase {
     final store = _recoveryStore;
     if (store == null || !store.isTvos) return Future<void>.value();
 
-    return _durabilityQueue.run(
-      () => _tvosRecoveryQueue.run(
-        () => store.acknowledgeRecoveryRequired(
-          readIdentity: _readProtectedIdentityRecoveryRows,
-          readPending: _readPendingRecoveryRows,
-        ),
+    return _tvosRecoveryQueue.run(
+      () => store.acknowledgeRecoveryRequired(
+        readIdentity: _readProtectedIdentityRecoveryRows,
+        readPending: _readPendingRecoveryRows,
       ),
     );
   }
@@ -175,17 +172,15 @@ class AppDatabase extends _$AppDatabase {
     if (store == null || !store.isTvos) return mutation();
     if (Zone.current[_durabilityZoneKey] == this) return mutation();
 
-    return _durabilityQueue.run(
-      () => _tvosRecoveryQueue.run(
-        () => runZoned(
-          () => store.runDurableMutation(
-            group: group,
-            mutation: mutation,
-            readIdentity: _readProtectedIdentityRecoveryRows,
-            readPending: _readPendingRecoveryRows,
-          ),
-          zoneValues: {_durabilityZoneKey: this},
+    return _tvosRecoveryQueue.run(
+      () => runZoned(
+        () => store.runDurableMutation(
+          group: group,
+          mutation: mutation,
+          readIdentity: _readProtectedIdentityRecoveryRows,
+          readPending: _readPendingRecoveryRows,
         ),
+        zoneValues: {_durabilityZoneKey: this},
       ),
     );
   }
@@ -1057,25 +1052,6 @@ class AppDatabase extends _$AppDatabase {
     });
   }
 
-  /// Delete a specific watch action outside a snapshotted replay.
-  Future<void> deleteWatchAction(int id) {
-    return _runPendingMutation(() async {
-      await (delete(offlineWatchProgress)..where((t) => t.id.equals(id))).go();
-    });
-  }
-
-  /// Update retry state outside a snapshotted replay.
-  Future<void> updateSyncAttempt(int id, String? errorMessage) {
-    return _runPendingMutation(() async {
-      final existing = await (select(offlineWatchProgress)..where((t) => t.id.equals(id))).getSingleOrNull();
-      if (existing == null) return;
-
-      await (update(offlineWatchProgress)..where((t) => t.id.equals(id))).write(
-        OfflineWatchProgressCompanion(syncAttempts: Value(existing.syncAttempts + 1), lastError: Value(errorMessage)),
-      );
-    });
-  }
-
   /// Get count of pending sync items
   Future<int> getPendingSyncCount({String? profileId, int? maxSyncAttempts}) async {
     final query = selectOnly(offlineWatchProgress)..addColumns([offlineWatchProgress.id.count()]);

test/database/app_database_test.dart

@@ -2008,33 +2008,33 @@ class _AppDatabaseTestSuite {
         expect(await db.getLatestWatchActionsForKeys({}), isEmpty);
       });
 
-      test('updateSyncAttempt increments syncAttempts and stores lastError', () async {
+      test('updateSyncAttemptIfUnchanged increments syncAttempts and stores lastError', () async {
         await db.insertWatchAction(serverId: ServerId('s'), ratingKey: '1', actionType: OfflineActionType.watched.id);
         final inserted = (await db.select(db.offlineWatchProgress).get()).single;
 
-        await db.updateSyncAttempt(inserted.id, 'boom');
+        expect(await db.updateSyncAttemptIfUnchanged(inserted.id, inserted.updatedAt, 'boom'), isTrue);
         var row = (await db.select(db.offlineWatchProgress).get()).single;
         expect(row.syncAttempts, 1);
         expect(row.lastError, 'boom');
 
-        await db.updateSyncAttempt(inserted.id, null);
+        expect(await db.updateSyncAttemptIfUnchanged(row.id, row.updatedAt, null), isTrue);
         row = (await db.select(db.offlineWatchProgress).get()).single;
         expect(row.syncAttempts, 2);
         expect(row.lastError, isNull);
       });
 
-      test('updateSyncAttempt is a no-op when id does not exist', () async {
-        await db.updateSyncAttempt(999, 'irrelevant');
+      test('updateSyncAttemptIfUnchanged is a no-op when id does not exist', () async {
+        expect(await db.updateSyncAttemptIfUnchanged(999, 0, 'irrelevant'), isFalse);
         expect(await db.select(db.offlineWatchProgress).get(), isEmpty);
       });
 
-      test('deleteWatchAction removes only the matching row', () async {
+      test('deleteWatchActionIfUnchanged removes only the matching row', () async {
         await db.insertWatchAction(serverId: ServerId('s'), ratingKey: '1', actionType: OfflineActionType.watched.id);
         await db.insertWatchAction(serverId: ServerId('s'), ratingKey: '2', actionType: OfflineActionType.watched.id);
         final rows = await db.select(db.offlineWatchProgress).get();
         expect(rows, hasLength(2));
 
-        await db.deleteWatchAction(rows.first.id);
+        expect(await db.deleteWatchActionIfUnchanged(rows.first.id, rows.first.updatedAt), isTrue);
         expect(await db.select(db.offlineWatchProgress).get(), hasLength(1));
       });
 

test/database/tvos_database_recovery_test.dart

@@ -164,8 +164,9 @@ Future<void> _seedCriticalRows(AppDatabase db) async {
       OfflineWatchProgressCompanion(createdAt: Value(3000 + index), updatedAt: Value(4000 + index)),
     );
   }
-  await db.updateSyncAttempt(rows.first.id, 'retry-without-protected-payload');
-  await db.updateSyncAttempt(rows.first.id, 'retry-without-protected-payload');
+  final revisedFirst = await (db.select(db.offlineWatchProgress)..where((t) => t.id.equals(rows.first.id))).getSingle();
+  await db.updateSyncAttemptIfUnchanged(revisedFirst.id, revisedFirst.updatedAt, 'retry-without-protected-payload');
+  await db.updateSyncAttemptIfUnchanged(revisedFirst.id, revisedFirst.updatedAt, 'retry-without-protected-payload');
 }
 
 void main() {
@@ -1023,8 +1024,8 @@ void main() {
       actionType: OfflineActionType.watched.id,
     );
     final rows = await result.database.getPendingWatchActions();
-    await result.database.updateSyncAttempt(rows.first.id, 'retry');
-    await result.database.deleteWatchAction(rows.last.id);
+    await result.database.updateSyncAttemptIfUnchanged(rows.first.id, rows.first.updatedAt, 'retry');
+    await result.database.deleteWatchActionIfUnchanged(rows.last.id, rows.last.updatedAt);
     await result.database.insertWatchAction(
       profileId: 'p2',
       serverId: ServerId('s1'),

test/services/offline_watch_sync_service_test.dart

@@ -338,7 +338,7 @@ void main() {
       await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
       var action = await db.getLatestWatchAction('srv:42');
       for (var i = 0; i < OfflineWatchSyncService.maxSyncAttempts; i++) {
-        await db.updateSyncAttempt(action!.id, 'server error');
+        await db.updateSyncAttemptIfUnchanged(action!.id, action.updatedAt, 'server error');
         action = await db.getLatestWatchAction('srv:42');
       }