Compare commits

...

2 Commits

Author SHA1 Message Date
Santo Shakil
6b2606f30d share the task lists between the cancel paths 2026-07-15 01:45:42 +06:00
Santo Shakil
8aa5067f9b fix(mobile): don't let a frozen sync block syncing on resume 2026-07-13 14:27:08 +06:00
4 changed files with 248 additions and 41 deletions

View File

@@ -0,0 +1,150 @@
import 'dart:async';
import 'package:flutter_test/flutter_test.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:immich_mobile/domain/utils/background_sync.dart';
import 'package:immich_mobile/main.dart' as app;
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/infrastructure/repositories/db.repository.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.
void main() {
final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized();
binding.framePolicy = LiveTestWidgetsFlutterBindingFramePolicy.fullyLive;
late Drift drift;
late FakeImmichServer server;
setUpAll(() async {
await app.initApp();
(drift, _) = await Bootstrap.initDomain();
});
setUp(() async {
await workerManagerPatch.init(dynamicSpawning: true);
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, leaving a clean pool for the next test.
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',
);
});
// End-to-end resume path. Runs last: handleAppResume is fire-and-forget and its
// frozen syncs outlive the test, so running it before another test would leave the
// worker pool warm and starve the next test's isolates.
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 this
// drift is shared across tests via setUpAll. The container only holds the shared
// drift and a fire-and-forget resume; the frozen isolates + server are drained by
// tearDown. Disposing here would close the shared DB and break other tests.
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();
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

@@ -10,6 +10,8 @@ class FakeImmichServer {
final (int, int, int) version;
final Completer<SyncStream> _streamOpened = Completer<SyncStream>();
final List<SyncStream> _streamOpens = [];
final Map<int, Completer<SyncStream>> _openWaiters = {};
int ackRequests = 0;
@@ -18,6 +20,17 @@ class FakeImmichServer {
/// Resolves when the sync isolate opens `POST /sync/stream`.
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 {
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<void> _respondJson(HttpRequest request, Object body) async {
@@ -83,8 +100,8 @@ class FakeImmichServer {
}
Future<void> close() async {
if (_streamOpened.isCompleted) {
await (await _streamOpened.future).close();
for (final stream in _streamOpens) {
await stream.close();
}
await _server.close(force: true);
}

View File

@@ -49,15 +49,37 @@ 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<Cancelable?> get _resumeSyncTasks => [_syncTask, _deviceAlbumSyncTask, _hashTask, _linkedAlbumSyncTask];
List<Cancelable?> get _allTasks => [_syncWebsocketTask, _cloudIdSyncTask, ..._resumeSyncTasks];
Future<void> cancel() async {
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<void> cancelResumeSyncs() async {
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 = [
for (final task in tasks)
if (task != null) task.future,
@@ -65,13 +87,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 {
@@ -82,14 +97,14 @@ class BackgroundSyncManager {
// No need to cancel the task, as it can also be run when the user logs out
Future<void> 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',
@@ -99,37 +114,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<void> 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());
}
});
}
@@ -140,23 +161,28 @@ 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);
return success;
})
.catchError((error) {
onRemoteSyncError?.call(error.toString());
_syncTask = null;
if (error is! CanceledError) {
onRemoteSyncError?.call(error.toString());
}
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 (identical(_syncTask, task)) {
_syncTask = null;
}
});
}
@@ -202,13 +228,21 @@ class BackgroundSyncManager {
Future<void> 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<void> syncCloudIds() {

View File

@@ -108,6 +108,12 @@ class AppLifeCycleNotifier extends StateNotifier<AppLifeCycleEnum> {
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 {