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.
This commit is contained in:
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<void> 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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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']);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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<http.RequestAbortedException>().having((e) => e.uri, 'uri', Uri.parse('https://example.test/slow')),
|
||||
),
|
||||
);
|
||||
});
|
||||
|
||||
test('closeGracefully can retry after a drain timeout', () async {
|
||||
|
||||
@@ -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<MediaServerHttpException>().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<void>.delayed(Duration.zero);
|
||||
inner.body.add([1, 2, 3]);
|
||||
await Future<void>.delayed(Duration.zero);
|
||||
|
||||
final closing = client.closeGracefully(drainTimeout: const Duration(milliseconds: 100));
|
||||
await expectLater(
|
||||
pending,
|
||||
throwsA(isA<MediaServerHttpException>().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<void>.delayed(Duration.zero);
|
||||
inner.body.add([1, 2, 3]);
|
||||
// Let the partial chunk reach the sink before the abort lands.
|
||||
await Future<void>.delayed(Duration.zero);
|
||||
|
||||
final closing = client.closeGracefully(drainTimeout: const Duration(milliseconds: 100));
|
||||
await expectLater(
|
||||
pending,
|
||||
throwsA(isA<MediaServerHttpException>().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<List<int>> body = StreamController<List<int>>();
|
||||
|
||||
@override
|
||||
Future<http.StreamedResponse> send(http.BaseRequest request) async {
|
||||
return http.StreamedResponse(body.stream, 200);
|
||||
}
|
||||
|
||||
@override
|
||||
void close() {}
|
||||
|
||||
Future<void> dispose() async {
|
||||
await body.close();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user