From e5dc9c3f89f55fe7cd92fb65fef3a37d1196c7d6 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Thu, 20 Aug 2026 20:58:31 +0200 Subject: [PATCH] fix(networking): surface cancelled responses as cancellation and stop crediting a failed Plex hub fetch A response cancelled before or during body consumption completed as a successful empty body, so cancellation-triggered aggregation could overwrite valid state and a cancelled download could commit an empty file. The managed client now delivers an abort error, and a request whose own abort has fired classifies secondary teardown errors as cancelled (timeouts keep their type). A Plex server whose /hubs leg failed was also counted as succeeded when only the optional music leg returned, letting Discover replace good hubs with a music-only result. --- lib/services/data_aggregation_service.dart | 8 ++- lib/utils/managed_http_client.dart | 24 ++++++- lib/utils/media_server_http_client.dart | 12 ++++ .../data_aggregation_bridge_test.dart | 65 ++++++++++++++++- test/utils/http_lifecycle_test.dart | 9 ++- ...edia_server_http_client_shutdown_test.dart | 69 +++++++++++++++++++ 6 files changed, 181 insertions(+), 6 deletions(-) diff --git a/lib/services/data_aggregation_service.dart b/lib/services/data_aggregation_service.dart index 289c73ca8..412a579d4 100644 --- a/lib/services/data_aggregation_service.dart +++ b/lib/services/data_aggregation_service.dart @@ -555,7 +555,13 @@ class DataAggregationService { legSucceededServerIds.add(serverId); } if (music != null) { - if (music.succeeded) legSucceededServerIds.add(serverId); + // The music append is optional: it must not vouch for the server + // when the global leg failed or was cancelled (Plex records that + // in diagnostics and returns [] instead of throwing), or a lone + // music row would replace every cached movie/TV home row. + if (music.succeeded && !globalDiagnostics.failed && !globalDiagnostics.cancelled) { + legSucceededServerIds.add(serverId); + } if (music.failed) legFailedServerIds.add(serverId); if (music.cancelled) legCancelledServerIds.add(serverId); } diff --git a/lib/utils/managed_http_client.dart b/lib/utils/managed_http_client.dart index bd230f6fa..7bc8170d5 100644 --- a/lib/utils/managed_http_client.dart +++ b/lib/utils/managed_http_client.dart @@ -156,6 +156,16 @@ class ManagedHttpClient extends http.BaseClient { var subscribed = false; var cancelledBeforeListen = false; + // Cancellation is not an empty successful response: deliver the abort as + // an error so downstream mapping (MediaServerHttpException.from) reports + // `cancelled` instead of handing consumers a clean empty body. Runs at + // most once — whichever of cancelResponse/onListen gets there first. + void abortOutput() { + if (controller.isClosed) return; + controller.addError(http.RequestAbortedException(tracked.url)); + unawaited(controller.close()); + } + Future cancelResponse() async { if (tracked.isDone) return; tracked.abort(); @@ -163,10 +173,18 @@ class ManagedHttpClient extends http.BaseClient { if (subscribed) { await subscription?.cancel(); } else { - final cancelSubscription = response.stream.listen(null, onError: (_) {}); + // Subscribe-and-cancel releases the inner transport stream nobody is + // reading. Post-abort transport errors are expected here, but not + // silently: the outer consumer gets the abort from abortOutput. + final cancelSubscription = response.stream.listen( + null, + onError: (Object e, StackTrace st) { + appLogger.d('HTTP response release stream error during cancellation', error: e, stackTrace: st); + }, + ); await cancelSubscription.cancel(); } - unawaited(controller.close()); + abortOutput(); _complete(tracked); } @@ -174,7 +192,7 @@ class ManagedHttpClient extends http.BaseClient { sync: true, onListen: () { if (cancelledBeforeListen) { - unawaited(controller.close()); + abortOutput(); return; } subscribed = true; diff --git a/lib/utils/media_server_http_client.dart b/lib/utils/media_server_http_client.dart index 7b34c1a20..b6e124aaa 100644 --- a/lib/utils/media_server_http_client.dart +++ b/lib/utils/media_server_http_client.dart @@ -334,8 +334,20 @@ class MediaServerHttpClient { ); return await consume(streamed, scope); } catch (e) { + // Once this request's abort has fired, any secondary teardown error that + // surfaces first (a file sink failing after the stream died, a socket + // reset) is still a cancellation to the caller. Timeouts abort the + // request themselves before rethrowing and must keep their own type. + final wasAborted = requestAbort.isAborted || (abort?.isAborted ?? false); requestAbort.abort(); await onError?.call(); + if (wasAborted && e is! MediaServerHttpException && e is! TimeoutException) { + throw MediaServerHttpException( + type: MediaServerHttpErrorType.cancelled, + requestUri: uri, + message: 'Request aborted: $e', + ); + } throw MediaServerHttpException.from(e, uri: uri); } finally { _activeAborts.remove(requestAbort); diff --git a/test/services/data_aggregation_bridge_test.dart b/test/services/data_aggregation_bridge_test.dart index a6ed5d4e6..89a142524 100644 --- a/test/services/data_aggregation_bridge_test.dart +++ b/test/services/data_aggregation_bridge_test.dart @@ -1100,7 +1100,7 @@ void main() { final movieLatest = captured.singleWhere( (uri) => uri.path == '/Users/user-1/Items/Latest' && uri.queryParameters['ParentId'] == 'movies', ); - expect(movieLatest.queryParameters['Fields'], 'Overview'); + expect(movieLatest.queryParameters['Fields'], 'Overview,DateCreated'); expect(movieLatest.queryParameters.containsKey('EnableUserData'), isFalse); }); @@ -1310,5 +1310,68 @@ void main() { '/hubs/sections/9', ]); }); + + test('Plex server with a failed global leg is not marked succeeded by the optional music append', () async { + // /hubs/promoted failures are recorded in diagnostics and returned as [] + // (never thrown), so only the success accounting separates "promoted + // hubs are gone" from "promoted hubs are empty". Counting the optional + // music leg as server success let DiscoverProvider replace every cached + // movie/TV home row with the lone music row. + final client = testPlexClient( + config: PlexConfig( + baseUrl: 'https://plex.example.com', + token: 'token', + clientIdentifier: 'client-id', + product: 'Plezy', + version: 'test', + ), + serverId: ServerId('plex-1'), + serverName: 'Plex', + promotedHubKey: '/hubs/promoted', + httpClient: MockClient((req) async { + if (req.url.path == '/library/sections') { + return _json({ + 'MediaContainer': { + 'Directory': [ + {'key': '1', 'type': 'movie', 'title': 'Movies'}, + {'key': '9', 'type': 'artist', 'title': 'Music'}, + ], + }, + }); + } + if (req.url.path == '/hubs/promoted') { + return http.Response('server error', 500); + } + if (req.url.path == '/hubs/sections/9') { + return _json({ + 'MediaContainer': { + 'Hub': [ + { + 'key': '/library/sections/9/recentlyAdded', + 'title': 'Recently Added Music', + 'type': 'album', + 'hubIdentifier': 'music.recent', + 'size': 1, + 'Metadata': [ + {'ratingKey': 'album-1', 'type': 'album', 'title': 'Album', 'librarySectionID': 9}, + ], + }, + ], + }, + }); + } + return http.Response('unexpected request', 500); + }), + ); + addTearDown(client.close); + manager.debugRegisterClientForTesting(client); + + final result = await service.getHubsFromAllServers(useGlobalHubs: true, includePlaybackHubs: false); + + expect(result.succeededServerIds, isEmpty, reason: 'the optional music leg must not vouch for the server'); + expect(result.failedServerIds, {'plex-1'}); + // The music rows that did arrive are still delivered. + expect(result.hubs.map((h) => h.identifier), ['music.recent']); + }); }); } diff --git a/test/utils/http_lifecycle_test.dart b/test/utils/http_lifecycle_test.dart index 5a2594918..709000a6d 100644 --- a/test/utils/http_lifecycle_test.dart +++ b/test/utils/http_lifecycle_test.dart @@ -36,7 +36,14 @@ void main() { expect(inner.closeCount, 1); expect(inner.responseCancelled, isTrue); await expectLater(inner.abortTrigger, completes); - await expectLater(response.stream.toList(), completion(isEmpty)); + // Cancellation is not an empty successful response: a body cancelled + // before anyone listened must surface the abort, not a clean [] close. + await expectLater( + response.stream.toList(), + throwsA( + isA().having((e) => e.uri, 'uri', Uri.parse('https://example.test/slow')), + ), + ); }); test('closeGracefully can retry after a drain timeout', () async { diff --git a/test/utils/media_server_http_client_shutdown_test.dart b/test/utils/media_server_http_client_shutdown_test.dart index 77387f046..c8e834f0e 100644 --- a/test/utils/media_server_http_client_shutdown_test.dart +++ b/test/utils/media_server_http_client_shutdown_test.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:io'; import 'package:flutter_test/flutter_test.dart'; import 'package:http/http.dart' as http; @@ -59,6 +60,56 @@ void main() { throwsA(isA().having((e) => e.type, 'type', MediaServerHttpErrorType.cancelled)), ); }); + + test('shutdown during an already-streaming body surfaces as cancelled, not an empty success', () async { + final inner = _HangingBodyClient(); + final client = MediaServerHttpClient( + client: ManagedHttpClient(inner, debugLabel: 'test'), + baseUrl: 'https://example.test/', + ); + addTearDown(inner.dispose); + + final pending = client.get('library/sections'); + await Future.delayed(Duration.zero); + inner.body.add([1, 2, 3]); + await Future.delayed(Duration.zero); + + final closing = client.closeGracefully(drainTimeout: const Duration(milliseconds: 100)); + await expectLater( + pending, + throwsA(isA().having((e) => e.type, 'type', MediaServerHttpErrorType.cancelled)), + ); + await closing; + }); + + test('downloadFile cancellation fails without committing a renamed file', () async { + final dir = await Directory.systemTemp.createTemp('plezy_download_cancel_test'); + addTearDown(() => dir.delete(recursive: true)); + final filePath = '${dir.path}${Platform.pathSeparator}subtitle.srt'; + + final inner = _HangingBodyClient(); + final client = MediaServerHttpClient( + client: ManagedHttpClient(inner, debugLabel: 'test'), + baseUrl: 'https://example.test/', + ); + addTearDown(inner.dispose); + + final pending = client.downloadFile('media/subtitle.srt', filePath); + await Future.delayed(Duration.zero); + inner.body.add([1, 2, 3]); + // Let the partial chunk reach the sink before the abort lands. + await Future.delayed(Duration.zero); + + final closing = client.closeGracefully(drainTimeout: const Duration(milliseconds: 100)); + await expectLater( + pending, + throwsA(isA().having((e) => e.type, 'type', MediaServerHttpErrorType.cancelled)), + ); + await closing; + + expect(File(filePath).existsSync(), isFalse, reason: 'a cancelled download must not commit the final file'); + expect(File('$filePath.download').existsSync(), isFalse, reason: 'the partial temp file must be cleaned up'); + }); }); } @@ -81,3 +132,21 @@ class _AbortAwareClient extends http.BaseClient { @override void close() {} } + +/// Returns a 200 whose body stays open until the test aborts it, so +/// cancellation lands while (or before) the consumer is reading. +class _HangingBodyClient extends http.BaseClient { + final StreamController> body = StreamController>(); + + @override + Future send(http.BaseRequest request) async { + return http.StreamedResponse(body.stream, 200); + } + + @override + void close() {} + + Future dispose() async { + await body.close(); + } +}