Démarrage : la lecture du quota ne fait plus attendre le flux

AccountLimits interrogeait player_api.php avant d'ouvrir le flux quand
sa valeur n'était pas en cache (premier flux, puis toutes les 10 min).
Le panneau met jusqu'à 4 s à répondre (mesuré en prod) : autant de
retard au démarrage d'une chaîne ou d'un film.

La valeur connue, même périmée, ou le repli (1) est rendue tout de suite ;
la lecture chez le panneau se fait en arrière-plan, une seule à la fois.
Sans risque : le quota ne sert qu'à choisir quels flux orphelins couper.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

bin/api/streaming_handler.dart

@@ -48,17 +48,31 @@ Future<void> _claimUpstream(PlaylistConfig playlist, String id) async {
   await Future<void>.delayed(const Duration(seconds: 1));
 }
 
+/// Délai sans consommation au-delà duquel un flux est tenu pour orphelin.
+///
+/// Un lecteur HLS redemande sa playlist toutes les 2 à 4 s ; un flux
+/// `turbo.ts` reçoit au moins une rafale toutes les ~7 s même sur un panneau
+/// qui livre par à-coups. 15 s laissent une marge à ces deux rythmes.
+const _viewerGrace = Duration(seconds: 15);
+
 /// Inscrit [process] comme connexion amont de [playlist] tant qu'il vit.
+///
+/// [lastActivity] : dernier signe de vie du spectateur ; la connexion n'est
+/// coupée pour en ouvrir une autre que s'il remonte à plus de
+/// [_viewerGrace].
 void _registerUpstream(
   PlaylistConfig playlist,
   String id,
   Process process, {
   required void Function() release,
+  required DateTime Function() lastActivity,
 }) {
   final token = _upstreamSlots.register(
     id: id,
     account: accountKeyOf(playlist),
     release: release,
+    isActive: () =>
+        DateTime.now().difference(lastActivity()) < _viewerGrace,
   );
   process.exitCode.then((_) => _upstreamSlots.unregister(id, token: token));
 }
@@ -274,6 +288,8 @@ Future<_SessionAttempt> _runSession({
       id,
       session.process,
       release: () => sessionManager.killSession(id),
+      // Touchée à chaque requête de playlist ou de segment.
+      lastActivity: () => session.lastAccess,
     );
   }
   final outcome =
@@ -580,11 +596,16 @@ Handler createLiveStreamHandler(
       return Response(503, body: 'FFmpeg start failed');
     }
 
+    // Dernier octet remis au client : un lecteur en vie en reçoit à chaque
+    // rafale ; un client parti (zap mal détecté derrière le proxy) n'en
+    // reçoit plus, la contre-pression bloquant FFmpeg.
+    var lastDelivered = DateTime.now();
     _registerUpstream(
       playlist,
       upstreamId,
       process,
       release: () => process.kill(ProcessSignal.sigterm),
+      lastActivity: () => lastDelivered,
     );
 
     // Journaliser les erreurs FFmpeg (redactées) sans bloquer le flux.
@@ -602,7 +623,10 @@ Handler createLiveStreamHandler(
     final body = pipeWithBackpressure(
       process.stdout,
       onStop: () => process.kill(ProcessSignal.sigterm),
-    );
+    ).map((chunk) {
+      lastDelivered = DateTime.now();
+      return chunk;
+    });
 
     return Response(
       200,

bin/services/upstream_slots.dart

@@ -15,10 +15,18 @@ import '../models/playlist_config.dart';
 /// le panneau répondait `HTTP 551` ou ne répondait pas (timeout), d'où les
 /// films « indisponibles » et les zaps qui échouent.
 ///
-/// Règle : le dernier flux demandé gagne. Avant d'ouvrir une nouvelle
-/// connexion amont, on coupe les plus anciennes du même compte jusqu'à lui
-/// faire de la place. Les enregistrements n'y sont pas inscrits : ils ne
-/// sont jamais coupés.
+/// Règle : avant d'ouvrir une nouvelle connexion amont, on coupe les
+/// connexions ORPHELINES du même compte (plus personne ne les consomme),
+/// les plus anciennes d'abord, jusqu'à lui faire de la place.
+///
+/// Un flux encore regardé n'est jamais coupé. Une première version coupait
+/// sans distinction (« le dernier gagne ») : deux lecteurs ouverts sur un
+/// compte à une connexion se coupaient alors en boucle, chacun relançant
+/// aussitôt — plus aucune image nulle part (constaté en prod). Le second
+/// lecteur est désormais refusé par le panneau, comme avant, et le premier
+/// continue.
+///
+/// Les enregistrements n'y sont pas inscrits : ils ne sont jamais coupés.
 class UpstreamSlots {
   final DateTime Function() _now;
   final Map<String, _Slot> _slots = {};
@@ -31,13 +39,17 @@ class UpstreamSlots {
   /// Renvoie un jeton à repasser à [unregister] : une session relancée sous
   /// le même identifiant ne doit pas être désinscrite par la fin de
   /// l'ancien processus.
+  ///
+  /// [isActive] : vrai tant qu'un spectateur consomme le flux. Une
+  /// connexion active n'est jamais coupée par [makeRoom].
   int register({
     required String id,
     required String account,
     required void Function() release,
+    bool Function()? isActive,
   }) {
     final token = ++_nextToken;
-    _slots[id] = _Slot(token, account, _now(), release);
+    _slots[id] = _Slot(token, account, _now(), release, isActive);
     return token;
   }
 
@@ -60,14 +72,16 @@ class UpstreamSlots {
     if (max <= 0) return const [];
     final others = _slots.entries
         .where((e) => e.value.account == account && e.key != keep)
-        .toList()
-      ..sort((a, b) => a.value.since.compareTo(b.value.since));
+        .toList();
 
     final excess = others.length - (max - 1);
     if (excess <= 0) return const [];
 
+    final orphans = others.where((e) => !e.value.active).toList()
+      ..sort((a, b) => a.value.since.compareTo(b.value.since));
+
     final freed = <String>[];
-    for (final entry in others.take(excess)) {
+    for (final entry in orphans.take(excess)) {
       _slots.remove(entry.key);
       try {
         entry.value.release();
@@ -83,7 +97,16 @@ class _Slot {
   final String account;
   final DateTime since;
   final void Function() release;
-  _Slot(this.token, this.account, this.since, this.release);
+  final bool Function()? isActive;
+  _Slot(this.token, this.account, this.since, this.release, this.isActive);
+
+  bool get active {
+    try {
+      return isActive?.call() ?? false;
+    } catch (_) {
+      return false;
+    }
+  }
 }
 
 /// Clé de compte : deux playlists sur les mêmes identifiants partagent le
@@ -101,13 +124,21 @@ int? parseMaxConnections(Object? json) {
 }
 
 /// Quota de connexions par compte, lu chez le fournisseur et mis en cache.
+///
+/// Ne fait JAMAIS attendre un flux : `player_api.php` met jusqu'à 4 s à
+/// répondre (mesuré en prod), délai qui s'ajoutait au démarrage du premier
+/// flux puis de chaque flux suivant l'expiration du cache. La valeur connue
+/// (même périmée), sinon [fallback], est rendue immédiatement ; la lecture
+/// chez le panneau se fait en arrière-plan. Sans risque : le quota ne sert
+/// qu'à décider quels flux ORPHELINS couper.
 class AccountLimits {
   final Duration ttl;
   final Future<int?> Function(PlaylistConfig) _fetch;
   final Map<String, ({int max, DateTime at})> _cache = {};
+  final Set<String> _refreshing = {};
 
-  /// Valeur retenue quand le panneau ne répond pas : un seul flux, le cas
-  /// de loin le plus courant chez les fournisseurs Xtream.
+  /// Valeur retenue tant que le panneau n'a pas répondu : un seul flux, le
+  /// cas de loin le plus courant chez les fournisseurs Xtream.
   static const fallback = 1;
 
   AccountLimits({
@@ -118,19 +149,20 @@ class AccountLimits {
   Future<int> maxFor(PlaylistConfig p) async {
     final key = accountKeyOf(p);
     final cached = _cache[key];
-    if (cached != null && DateTime.now().difference(cached.at) < ttl) {
-      return cached.max;
+    if (cached == null || DateTime.now().difference(cached.at) >= ttl) {
+      _refresh(key, p);
     }
-    int? value;
-    try {
-      value = await _fetch(p);
-    } catch (_) {
-      value = null;
-    }
-    // Un échec n'est pas mis en cache longtemps : on réessaiera au prochain
-    // flux plutôt que de rester 10 min sur une valeur devinée.
-    if (value != null) _cache[key] = (max: value, at: DateTime.now());
-    return value ?? fallback;
+    return cached?.max ?? fallback;
+  }
+
+  void _refresh(String key, PlaylistConfig p) {
+    if (!_refreshing.add(key)) return; // une seule lecture à la fois
+    Future<int?>.sync(() => _fetch(p))
+        .then<int?>((v) => v, onError: (Object _) => null)
+        .then((value) {
+      // Un échec n'est pas mis en cache : on réessaiera au prochain flux.
+      if (value != null) _cache[key] = (max: value, at: DateTime.now());
+    }).whenComplete(() => _refreshing.remove(key));
   }
 
   static Future<int?> _fetchFromPanel(PlaylistConfig p) async {

bin/test/upstream_slots_test.dart

@@ -1,4 +1,7 @@
+import 'dart:async';
+
 import 'package:test/test.dart';
+import '../models/playlist_config.dart';
 import '../services/upstream_slots.dart';
 
 void main() {
@@ -53,12 +56,81 @@ void main() {
       expect(slots.countFor('acc'), 1);
     });
 
+    test('un flux encore regardé n\'est jamais coupé (pas de ping-pong entre deux lecteurs)', () {
+      slots.register(
+        id: 'watched',
+        account: 'acc',
+        release: () => released.add('watched'),
+        isActive: () => true,
+      );
+      expect(slots.makeRoom('acc', max: 1, keep: 'new'), isEmpty);
+      expect(released, isEmpty);
+      expect(slots.countFor('acc'), 1);
+    });
+
+    test('seuls les orphelins partent, même plus récents qu\'un flux regardé', () {
+      slots.register(
+        id: 'watched',
+        account: 'acc',
+        release: () => released.add('watched'),
+        isActive: () => true,
+      );
+      add('orphan', 'acc');
+      expect(slots.makeRoom('acc', max: 1, keep: 'new'), ['orphan']);
+      expect(released, ['orphan']);
+    });
+
     test('quota inconnu ou nul : rien n\'est coupé', () {
       add('a', 'acc');
       expect(slots.makeRoom('acc', max: 0, keep: 'new'), isEmpty);
     });
   });
 
+  group('AccountLimits', () {
+    PlaylistConfig playlist() => PlaylistConfig(
+          id: 'p',
+          name: 'n',
+          dns: 'http://panel.test',
+          username: 'u',
+          password: 'x',
+          createdAt: DateTime(2026),
+        );
+
+    test('ne fait jamais attendre un flux : repli immédiat, valeur réelle ensuite', () async {
+      final pending = Completer<int?>();
+      final limits = AccountLimits(fetch: (_) => pending.future);
+
+      // Le panneau répond lentement (4 s mesurées en prod) : le premier flux
+      // part tout de suite sur la valeur de repli.
+      expect(await limits.maxFor(playlist()), AccountLimits.fallback);
+
+      pending.complete(3);
+      await Future<void>.delayed(Duration.zero);
+      expect(await limits.maxFor(playlist()), 3);
+    });
+
+    test('valeur périmée servie pendant le rafraîchissement', () async {
+      var calls = 0;
+      final limits = AccountLimits(
+        ttl: Duration.zero,
+        fetch: (_) async => ++calls == 1 ? 2 : 5,
+      );
+      await limits.maxFor(playlist());
+      await Future<void>.delayed(Duration.zero);
+      // Périmée (ttl nul) : la valeur connue sort sans attendre la nouvelle.
+      expect(await limits.maxFor(playlist()), 2);
+      await Future<void>.delayed(Duration.zero);
+      expect(await limits.maxFor(playlist()), 5);
+    });
+
+    test('panneau en échec : repli, sans exception', () async {
+      final limits = AccountLimits(fetch: (_) async => throw Exception('down'));
+      expect(await limits.maxFor(playlist()), AccountLimits.fallback);
+      await Future<void>.delayed(Duration.zero);
+      expect(await limits.maxFor(playlist()), AccountLimits.fallback);
+    });
+  });
+
   group('parseMaxConnections', () {
     test('lit la valeur texte renvoyée par player_api', () {
       expect(