fix(mobile): don't let a frozen sync block syncing on resume (#29870)

* fix(mobile): don't let a frozen sync block syncing on resume

* share the task lists between the cancel paths

* fix analyzer issues in resume integration test

* isolate the resume e2e test and pin the worker pool setup
This commit is contained in:
Santo Shakil 2026-08-10 21:39:38 +06:00 committed by GitHub
parent 9862e50aab
commit 0ff47f4178
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 336 additions and 43 deletions

View file

@ -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));
});
}

View file

@ -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 = <String>[];
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',
);
});
}

View file

@ -10,6 +10,8 @@ class FakeImmichServer {
final (int, int, int) version; final (int, int, int) version;
final Completer<SyncStream> _streamOpened = Completer<SyncStream>(); final Completer<SyncStream> _streamOpened = Completer<SyncStream>();
final List<SyncStream> _streamOpens = [];
final Map<int, Completer<SyncStream>> _openWaiters = {};
int ackRequests = 0; int ackRequests = 0;
@ -18,6 +20,17 @@ class FakeImmichServer {
/// Resolves when the sync isolate opens `POST /sync/stream`. /// Resolves when the sync isolate opens `POST /sync/stream`.
Future<SyncStream> get streamOpened => _streamOpened.future; Future<SyncStream> 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<SyncStream> streamOpenedNth(int n) {
if (_streamOpens.length >= n) {
return Future.value(_streamOpens[n - 1]);
}
return (_openWaiters[n] ??= Completer<SyncStream>()).future;
}
static Future<FakeImmichServer> start({(int, int, int) version = (3, 0, 0)}) async { static Future<FakeImmichServer> start({(int, int, int) version = (3, 0, 0)}) async {
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
final fake = FakeImmichServer._(server, version); final fake = FakeImmichServer._(server, version);
@ -58,13 +71,17 @@ class FakeImmichServer {
request.response request.response
..statusCode = HttpStatus.ok ..statusCode = HttpStatus.ok
..headers.contentType = ContentType('application', 'jsonlines+json') ..headers.contentType = ContentType('application', 'jsonlines+json')
..contentLength = -1 // chunked: stays open to stream incrementally ..contentLength =
-1 // chunked: stays open to stream incrementally
..bufferOutput = false; ..bufferOutput = false;
// Flush headers so the client's send() resolves and enters its read loop. // Flush headers so the client's send() resolves and enters its read loop.
await request.response.flush(); await request.response.flush();
final stream = SyncStream._(request.response);
_streamOpens.add(stream);
if (!_streamOpened.isCompleted) { if (!_streamOpened.isCompleted) {
_streamOpened.complete(SyncStream._(request.response)); _streamOpened.complete(stream);
} }
_openWaiters.remove(_streamOpens.length)?.complete(stream);
} }
Future<void> _respondJson(HttpRequest request, Object body) async { Future<void> _respondJson(HttpRequest request, Object body) async {
@ -83,8 +100,8 @@ class FakeImmichServer {
} }
Future<void> close() async { Future<void> close() async {
if (_streamOpened.isCompleted) { for (final stream in _streamOpens) {
await (await _streamOpened.future).close(); await stream.close();
} }
await _server.close(force: true); await _server.close(force: true);
} }

View file

@ -50,16 +50,39 @@ class BackgroundSyncManager {
this.onCloudIdSyncError, 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<Cancelable?> get _resumeSyncTasks => [_syncTask, _deviceAlbumSyncTask, _hashTask, _linkedAlbumSyncTask];
List<Cancelable?> get _allTasks => [_syncWebsocketTask, _cloudIdSyncTask, ..._resumeSyncTasks];
Future<void> cancel() async { Future<void> cancel() async {
_syncQueued = false; _syncQueued = false;
final tasks = [ final tasks = _allTasks;
_syncTask, _syncTask = null;
_syncWebsocketTask, _syncWebsocketTask = null;
_cloudIdSyncTask, _cloudIdSyncTask = null;
_linkedAlbumSyncTask, _linkedAlbumSyncTask = null;
_deviceAlbumSyncTask, _deviceAlbumSyncTask = null;
_hashTask, _hashTask = null;
]; await _cancelAll(tasks);
}
Future<void> 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<void> _cancelAll(List<Cancelable?> tasks) async {
final futures = [ final futures = [
for (final task in tasks) for (final task in tasks)
if (task != null) task.future, if (task != null) task.future,
@ -67,13 +90,6 @@ class BackgroundSyncManager {
for (final task in tasks) { for (final task in tasks) {
task?.cancel(); task?.cancel();
} }
_syncTask = null;
_syncWebsocketTask = null;
_cloudIdSyncTask = null;
_linkedAlbumSyncTask = null;
_deviceAlbumSyncTask = null;
_hashTask = null;
try { try {
await Future.wait(futures); await Future.wait(futures);
} on CanceledError { } 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 // No need to cancel the task, as it can also be run when the user logs out
Future<void> syncLocal({bool full = false}) { Future<void> syncLocal({bool full = false}) {
if (_deviceAlbumSyncTask != null) { if (_deviceAlbumSyncTask != null) {
return _deviceAlbumSyncTask!.future; return _deviceAlbumSyncTask!.future.catchError((_) {}, test: (error) => error is CanceledError);
} }
onLocalSyncStart?.call(); onLocalSyncStart?.call();
// We use a ternary operator to avoid [_deviceAlbumSyncTask] from being // We use a ternary operator to avoid [_deviceAlbumSyncTask] from being
// captured by the closure passed to [runInIsolateGentle]. // captured by the closure passed to [runInIsolateGentle].
_deviceAlbumSyncTask = full final task = _deviceAlbumSyncTask = full
? runInIsolateGentle( ? runInIsolateGentle(
computation: (ref) => ref.read(localSyncServiceProvider).sync(full: true), computation: (ref) => ref.read(localSyncServiceProvider).sync(full: true),
debugLabel: 'local-sync-full-true', debugLabel: 'local-sync-full-true',
@ -101,37 +117,43 @@ class BackgroundSyncManager {
debugLabel: 'local-sync-full-false', debugLabel: 'local-sync-full-false',
); );
return _deviceAlbumSyncTask! return task
.whenComplete(() { .whenComplete(() {
_deviceAlbumSyncTask = null; if (identical(_deviceAlbumSyncTask, task)) {
_deviceAlbumSyncTask = null;
}
onLocalSyncComplete?.call(); onLocalSyncComplete?.call();
}) })
.catchError((error) { .catchError((error) {
onLocalSyncError?.call(error.toString()); if (error is! CanceledError) {
_deviceAlbumSyncTask = null; onLocalSyncError?.call(error.toString());
}
}); });
} }
Future<void> hashAssets() { Future<void> hashAssets() {
if (_hashTask != null) { if (_hashTask != null) {
return _hashTask!.future; return _hashTask!.future.catchError((_) {}, test: (error) => error is CanceledError);
} }
onHashingStart?.call(); onHashingStart?.call();
_hashTask = runInIsolateGentle( final task = _hashTask = runInIsolateGentle(
computation: (ref) => ref.read(hashServiceProvider).hashAssets(), computation: (ref) => ref.read(hashServiceProvider).hashAssets(),
debugLabel: 'hash-assets', debugLabel: 'hash-assets',
); );
return _hashTask! return task
.whenComplete(() { .whenComplete(() {
onHashingComplete?.call(); onHashingComplete?.call();
_hashTask = null; if (identical(_hashTask, task)) {
_hashTask = null;
}
}) })
.catchError((error) { .catchError((error) {
onHashingError?.call(error.toString()); if (error is! CanceledError) {
_hashTask = null; onHashingError?.call(error.toString());
}
}); });
} }
@ -143,11 +165,11 @@ class BackgroundSyncManager {
onRemoteSyncStart?.call(); onRemoteSyncStart?.call();
_syncTask = runInIsolateGentle( final task = _syncTask = runInIsolateGentle(
computation: (ref) => ref.read(syncStreamServiceProvider).sync(), computation: (ref) => ref.read(syncStreamServiceProvider).sync(),
debugLabel: 'remote-sync', debugLabel: 'remote-sync',
); );
return _syncTask! return task
.then((result) { .then((result) {
final success = result ?? false; final success = result ?? false;
onRemoteSyncComplete?.call(success); onRemoteSyncComplete?.call(success);
@ -155,15 +177,21 @@ class BackgroundSyncManager {
return success; return success;
}) })
.catchError((error) { .catchError((error) {
onRemoteSyncError?.call(error.toString()); if (error is! CanceledError) {
onRemoteSyncError?.call(error.toString());
}
_syncQueued = false; _syncQueued = false;
return 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(() { .whenComplete(() {
_syncTask = null; if (identical(_syncTask, task)) {
if (_syncQueued) { _syncTask = null;
_syncQueued = false; if (_syncQueued) {
unawaited(syncRemote()); _syncQueued = false;
unawaited(syncRemote());
}
} }
}); });
} }
@ -210,13 +238,21 @@ class BackgroundSyncManager {
Future<void> syncLinkedAlbum() { Future<void> syncLinkedAlbum() {
if (_linkedAlbumSyncTask != null) { if (_linkedAlbumSyncTask != null) {
return _linkedAlbumSyncTask!.future; return _linkedAlbumSyncTask!.future.catchError((_) {}, test: (error) => error is CanceledError);
} }
_linkedAlbumSyncTask = runInIsolateGentle(computation: syncLinkedAlbumsIsolated, debugLabel: 'linked-album-sync'); final task = _linkedAlbumSyncTask = runInIsolateGentle(
return _linkedAlbumSyncTask!.whenComplete(() { computation: syncLinkedAlbumsIsolated,
_linkedAlbumSyncTask = null; 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<void> syncCloudIds() { Future<void> syncCloudIds() {

View file

@ -112,6 +112,12 @@ class AppLifeCycleNotifier extends StateNotifier<AppLifeCycleEnum> {
await Future.delayed(const Duration(milliseconds: 500)); await Future.delayed(const Duration(milliseconds: 500));
final backgroundManager = _ref.read(backgroundSyncProvider); 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; final isAlbumLinkedSyncEnable = _ref.read(appConfigProvider).backup.syncAlbums;
try { try {