fix(images): prevent artwork limiter permit leaks

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<http.StreamedResponse> 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<List<int>> _releaseWhenDone(Stream<List<int>> stream, void Function() release) async* {
-  try {
-    await for (final chunk in stream) {
-      yield chunk;
-    }
-  } finally {
+Stream<List<int>> _releaseWhenDone(
+  Stream<List<int>> 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<List<int>> stream) {
+  try {
+    final subscription = stream.listen((_) {}, onError: (_, _) {});
+    unawaited(subscription.cancel().catchError((_) {}));
+  } catch (_) {
+    // The body may already have terminated while the timeout callback ran.
   }
 }
 

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<void>();
+    final body = StreamController<List<int>>(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<void>();
     final body = StreamController<List<int>>(onListen: () => listened.complete());