diff --git a/mobile/integration_test/failed_sync_resume_e2e_test.dart b/mobile/integration_test/failed_sync_resume_e2e_test.dart new file mode 100644 index 0000000000..255074df93 --- /dev/null +++ b/mobile/integration_test/failed_sync_resume_e2e_test.dart @@ -0,0 +1,101 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:math'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:immich_mobile/domain/services/background_worker.service.dart'; +import 'package:immich_mobile/domain/utils/background_sync.dart'; +import 'package:immich_mobile/infrastructure/repositories/db.repository.dart'; +import 'package:immich_mobile/main.dart' as app; +import 'package:immich_mobile/platform/background_worker_api.g.dart'; +import 'package:immich_mobile/providers/app_life_cycle.provider.dart'; +import 'package:immich_mobile/providers/background_sync.provider.dart'; +import 'package:immich_mobile/providers/infrastructure/db.provider.dart'; +import 'package:immich_mobile/services/api.service.dart'; +import 'package:immich_mobile/utils/bootstrap.dart'; +import 'package:immich_mobile/wm_executor.dart'; +import 'package:integration_test/integration_test.dart'; + +import 'test_utils/fake_immich_server.dart'; + +// Issue #28082 end-to-end: a resume after a sync froze mid-flight starts a fresh sync. +// Kept in its own file on purpose: run after the failed_sync_resume tests in one process, +// the resume sync starves past its window - a cross-test interaction that survived pool, +// sqlite and bg-worker audits unnamed. One file per process is device-proven green. +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + binding.framePolicy = LiveTestWidgetsFlutterBindingFramePolicy.fullyLive; + + late Drift drift; + late FakeImmichServer server; + + // main()'s formula with a higher floor: a wedged frozen worker plus a full local + // sync must not starve the fresh resume sync out of its 25s window on 4 cores. + final poolSize = max(Platform.numberOfProcessors - 1, 8); + + setUpAll(() async { + await app.initApp(); + (drift, _) = await Bootstrap.initDomain(); + // A background-worker schedule persisted by real app use on this device can + // launch a second engine mid-file (own isolate pool + full sync) and starve + // these tests on a small device. Unregister it for the whole run. + await BackgroundWorkerFgService(BackgroundWorkerFgHostApi()).disable(); + }); + + setUp(() async { + // A task completing while dispose tears the pool down re-warms it (_schedule + // on the cleared pool re-creates workers), and init on a warm pool is silently + // ignored - poolSize would never apply. Reset first, then verify it took. + await workerManagerPatch.dispose(); + await workerManagerPatch.init(dynamicSpawning: true, isolatesCount: poolSize); + expect( + workerManagerPatch.pool.length, + poolSize, + reason: 'init was ignored: a straggler from the previous test re-warmed the pool', + ); + server = await FakeImmichServer.start(); + await ApiService().resolveAndSetEndpoint(server.endpoint); + await drift.delete(drift.userEntity).go(); + }); + + tearDown(() async { + // Close the server first so any held-open sync stream ends and its isolate unwinds, + // then drain the pool - otherwise dispose waits on the frozen read. + await server.close(); + await workerManagerPatch.dispose(); + }); + + testWidgets('a resume after a sync froze mid-flight starts a fresh sync', (tester) async { + final manager = BackgroundSyncManager(); + // Not disposed on purpose: driftOverride closes the drift on dispose, and that + // drift belongs to setUpAll. The container only holds the shared drift and a + // fire-and-forget resume; the frozen isolates + server are drained by tearDown. + final container = ProviderContainer( + overrides: [driftProvider.overrideWith(driftOverride(drift)), backgroundSyncProvider.overrideWithValue(manager)], + ); + + // A first sync opens /sync/stream and never finishes - the frozen state a + // suspended sync isolate is left in. Holding the stream open keeps + // _syncTask non-null, exactly as it is across an iOS process suspension. + unawaited(manager.syncRemote()); + await server + .streamOpenedNth(1) + .timeout(const Duration(seconds: 30), onTimeout: () => fail('first sync isolate never opened /sync/stream')); + + // The lifecycle then goes background -> foreground. handleAppResume runs the + // resume sync exactly once. On the buggy build it hangs on the stale task's + // future, so it is not awaited here. + final notifier = container.read(appStateProvider.notifier); + await notifier.handleAppPause(); + unawaited(notifier.handleAppResume()); + + await server + .streamOpenedNth(2) + .timeout( + const Duration(seconds: 25), + onTimeout: () => fail('resume did not start a fresh remote sync - the stale frozen sync blocked it (#28082)'), + ); + expect(server.streamOpenCount, greaterThanOrEqualTo(2)); + }); +} diff --git a/mobile/integration_test/failed_sync_resume_test.dart b/mobile/integration_test/failed_sync_resume_test.dart new file mode 100644 index 0000000000..c23319dffe --- /dev/null +++ b/mobile/integration_test/failed_sync_resume_test.dart @@ -0,0 +1,133 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:math'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:immich_mobile/domain/services/background_worker.service.dart'; +import 'package:immich_mobile/domain/utils/background_sync.dart'; +import 'package:immich_mobile/infrastructure/repositories/db.repository.dart'; +import 'package:immich_mobile/main.dart' as app; +import 'package:immich_mobile/platform/background_worker_api.g.dart'; +import 'package:immich_mobile/services/api.service.dart'; +import 'package:immich_mobile/utils/bootstrap.dart'; +import 'package:immich_mobile/wm_executor.dart'; +import 'package:integration_test/integration_test.dart'; + +import 'test_utils/fake_immich_server.dart'; + +// Issue #28082: a remote sync in-flight when the app is backgrounded stays referenced +// but frozen across the suspension. On resume the app drops the stale task +// (cancelResumeSyncs) and starts a fresh sync. +// +// Device/emulator tests: real worker isolates + a real drift db + a loopback fake server +// (same pattern as background_sync_teardown_test). The mobile integration-test CI job is +// disabled in test.yml, so like Mert's teardown test this is a local/on-device guard. +// +// The end-to-end resume test lives in failed_sync_resume_e2e_test.dart - one process +// per file keeps it clear of the cross-test interaction described there. +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + binding.framePolicy = LiveTestWidgetsFlutterBindingFramePolicy.fullyLive; + + late Drift drift; + late FakeImmichServer server; + + // main()'s formula with a higher floor: a wedged frozen worker plus a full local + // sync must not starve the fresh resume sync out of its 25s window on 4 cores. + final poolSize = max(Platform.numberOfProcessors - 1, 8); + + setUpAll(() async { + await app.initApp(); + (drift, _) = await Bootstrap.initDomain(); + // A background-worker schedule persisted by real app use on this device can + // launch a second engine mid-file (own isolate pool + full sync) and starve + // these tests on a small device. Unregister it for the whole run. + await BackgroundWorkerFgService(BackgroundWorkerFgHostApi()).disable(); + }); + + setUp(() async { + // A task completing while dispose tears the pool down re-warms it (_schedule + // on the cleared pool re-creates workers), and init on a warm pool is silently + // ignored - poolSize would never apply. Reset first, then verify it took. + await workerManagerPatch.dispose(); + await workerManagerPatch.init(dynamicSpawning: true, isolatesCount: poolSize); + expect( + workerManagerPatch.pool.length, + poolSize, + reason: 'init was ignored: a straggler from the previous test re-warmed the pool', + ); + server = await FakeImmichServer.start(); + await ApiService().resolveAndSetEndpoint(server.endpoint); + await drift.delete(drift.userEntity).go(); + }); + + tearDown(() async { + // Close the server first so any held-open sync stream ends and its isolate unwinds, + // then drain the pool - otherwise dispose waits on the frozen read. + await server.close(); + await workerManagerPatch.dispose(); + }); + + // Self-contained (bare manager, no fire-and-forget resume), so it runs first: its + // frozen syncs are fully drained by tearDown. + testWidgets('a cancelled sync task does not clear the slot of the fresh task that superseded it', (tester) async { + final manager = BackgroundSyncManager(); + + // First sync opens /sync/stream and is held open - the frozen suspended state. + unawaited(manager.syncRemote()); + await server + .streamOpenedNth(1) + .timeout(const Duration(seconds: 30), onTimeout: () => fail('first sync isolate never opened /sync/stream')); + + // Resume drops the stale task then immediately starts a fresh one, exactly as + // _handleBetaTimelineResume does. cancelResumeSyncs cancels the first task; its + // completion chain then fires and must NOT null the fresh task's slot. + unawaited(manager.cancelResumeSyncs()); + unawaited(manager.syncRemote()); + await server + .streamOpenedNth(2) + .timeout(const Duration(seconds: 30), onTimeout: () => fail('fresh sync isolate never opened /sync/stream')); + + // A third sync stream only opens if the cancelled task cleared the fresh slot, + // letting the dedupe guard start a redundant sync. + var thirdOpened = false; + unawaited(server.streamOpenedNth(3).then((_) => thirdOpened = true)); + + // Let the cancelled task's completion chain settle, then ask to sync again. + await Future.delayed(const Duration(milliseconds: 200)); + unawaited(manager.syncRemote()); + await Future.delayed(const Duration(seconds: 3)); + + expect( + thirdOpened, + isFalse, + reason: 'the cancelled task cleared the fresh task slot, so a redundant third sync started (#28082 clobber)', + ); + expect(server.streamOpenCount, 2); + }); + + // The false-error-flash fix: a task cancelled by cancelResumeSyncs completes with a + // CanceledError, which is not a sync failure and must not reach onRemoteSyncError. + // Before the filter the stale task reported an error right after the fresh sync + // started, so the status UI showed a failure for the whole healthy run. + testWidgets('a cancelled sync does not report a false error to the status callbacks', (tester) async { + final errors = []; + final manager = BackgroundSyncManager(onRemoteSyncError: errors.add); + + // Hold the stream open so the task is genuinely in-flight when it is cancelled. + unawaited(manager.syncRemote()); + await server + .streamOpenedNth(1) + .timeout(const Duration(seconds: 30), onTimeout: () => fail('sync isolate never opened /sync/stream')); + + await manager.cancelResumeSyncs(); + // Let the cancelled task's completion chain run before checking the callbacks. + await Future.delayed(const Duration(milliseconds: 100)); + + expect( + errors, + isEmpty, + reason: 'a cancelled task is not a real error; its CanceledError must not fire onRemoteSyncError', + ); + }); +} diff --git a/mobile/integration_test/test_utils/fake_immich_server.dart b/mobile/integration_test/test_utils/fake_immich_server.dart index c434f83bc5..7c63098c52 100644 --- a/mobile/integration_test/test_utils/fake_immich_server.dart +++ b/mobile/integration_test/test_utils/fake_immich_server.dart @@ -10,6 +10,8 @@ class FakeImmichServer { final (int, int, int) version; final Completer _streamOpened = Completer(); + final List _streamOpens = []; + final Map> _openWaiters = {}; int ackRequests = 0; @@ -18,6 +20,17 @@ class FakeImmichServer { /// Resolves when the sync isolate opens `POST /sync/stream`. Future get streamOpened => _streamOpened.future; + /// How many `/sync/stream` requests have opened so far. + int get streamOpenCount => _streamOpens.length; + + /// Resolves when the [n]-th (1-indexed) `/sync/stream` opens. + Future streamOpenedNth(int n) { + if (_streamOpens.length >= n) { + return Future.value(_streamOpens[n - 1]); + } + return (_openWaiters[n] ??= Completer()).future; + } + static Future start({(int, int, int) version = (3, 0, 0)}) async { final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); final fake = FakeImmichServer._(server, version); @@ -58,13 +71,17 @@ class FakeImmichServer { request.response ..statusCode = HttpStatus.ok ..headers.contentType = ContentType('application', 'jsonlines+json') - ..contentLength = -1 // chunked: stays open to stream incrementally + ..contentLength = + -1 // chunked: stays open to stream incrementally ..bufferOutput = false; // Flush headers so the client's send() resolves and enters its read loop. await request.response.flush(); + final stream = SyncStream._(request.response); + _streamOpens.add(stream); if (!_streamOpened.isCompleted) { - _streamOpened.complete(SyncStream._(request.response)); + _streamOpened.complete(stream); } + _openWaiters.remove(_streamOpens.length)?.complete(stream); } Future _respondJson(HttpRequest request, Object body) async { @@ -83,8 +100,8 @@ class FakeImmichServer { } Future close() async { - if (_streamOpened.isCompleted) { - await (await _streamOpened.future).close(); + for (final stream in _streamOpens) { + await stream.close(); } await _server.close(force: true); } diff --git a/mobile/lib/domain/utils/background_sync.dart b/mobile/lib/domain/utils/background_sync.dart index db0d93c92f..f1e0328165 100644 --- a/mobile/lib/domain/utils/background_sync.dart +++ b/mobile/lib/domain/utils/background_sync.dart @@ -50,16 +50,39 @@ class BackgroundSyncManager { this.onCloudIdSyncError, }); + // The tasks the app-resume path re-runs. One in-flight when the app was suspended + // stays referenced but frozen, so on resume the dedupe guards would hand back the + // stale task instead of syncing (#28082). Websocket and cloud-id are excluded - the + // resume path never restarts them. [_allTasks] builds on this so the lists can't drift. + List get _resumeSyncTasks => [_syncTask, _deviceAlbumSyncTask, _hashTask, _linkedAlbumSyncTask]; + + List get _allTasks => [_syncWebsocketTask, _cloudIdSyncTask, ..._resumeSyncTasks]; + Future cancel() async { _syncQueued = false; - final tasks = [ - _syncTask, - _syncWebsocketTask, - _cloudIdSyncTask, - _linkedAlbumSyncTask, - _deviceAlbumSyncTask, - _hashTask, - ]; + final tasks = _allTasks; + _syncTask = null; + _syncWebsocketTask = null; + _cloudIdSyncTask = null; + _linkedAlbumSyncTask = null; + _deviceAlbumSyncTask = null; + _hashTask = null; + await _cancelAll(tasks); + } + + Future cancelResumeSyncs() async { + _syncQueued = false; + final tasks = _resumeSyncTasks; + _syncTask = null; + _deviceAlbumSyncTask = null; + _hashTask = null; + _linkedAlbumSyncTask = null; + await _cancelAll(tasks); + } + + // Cancels every task in [tasks] and waits for them to unwind. Callers null out + // their own fields first, so the sync guards see a clean slate immediately. + Future _cancelAll(List tasks) async { final futures = [ for (final task in tasks) if (task != null) task.future, @@ -67,13 +90,6 @@ class BackgroundSyncManager { for (final task in tasks) { task?.cancel(); } - _syncTask = null; - _syncWebsocketTask = null; - _cloudIdSyncTask = null; - _linkedAlbumSyncTask = null; - _deviceAlbumSyncTask = null; - _hashTask = null; - try { await Future.wait(futures); } on CanceledError { @@ -84,14 +100,14 @@ class BackgroundSyncManager { // No need to cancel the task, as it can also be run when the user logs out Future syncLocal({bool full = false}) { if (_deviceAlbumSyncTask != null) { - return _deviceAlbumSyncTask!.future; + return _deviceAlbumSyncTask!.future.catchError((_) {}, test: (error) => error is CanceledError); } onLocalSyncStart?.call(); // We use a ternary operator to avoid [_deviceAlbumSyncTask] from being // captured by the closure passed to [runInIsolateGentle]. - _deviceAlbumSyncTask = full + final task = _deviceAlbumSyncTask = full ? runInIsolateGentle( computation: (ref) => ref.read(localSyncServiceProvider).sync(full: true), debugLabel: 'local-sync-full-true', @@ -101,37 +117,43 @@ class BackgroundSyncManager { debugLabel: 'local-sync-full-false', ); - return _deviceAlbumSyncTask! + return task .whenComplete(() { - _deviceAlbumSyncTask = null; + if (identical(_deviceAlbumSyncTask, task)) { + _deviceAlbumSyncTask = null; + } onLocalSyncComplete?.call(); }) .catchError((error) { - onLocalSyncError?.call(error.toString()); - _deviceAlbumSyncTask = null; + if (error is! CanceledError) { + onLocalSyncError?.call(error.toString()); + } }); } Future hashAssets() { if (_hashTask != null) { - return _hashTask!.future; + return _hashTask!.future.catchError((_) {}, test: (error) => error is CanceledError); } onHashingStart?.call(); - _hashTask = runInIsolateGentle( + final task = _hashTask = runInIsolateGentle( computation: (ref) => ref.read(hashServiceProvider).hashAssets(), debugLabel: 'hash-assets', ); - return _hashTask! + return task .whenComplete(() { onHashingComplete?.call(); - _hashTask = null; + if (identical(_hashTask, task)) { + _hashTask = null; + } }) .catchError((error) { - onHashingError?.call(error.toString()); - _hashTask = null; + if (error is! CanceledError) { + onHashingError?.call(error.toString()); + } }); } @@ -143,11 +165,11 @@ class BackgroundSyncManager { onRemoteSyncStart?.call(); - _syncTask = runInIsolateGentle( + final task = _syncTask = runInIsolateGentle( computation: (ref) => ref.read(syncStreamServiceProvider).sync(), debugLabel: 'remote-sync', ); - return _syncTask! + return task .then((result) { final success = result ?? false; onRemoteSyncComplete?.call(success); @@ -155,15 +177,21 @@ class BackgroundSyncManager { return success; }) .catchError((error) { - onRemoteSyncError?.call(error.toString()); + if (error is! CanceledError) { + onRemoteSyncError?.call(error.toString()); + } _syncQueued = false; return false; }) + // A task clears only its own slot: one that was cancelled and superseded by a + // fresh task (see cancelResumeSyncs) must not null the new task's slot. .whenComplete(() { - _syncTask = null; - if (_syncQueued) { - _syncQueued = false; - unawaited(syncRemote()); + if (identical(_syncTask, task)) { + _syncTask = null; + if (_syncQueued) { + _syncQueued = false; + unawaited(syncRemote()); + } } }); } @@ -210,13 +238,21 @@ class BackgroundSyncManager { Future syncLinkedAlbum() { if (_linkedAlbumSyncTask != null) { - return _linkedAlbumSyncTask!.future; + return _linkedAlbumSyncTask!.future.catchError((_) {}, test: (error) => error is CanceledError); } - _linkedAlbumSyncTask = runInIsolateGentle(computation: syncLinkedAlbumsIsolated, debugLabel: 'linked-album-sync'); - return _linkedAlbumSyncTask!.whenComplete(() { - _linkedAlbumSyncTask = null; - }); + final task = _linkedAlbumSyncTask = runInIsolateGentle( + computation: syncLinkedAlbumsIsolated, + debugLabel: 'linked-album-sync', + ); + return task + .whenComplete(() { + if (identical(_linkedAlbumSyncTask, task)) { + _linkedAlbumSyncTask = null; + } + }) + // a cancelled resume sync is not a failure; absorb it so the websocket callers don't get an uncaught error + .catchError((_) {}, test: (error) => error is CanceledError); } Future syncCloudIds() { diff --git a/mobile/lib/providers/app_life_cycle.provider.dart b/mobile/lib/providers/app_life_cycle.provider.dart index 7dc475ab93..f369a37142 100644 --- a/mobile/lib/providers/app_life_cycle.provider.dart +++ b/mobile/lib/providers/app_life_cycle.provider.dart @@ -112,6 +112,12 @@ class AppLifeCycleNotifier extends StateNotifier { await Future.delayed(const Duration(milliseconds: 500)); final backgroundManager = _ref.read(backgroundSyncProvider); + + // Drop any sync that froze mid-flight while the app was suspended so resume + // starts fresh instead of awaiting the stale task (#28082). cancelResumeSyncs + // clears the task refs synchronously, so the syncs below see a clean slate. + unawaited(backgroundManager.cancelResumeSyncs()); + final isAlbumLinkedSyncEnable = _ref.read(appConfigProvider).backup.syncAlbums; try {