Zap : ouvrir les chaînes par le HLS du panneau, ~4 s gagnées

Mesuré en prod avec la sonde admin : le panneau met 4,2 à 4,4 s à
répondre la redirection d'un .ts, contre 0,2 à 1,3 s pour un .m3u8 dont
le premier segment arrive ensuite en 50 ms.

turbo.ts et les sessions HLS live lisent désormais le .m3u8 du panneau
(-live_start_index -2 : deux segments d'avance d'un coup), avec repli
automatique sur le .ts si le HLS ne livre rien — dans la même réponse
pour turbo.ts, le lecteur ne voit qu'un démarrage plus long. Pas de
reconnect_at_eof en HLS : chaque segment finit par un EOF (leçon de
xtremobile). Le navigateur reçoit toujours le même flux MPEG-TS.

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

bin/api/streaming_handler.dart

@@ -9,7 +9,6 @@ import '../services/gpu_fallback.dart';
 import '../services/upstream_slots.dart';
 import '../utils/log_redactor.dart';
 import '../utils/media_probe.dart';
-import '../utils/stream_pipe.dart';
 import 'recording_playlist.dart';
 
 /// Directory for temporary HLS segments
@@ -441,6 +440,46 @@ Stream<List<int>> _resilientLiveBody(
   }
 }
 
+/// URL d'une chaîne chez le panneau, en HLS (`.m3u8`) ou en flux continu
+/// (`.ts`).
+String _liveSourceUrl(PlaylistConfig p, String streamId, {required bool hls}) =>
+    '${p.dns}/live/${p.username}/${p.password}/$streamId.${hls ? 'm3u8' : 'ts'}';
+
+/// Arguments d'entrée FFmpeg pour une chaîne du panneau.
+///
+/// POURQUOI le HLS d'abord : mesuré en prod (sonde admin), le panneau met
+/// 4,2 à 4,4 s à répondre la redirection d'un `.ts`, contre 0,2 à 1,3 s pour
+/// un `.m3u8` dont le premier segment arrive ensuite en 50 ms — un zap ~4×
+/// plus rapide. La playlist du panneau garde en plus ~60 s de segments :
+/// `-live_start_index -2` en récupère deux d'un coup, ce qui donne au
+/// lecteur une avance immédiate pour absorber les à-coups.
+///
+/// Pas de `-reconnect_at_eof` en HLS : chaque segment se termine par une
+/// fin de fichier, l'option provoquerait une reconnexion à chaque segment
+/// (leçon tirée de xtremobile, où elle causait gels et sauts).
+List<String> liveInputArgs(String url, {required bool hls}) => [
+      '-headers', 'User-Agent: VLC/3.0.18 LibVLC/3.0.18\r\n',
+      if (hls) ...[
+        '-reconnect', '1',
+        '-reconnect_on_network_error', '1',
+        '-reconnect_delay_max', '5',
+        '-live_start_index', '-2',
+      ] else ...[
+        '-reconnect', '1', '-reconnect_streamed', '1',
+        '-reconnect_at_eof', '1',
+        '-reconnect_delay_max', '10',
+      ],
+      // Abort reads stuck for 30s so a stalled upstream triggers the
+      // reconnect logic instead of wedging the process forever.
+      '-rw_timeout', '30000000',
+      // Démarrage rapide : sans borne, FFmpeg peut passer plusieurs
+      // secondes à sonder le flux avant d'écrire la première sortie.
+      '-fflags', 'nobuffer',
+      '-probesize', '1000000',
+      '-analyzeduration', '1000000',
+      '-i', url,
+    ];
+
 Handler createLiveStreamHandler(
   Future<PlaylistConfig?> Function(Request) getPlaylist, {
   bool Function()? isGpuEnabled,
@@ -458,18 +497,14 @@ Handler createLiveStreamHandler(
     final playlist = await getPlaylist(request);
     if (playlist == null) return Response.forbidden('No playlist');
 
-    final targetUrl =
-        '${playlist.dns}/live/${playlist.username}/${playlist.password}/$streamId.ts';
     final useNvidiaGpu = isGpuEnabled?.call() ?? _isNvidiaGpuEnabled();
     final sessionId = 'live_${streamId}_$quality';
 
     if (!sessionManager.contains(sessionId)) {
-      print(
-        '[Live HLS] Starting $sessionId: ${LogRedactor.redactUrl(targetUrl)}',
-      );
+      print('[Live HLS] Starting $sessionId');
     }
 
-    final result = await _runWithGpuFallback(
+    Future<_SessionAttempt> run({required bool hls}) => _runWithGpuFallback(
       id: sessionId,
       isLive: true,
       upstream: playlist,
@@ -477,19 +512,10 @@ Handler createLiveStreamHandler(
       buildArgs: (gpu) => [
         '-hide_banner', '-loglevel', 'warning',
         if (gpu) ...['-hwaccel', 'cuda'],
-        '-headers', 'User-Agent: VLC/3.0.18 LibVLC/3.0.18\r\n',
-        '-reconnect', '1', '-reconnect_streamed', '1',
-        '-reconnect_at_eof', '1',
-        '-reconnect_delay_max', '10',
-        // Abort reads stuck for 30s so a stalled upstream triggers the
-        // reconnect logic instead of wedging the process forever.
-        '-rw_timeout', '30000000',
-        // Démarrage rapide : sans borne, FFmpeg peut passer plusieurs
-        // secondes à sonder le flux avant d'écrire le premier segment.
-        '-fflags', 'nobuffer',
-        '-probesize', '1000000',
-        '-analyzeduration', '1000000',
-        '-i', targetUrl,
+        ...liveInputArgs(
+          _liveSourceUrl(playlist, streamId, hls: hls),
+          hls: hls,
+        ),
         ..._liveVideoArgs(quality, gpu),
         ..._audioArgs(withFilters: false),
         // HLS sliding window: 10 x 2s segments (lower live latency than the
@@ -507,6 +533,15 @@ Handler createLiveStreamHandler(
       ],
     );
 
+    var result = await run(hls: true);
+    if (!result.ready) {
+      // Panneau sans HLS pour cette chaîne (ou HLS en panne) : repli sur le
+      // flux continu, plus lent à ouvrir mais universel.
+      print('[Live HLS] $sessionId : HLS du panneau indisponible — repli .ts');
+      sessionManager.killSession(sessionId);
+      result = await run(hls: false);
+    }
+
     final session = result.session;
     if (!result.ready || session == null) {
       sessionManager.killSession(sessionId);
@@ -554,79 +589,93 @@ Handler createLiveStreamHandler(
     final playlist = await getPlaylist(request);
     if (playlist == null) return Response.forbidden('No playlist');
 
-    final targetUrl =
-        '${playlist.dns}/live/${playlist.username}/${playlist.password}/$streamId.ts';
-    print(
-      '[Live Turbo] $streamId: ${LogRedactor.redactUrl(targetUrl)}',
-    );
+    print('[Live Turbo] $streamId');
 
     // Un identifiant par requête : deux onglets sur la même chaîne sont
     // deux connexions distinctes chez le fournisseur.
     final upstreamId = 'turbo_${streamId}_${++_turboCounter}';
     await _claimUpstream(playlist, upstreamId);
 
-    final Process process;
-    try {
-      process = await Process.start(_getFFmpegPath(), [
-        '-hide_banner', '-loglevel', 'warning',
-        '-headers', 'User-Agent: VLC/3.0.18 LibVLC/3.0.18\r\n',
-        '-reconnect', '1', '-reconnect_streamed', '1',
-        '-reconnect_at_eof', '1',
-        '-reconnect_delay_max', '10',
-        '-rw_timeout', '30000000',
-        // Démarrage rapide : ne pas bufferiser l'analyse, sonde réduite.
-        '-fflags', 'nobuffer',
-        '-flags', 'low_delay',
-        '-probesize', '1000000',
-        '-analyzeduration', '1000000',
-        '-i', targetUrl,
-        '-c:v', 'copy',
-        '-c:a', 'aac', '-b:a', '160k', '-ac', '2', '-ar', '48000',
-        '-af', 'aresample=async=1',
-        // Pas de délai de mux : les paquets partent dès qu'ils existent.
-        '-muxdelay', '0', '-muxpreload', '0',
-        '-f', 'mpegts', 'pipe:1',
-      ]);
-    } on ProcessException catch (e) {
-      // Serveur saturé (plus de processus ou de mémoire) : un 503 que le
-      // lecteur sait traiter, plutôt qu'une exception qui remonte en 500.
-      // `ProcessException` liste les arguments, donc l'URL `-i` du panneau.
-      print('[Live Turbo] $streamId : FFmpeg n\'a pas démarré '
-          '(${LogRedactor.redactUrl('$e')})');
-      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.
-    process.stderr.transform(const SystemEncoding().decoder).listen((line) {
-      final trimmed = line.trim();
-      if (trimmed.isNotEmpty) {
-        print('[Live Turbo] $streamId ffmpeg: '
-            '${LogRedactor.redactUrl(trimmed)}');
+    Future<Process?> start({required bool hls}) async {
+      final Process process;
+      try {
+        process = await Process.start(_getFFmpegPath(), [
+          '-hide_banner', '-loglevel', 'warning',
+          ...liveInputArgs(
+            _liveSourceUrl(playlist, streamId, hls: hls),
+            hls: hls,
+          ),
+          '-flags', 'low_delay',
+          '-c:v', 'copy',
+          '-c:a', 'aac', '-b:a', '160k', '-ac', '2', '-ar', '48000',
+          '-af', 'aresample=async=1',
+          // Pas de délai de mux : les paquets partent dès qu'ils existent.
+          '-muxdelay', '0', '-muxpreload', '0',
+          '-f', 'mpegts', 'pipe:1',
+        ]);
+      } on ProcessException catch (e) {
+        // `ProcessException` liste les arguments, donc l'URL `-i` du panneau.
+        print('[Live Turbo] $streamId : FFmpeg n\'a pas démarré '
+            '(${LogRedactor.redactUrl('$e')})');
+        return null;
       }
-    });
+      _registerUpstream(
+        playlist,
+        upstreamId,
+        process,
+        release: () => process.kill(ProcessSignal.sigterm),
+        lastActivity: () => lastDelivered,
+      );
+      // Journaliser les erreurs FFmpeg (redactées) sans bloquer le flux.
+      process.stderr.transform(const SystemEncoding().decoder).listen((line) {
+        final trimmed = line.trim();
+        if (trimmed.isNotEmpty) {
+          print('[Live Turbo] $streamId ffmpeg: '
+              '${LogRedactor.redactUrl(trimmed)}');
+        }
+      });
+      return process;
+    }
 
-    // Relayer stdout vers le client en gardant la contre-pression, et tuer
-    // FFmpeg dès que le client zappe ou ferme l'onglet (sinon les processus
-    // s'accumulent à chaque zap).
-    final body = pipeWithBackpressure(
-      process.stdout,
-      onStop: () => process.kill(ProcessSignal.sigterm),
-    ).map((chunk) {
-      lastDelivered = DateTime.now();
-      return chunk;
-    });
+    final first = await start(hls: true);
+    // Serveur saturé (plus de processus ou de mémoire) : un 503 que le
+    // lecteur sait traiter, plutôt qu'une exception qui remonte en 500.
+    if (first == null) return Response(503, body: 'FFmpeg start failed');
+
+    // Relais vers le client. Un générateur `async*` garde la contre-pression
+    // (`yield` attend tant que le client ne lit pas) et son `finally` tue
+    // FFmpeg dès que le client zappe ou ferme l'onglet.
+    //
+    // Si le HLS du panneau n'a livré aucun octet (format refusé pour cette
+    // chaîne, playlist vide), on rebascule sur le `.ts` dans la MÊME
+    // réponse : le lecteur ne voit qu'un démarrage un peu plus long.
+    Stream<List<int>> relay() async* {
+      Process? process = first;
+      var hls = true;
+      try {
+        while (process != null) {
+          var delivered = false;
+          await for (final chunk in process.stdout) {
+            delivered = true;
+            lastDelivered = DateTime.now();
+            yield chunk;
+          }
+          if (delivered || !hls) break;
+          print('[Live Turbo] $streamId : HLS du panneau muet — repli .ts');
+          hls = false;
+          process = await start(hls: false);
+        }
+      } finally {
+        process?.kill(ProcessSignal.sigterm);
+      }
+    }
+
+    final body = relay();
 
     return Response(
       200,

bin/api/upstream_probe.dart

@@ -0,0 +1,133 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:http/http.dart' as http;
+
+import '../models/playlist_config.dart';
+
+/// Diagnostic admin : chronomètre l'ouverture d'une chaîne chez le
+/// fournisseur, en `.ts` et en `.m3u8`.
+///
+/// POURQUOI : ~4,3 s des ~4,6 s d'un zap se passent avant le premier octet
+/// du panneau (mesuré en prod). Le panneau autorise aussi le HLS ; reste à
+/// savoir lequel des deux formats démarre le plus vite, et où part le temps
+/// (redirection, en-têtes, premier octet). Les identifiants n'apparaissent
+/// jamais dans la réponse : seuls l'hôte et les durées sont rendus.
+Future<Map<String, Object?>> probeUpstream(
+  PlaylistConfig playlist,
+  String streamId, {
+  http.Client? client,
+}) async {
+  final c = client ?? http.Client();
+  try {
+    final base = '${playlist.dns}/live/${playlist.username}/'
+        '${playlist.password}/$streamId';
+    final ts = await _probeStream(c, Uri.parse('$base.ts'));
+    // Laisser le panneau libérer la connexion (comptes à une connexion).
+    await Future<void>.delayed(const Duration(seconds: 2));
+    final hls = await _probeHls(c, Uri.parse('$base.m3u8'));
+    return {'stream': streamId, 'ts': ts, 'm3u8': hls};
+  } finally {
+    if (client == null) c.close();
+  }
+}
+
+/// Suit les redirections à la main pour chronométrer chaque saut.
+Future<({http.StreamedResponse? response, List<Map<String, Object?>> hops, Uri url})>
+    _open(http.Client c, Uri url, Stopwatch sw) async {
+  final hops = <Map<String, Object?>>[];
+  var current = url;
+  for (var i = 0; i < 5; i++) {
+    final request = http.Request('GET', current)
+      ..followRedirects = false
+      ..headers['User-Agent'] = 'VLC/3.0.18 LibVLC/3.0.18';
+    final response =
+        await c.send(request).timeout(const Duration(seconds: 20));
+    hops.add({
+      'host': current.host,
+      'status': response.statusCode,
+      'headersMs': sw.elapsedMilliseconds,
+    });
+    final location = response.headers['location'];
+    if (const {301, 302, 303, 307, 308}.contains(response.statusCode) &&
+        location != null) {
+      await response.stream.drain<void>().catchError((_) {});
+      current = current.resolve(location);
+      continue;
+    }
+    return (response: response, hops: hops, url: current);
+  }
+  return (response: null, hops: hops, url: current);
+}
+
+Future<Map<String, Object?>> _probeStream(http.Client c, Uri url) async {
+  final sw = Stopwatch()..start();
+  try {
+    final opened = await _open(c, url, sw);
+    final response = opened.response;
+    if (response == null || response.statusCode != 200) {
+      return {'hops': opened.hops, 'error': 'status ${response?.statusCode}'};
+    }
+    var bytes = 0;
+    int? firstByteMs;
+    final sub = response.stream.listen(null);
+    final done = Completer<void>();
+    sub.onData((chunk) {
+      firstByteMs ??= sw.elapsedMilliseconds;
+      bytes += chunk.length;
+      // 1 Mo suffit à voir le premier envoi (préchargement du panneau).
+      if (bytes > 1000000 && !done.isCompleted) done.complete();
+    });
+    sub.onDone(() {
+      if (!done.isCompleted) done.complete();
+    });
+    await done.future.timeout(const Duration(seconds: 15), onTimeout: () {});
+    await sub.cancel();
+    return {
+      'hops': opened.hops,
+      'firstByteMs': firstByteMs,
+      'firstMegabyteMs': bytes > 1000000 ? sw.elapsedMilliseconds : null,
+    };
+  } catch (e) {
+    return {'error': e.runtimeType.toString(), 'afterMs': sw.elapsedMilliseconds};
+  }
+}
+
+Future<Map<String, Object?>> _probeHls(http.Client c, Uri url) async {
+  final sw = Stopwatch()..start();
+  try {
+    final opened = await _open(c, url, sw);
+    final response = opened.response;
+    if (response == null || response.statusCode != 200) {
+      return {'hops': opened.hops, 'error': 'status ${response?.statusCode}'};
+    }
+    final body = await response.stream.bytesToString();
+    final playlistMs = sw.elapsedMilliseconds;
+    final lines = const LineSplitter().convert(body);
+    final segments =
+        lines.where((l) => l.isNotEmpty && !l.startsWith('#')).toList();
+    final target = lines
+        .firstWhere((l) => l.startsWith('#EXT-X-TARGETDURATION'),
+            orElse: () => '')
+        .split(':')
+        .last;
+    if (segments.isEmpty) {
+      return {'hops': opened.hops, 'playlistMs': playlistMs, 'segments': 0};
+    }
+    // Un lecteur démarre près du direct : chronométrer l'avant-dernier.
+    final pick = segments.length >= 2
+        ? segments[segments.length - 2]
+        : segments.last;
+    final segment = await _probeStream(c, opened.url.resolve(pick));
+    return {
+      'hops': opened.hops,
+      'playlistMs': playlistMs,
+      'segments': segments.length,
+      'targetDuration': target,
+      'segment': segment,
+      'segmentStartedAtMs': playlistMs,
+    };
+  } catch (e) {
+    return {'error': e.runtimeType.toString(), 'afterMs': sw.elapsedMilliseconds};
+  }
+}

bin/server.dart

@@ -28,6 +28,7 @@ import 'middleware/security_middleware.dart';
 import 'services/cleanup_service.dart';
 import 'services/recording_scheduler.dart';
 import 'utils/asset_versioning.dart';
+import 'api/upstream_probe.dart';
 
 void main(List<String> args) async {
   // Parse command line arguments
@@ -269,6 +270,29 @@ void main(List<String> args) async {
         );
       });
 
+      // GET /api/admin/upstream-probe?stream=<id> : chronomètre l'ouverture
+      // d'une chaîne chez le fournisseur en .ts et en .m3u8 (diagnostic du
+      // délai de zap ; aucun identifiant dans la réponse).
+      router.get('/upstream-probe', (Request req) async {
+        final user = req.context['user'] as User?;
+        if (user == null || !user.isAdmin) {
+          return Response.forbidden(
+            jsonEncode({'error': 'Admin access required'}),
+          );
+        }
+        final streamId = req.url.queryParameters['stream'] ?? '';
+        if (!RegExp(r'^[0-9]+$').hasMatch(streamId)) {
+          return Response.badRequest(body: 'stream=<id numérique> requis');
+        }
+        final playlist = await getPlaylist(req);
+        if (playlist == null) return Response.forbidden('No playlist');
+        final result = await probeUpstream(playlist, streamId);
+        return Response.ok(
+          jsonEncode(result),
+          headers: {'content-type': 'application/json'},
+        );
+      });
+
       return router(request);
     }),
   );

bin/test/live_input_args_test.dart

@@ -0,0 +1,35 @@
+import 'package:test/test.dart';
+import '../api/streaming_handler.dart';
+
+void main() {
+  group('liveInputArgs', () {
+    test('HLS : jamais de reconnect_at_eof (chaque segment finit par un EOF)',
+        () {
+      final args = liveInputArgs('http://p/live/u/p/1.m3u8', hls: true);
+      expect(args, isNot(contains('-reconnect_at_eof')));
+      expect(args, isNot(contains('-reconnect_streamed')));
+    });
+
+    test('HLS : démarre deux segments avant le direct pour avoir de l\'avance',
+        () {
+      final args = liveInputArgs('http://p/live/u/p/1.m3u8', hls: true);
+      final i = args.indexOf('-live_start_index');
+      expect(i, isNot(-1));
+      expect(args[i + 1], '-2');
+      // Option de démuxeur : doit précéder -i.
+      expect(i, lessThan(args.indexOf('-i')));
+    });
+
+    test('.ts : garde la reconnexion en fin de flux (coupures du panneau)', () {
+      final args = liveInputArgs('http://p/live/u/p/1.ts', hls: false);
+      expect(args, contains('-reconnect_at_eof'));
+      expect(args, isNot(contains('-live_start_index')));
+    });
+
+    test('l\'URL est la dernière entrée, juste après -i', () {
+      final args = liveInputArgs('http://p/x.m3u8', hls: true);
+      expect(args.last, 'http://p/x.m3u8');
+      expect(args[args.length - 2], '-i');
+    });
+  });
+}

bin/test/upstream_probe_test.dart

@@ -0,0 +1,62 @@
+import 'dart:convert';
+
+import 'package:http/http.dart' as http;
+import 'package:http/testing.dart';
+import 'package:test/test.dart';
+
+import '../api/upstream_probe.dart';
+import '../models/playlist_config.dart';
+
+void main() {
+  final playlist = PlaylistConfig(
+    id: 'p',
+    name: 'n',
+    dns: 'http://panel.test',
+    username: 'secretuser',
+    password: 'secretpass',
+    createdAt: DateTime(2026),
+  );
+
+  // Panneau simulé : redirige vers un serveur de diffusion avec jeton.
+  final client = MockClient.streaming((request, _) async {
+    final path = request.url.path;
+    if (request.url.host == 'panel.test') {
+      return http.StreamedResponse(
+        const Stream.empty(),
+        302,
+        headers: {
+          'location': 'http://lb.test$path?token=secrettoken',
+        },
+      );
+    }
+    if (path.endsWith('.m3u8')) {
+      return http.StreamedResponse(
+        Stream.value(utf8.encode(
+          '#EXTM3U\n#EXT-X-TARGETDURATION:6\n#EXTINF:6,\nseg1.ts\n'
+          '#EXTINF:6,\nseg2.ts\n',
+        )),
+        200,
+      );
+    }
+    return http.StreamedResponse(Stream.value(List.filled(2000, 0x47)), 200);
+  });
+
+  test('chronomètre .ts et .m3u8 en suivant les redirections', () async {
+    final result = await probeUpstream(playlist, '123', client: client);
+    final ts = result['ts'] as Map;
+    expect((ts['hops'] as List).map((h) => (h as Map)['status']), [302, 200]);
+    expect(ts['firstByteMs'], isNotNull);
+
+    final hls = result['m3u8'] as Map;
+    expect(hls['segments'], 2);
+    expect(hls['targetDuration'], '6');
+    expect((hls['segment'] as Map)['firstByteMs'], isNotNull);
+  });
+
+  test('aucun identifiant ni jeton dans la réponse', () async {
+    final json = jsonEncode(await probeUpstream(playlist, '123', client: client));
+    expect(json, isNot(contains('secretuser')));
+    expect(json, isNot(contains('secretpass')));
+    expect(json, isNot(contains('secrettoken')));
+  });
+}