mirror of
https://github.com/immich-app/immich.git
synced 2026-07-18 05:34:18 +03:00
Compare commits
2 Commits
main
...
fix/failed
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6b2606f30d | ||
|
|
8aa5067f9b |
150
mobile/integration_test/failed_sync_resume_test.dart
Normal file
150
mobile/integration_test/failed_sync_resume_test.dart
Normal 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));
|
||||
});
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user