diff --git a/lib/services/image_cache_service.dart b/lib/services/image_cache_service.dart index cd248465..7e02d8c8 100644 --- a/lib/services/image_cache_service.dart +++ b/lib/services/image_cache_service.dart @@ -66,8 +66,13 @@ class PlexImageCacheManager extends ce_cache.DefaultCacheManager { class _SharedHttpClient extends http.BaseClient { final http.Client _inner; final _RequestLimiter _limiter; + final Duration _unclaimedResponseTimeout; - _SharedHttpClient(this._inner, this._limiter); + _SharedHttpClient( + this._inner, + this._limiter, { + this._unclaimedResponseTimeout = const Duration(seconds: 2), + }); @override Future send(http.BaseRequest request) async { @@ -106,7 +111,7 @@ class _SharedHttpClient extends http.BaseClient { } return http.StreamedResponse( - _releaseWhenDone(response.stream, release), + _releaseWhenDone(response.stream, release, claimTimeout: _unclaimedResponseTimeout), response.statusCode, contentLength: response.contentLength, request: response.request, @@ -128,16 +133,55 @@ class _SharedHttpClient extends http.BaseClient { // ignore: unused-code /// Test hook: builds the throttled artwork client with an isolated limiter. @visibleForTesting -http.Client createArtworkHttpClientForTest(http.Client inner, {int maxConcurrent = 6}) => - _SharedHttpClient(inner, _RequestLimiter(maxConcurrent)); +http.Client createArtworkHttpClientForTest( + http.Client inner, { + int maxConcurrent = 6, + Duration unclaimedResponseTimeout = const Duration(seconds: 2), +}) => _SharedHttpClient(inner, _RequestLimiter(maxConcurrent), unclaimedResponseTimeout: unclaimedResponseTimeout); -Stream> _releaseWhenDone(Stream> stream, void Function() release) async* { - try { - await for (final chunk in stream) { - yield chunk; - } - } finally { +Stream> _releaseWhenDone( + Stream> stream, + void Function() release, { + required Duration claimTimeout, +}) { + var claimed = false; + var abandoned = false; + + // A cache request can be cancelled after response headers arrive but before + // CE subscribes to the body (for example when a rail card is disposed). + // An async* wrapper that is never listened to never enters its `finally`, so + // without this guard the permit is lost permanently and artwork wedges once + // every slot has leaked. Give CE ample time to claim the body, then release + // the slot and cancel the orphaned transport request. + final claimTimer = Timer(claimTimeout, () { + if (claimed) return; + abandoned = true; release(); + _cancelUnclaimedBody(stream); + }); + + return (() async* { + if (abandoned) { + throw http.ClientException('Artwork response body was abandoned before it was consumed'); + } + claimed = true; + claimTimer.cancel(); + try { + await for (final chunk in stream) { + yield chunk; + } + } finally { + release(); + } + })(); +} + +void _cancelUnclaimedBody(Stream> stream) { + try { + final subscription = stream.listen((_) {}, onError: (_, _) {}); + unawaited(subscription.cancel().catchError((_) {})); + } catch (_) { + // The body may already have terminated while the timeout callback ran. } } diff --git a/test/services/image_cache_service_test.dart b/test/services/image_cache_service_test.dart index f8f57d03..ed10170d 100644 --- a/test/services/image_cache_service_test.dart +++ b/test/services/image_cache_service_test.dart @@ -60,6 +60,35 @@ void main() { expect(secondDone, isTrue); }); + test('an unclaimed successful body cannot permanently leak a permit', () async { + final client = createArtworkHttpClientForTest( + mixedClient(), + maxConcurrent: 1, + unclaimedResponseTimeout: const Duration(milliseconds: 20), + ); + + // Mirrors an image widget being disposed after headers arrive but before + // the cache manager starts listening to the response body. + await client.send(get('/abandoned')).timeout(timeout); + + final next = await client.send(get('/poster')).timeout(timeout); + expect(await next.stream.bytesToString().timeout(timeout), 'poster-bytes'); + }); + + test('an unclaimed body is cancelled so the transport can reclaim its connection', () async { + final cancelled = Completer(); + final body = StreamController>(onCancel: () => cancelled.complete()); + final client = createArtworkHttpClientForTest( + MockClient.streaming((request, _) async => http.StreamedResponse(body.stream, 200)), + maxConcurrent: 1, + unclaimedResponseTimeout: const Duration(milliseconds: 20), + ); + + await client.send(get('/abandoned')).timeout(timeout); + await cancelled.future.timeout(timeout); + await body.close(); + }); + test('error bodies are drained so the transport can reclaim the connection', () async { final listened = Completer(); final body = StreamController>(onListen: () => listened.complete());