From 908b4a8dac3459849c9777799dd4e9bd1b5cc46e Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Mon, 17 Aug 2026 01:48:48 +0200 Subject: [PATCH] refactor(database): serialize durable mutations through one queue, keep only guarded watch-action mutators MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 48 +++++-------------- test/database/app_database_test.dart | 14 +++--- .../database/tvos_database_recovery_test.dart | 9 ++-- .../offline_watch_sync_service_test.dart | 2 +- 4 files changed, 25 insertions(+), 48 deletions(-) diff --git a/lib/database/app_database.dart b/lib/database/app_database.dart index cab808147..412c74dbe 100644 --- a/lib/database/app_database.dart +++ b/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.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 deleteWatchAction(int id) { - return _runPendingMutation(() async { - await (delete(offlineWatchProgress)..where((t) => t.id.equals(id))).go(); - }); - } - - /// Update retry state outside a snapshotted replay. - Future 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 getPendingSyncCount({String? profileId, int? maxSyncAttempts}) async { final query = selectOnly(offlineWatchProgress)..addColumns([offlineWatchProgress.id.count()]); diff --git a/test/database/app_database_test.dart b/test/database/app_database_test.dart index 0a467533e..685455ea3 100644 --- a/test/database/app_database_test.dart +++ b/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)); }); diff --git a/test/database/tvos_database_recovery_test.dart b/test/database/tvos_database_recovery_test.dart index 1d33d7b78..d3235e719 100644 --- a/test/database/tvos_database_recovery_test.dart +++ b/test/database/tvos_database_recovery_test.dart @@ -164,8 +164,9 @@ Future _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'), diff --git a/test/services/offline_watch_sync_service_test.dart b/test/services/offline_watch_sync_service_test.dart index 800ebac24..07647f1c4 100644 --- a/test/services/offline_watch_sync_service_test.dart +++ b/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'); }