diff --git a/bin/api/streaming_handler.dart b/bin/api/streaming_handler.dart index 36b115d..691e0ff 100644 --- a/bin/api/streaming_handler.dart +++ b/bin/api/streaming_handler.dart @@ -6,6 +6,7 @@ import '../database/database.dart'; import '../models/playlist_config.dart'; import '../services/ffmpeg_session_manager.dart'; 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'; @@ -21,6 +22,47 @@ final FfmpegSessionManager sessionManager = FfmpegSessionManager(_hlsTempDir); /// Pannes GPU récentes, partagées par toutes les routes de transcodage. final GpuHealth _gpuHealth = GpuHealth(); +/// Connexions de lecture ouvertes chez le fournisseur, par compte. +final UpstreamSlots _upstreamSlots = UpstreamSlots(); + +/// Numérote les flux `turbo.ts` (un processus FFmpeg par requête). +int _turboCounter = 0; + +/// Quota `max_connections` de chaque compte Xtream. +final AccountLimits _accountLimits = AccountLimits(); + +/// Fait de la place chez le fournisseur avant d'ouvrir la connexion [id]. +/// +/// Voir [UpstreamSlots] : sur un compte à une connexion, le flux précédent +/// (souvent un FFmpeg orphelin que plus personne ne regarde) bloquait le +/// suivant. +Future _claimUpstream(PlaylistConfig playlist, String id) async { + final max = await _accountLimits.maxFor(playlist); + final freed = + _upstreamSlots.makeRoom(accountKeyOf(playlist), max: max, keep: id); + if (freed.isEmpty) return; + print('[Upstream] $id : ${freed.join(', ')} coupé(s) ' + '(quota fournisseur : $max connexion(s))'); + // Laisser le panneau constater la fermeture : rouvrir aussitôt se fait + // refuser sur les comptes à une seule connexion. + await Future.delayed(const Duration(seconds: 1)); +} + +/// Inscrit [process] comme connexion amont de [playlist] tant qu'il vit. +void _registerUpstream( + PlaylistConfig playlist, + String id, + Process process, { + required void Function() release, +}) { + final token = _upstreamSlots.register( + id: id, + account: accountKeyOf(playlist), + release: release, + ); + process.exitCode.then((_) => _upstreamSlots.unregister(id, token: token)); +} + /// Helper to resolve FFmpeg path (SYSTEM PATH vs Portable) String _getFFmpegPath() { if (Platform.isWindows) { @@ -192,12 +234,19 @@ typedef _SessionAttempt = ({FfmpegSession? session, bool ready, String? error}); /// Un `Process.start` qui échoue (serveur à court de processus ou de /// mémoire) devient un échec ordinaire au lieu d'une exception non /// rattrapée qui remontait en 500. +/// +/// [upstream] : playlist dont la session ouvre une connexion chez le +/// fournisseur (direct, film). Null pour un enregistrement, lu sur disque. Future<_SessionAttempt> _runSession({ required String id, required bool isLive, required List Function(Directory dir) argsBuilder, int minSegments = 1, + PlaylistConfig? upstream, }) async { + final starting = sessionManager.needsStart(id, isLive: isLive); + if (upstream != null && starting) await _claimUpstream(upstream, id); + final FfmpegSession session; try { session = await sessionManager.getOrStart( @@ -209,6 +258,14 @@ Future<_SessionAttempt> _runSession({ } on ProcessException catch (e) { return (session: null, ready: false, error: 'FFmpeg start failed: $e'); } + if (upstream != null && starting) { + _registerUpstream( + upstream, + id, + session.process, + release: () => sessionManager.killSession(id), + ); + } final outcome = await sessionManager.waitForPlaylist(session, minSegments: minSegments); return (session: session, ready: outcome.ready, error: outcome.error); @@ -226,6 +283,7 @@ Future<_SessionAttempt> _runWithGpuFallback({ required bool wantGpu, required List Function(bool gpu) buildArgs, int minSegments = 1, + PlaylistConfig? upstream, }) async { final gpu = wantGpu && _gpuHealth.available; var attempt = await _runSession( @@ -233,6 +291,7 @@ Future<_SessionAttempt> _runWithGpuFallback({ isLive: isLive, argsBuilder: (_) => buildArgs(gpu), minSegments: minSegments, + upstream: upstream, ); if (!attempt.ready && gpu && isGpuFailure(attempt.error)) { @@ -245,6 +304,7 @@ Future<_SessionAttempt> _runWithGpuFallback({ isLive: isLive, argsBuilder: (_) => buildArgs(false), minSegments: minSegments, + upstream: upstream, ); } return attempt; @@ -386,6 +446,7 @@ Handler createLiveStreamHandler( final result = await _runWithGpuFallback( id: sessionId, isLive: true, + upstream: playlist, wantGpu: useNvidiaGpu && quality != 'source', buildArgs: (gpu) => [ '-hide_banner', '-loglevel', 'warning', @@ -473,6 +534,11 @@ Handler createLiveStreamHandler( '[Live Turbo] $streamId: ${LogRedactor.redactUrl(targetUrl)}', ); + // 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(), [ @@ -504,6 +570,13 @@ Handler createLiveStreamHandler( return Response(503, body: 'FFmpeg start failed'); } + _registerUpstream( + playlist, + upstreamId, + process, + release: () => process.kill(ProcessSignal.sigterm), + ); + // Journaliser les erreurs FFmpeg (redactées) sans bloquer le flux. process.stderr.transform(const SystemEncoding().decoder).listen((line) { final trimmed = line.trim(); @@ -621,6 +694,7 @@ Handler createVodStreamHandler( final result = await _runWithGpuFallback( id: sessionId, isLive: false, + upstream: playlist, wantGpu: useNvidiaGpu && quality != 'source', buildArgs: (gpu) => [ '-hide_banner', '-loglevel', 'warning', diff --git a/bin/services/ffmpeg_session_manager.dart b/bin/services/ffmpeg_session_manager.dart index 5e7f3a7..3cbde29 100644 --- a/bin/services/ffmpeg_session_manager.dart +++ b/bin/services/ffmpeg_session_manager.dart @@ -74,6 +74,17 @@ class FfmpegSessionManager { bool contains(String id) => _sessions.containsKey(id); + /// Vrai si [getOrStart] lancerait un nouveau processus FFmpeg pour [id] — + /// donc une nouvelle connexion au fournisseur. + bool needsStart(String id, {required bool isLive}) { + final existing = _sessions[id]; + if (existing == null) return true; + if (!existing.exited) return false; + // A VOD/recording transcode that finished cleanly is still fully + // playable from its segments — reuse it instead of re-transcoding. + return isLive || existing.exitCode != 0 || !_playlistComplete(existing); + } + /// Returns the existing healthy session or starts a new FFmpeg process. /// /// [argsBuilder] receives the session working directory and returns the @@ -86,17 +97,11 @@ class FfmpegSessionManager { required List Function(Directory dir) argsBuilder, }) async { final existing = _sessions[id]; - if (existing != null && !existing.exited) { + if (existing != null && !needsStart(id, isLive: isLive)) { existing.touch(); return existing; } if (existing != null) { - // A VOD/recording transcode that finished cleanly is still fully - // playable from its segments — reuse it instead of re-transcoding. - if (!isLive && existing.exitCode == 0 && _playlistComplete(existing)) { - existing.touch(); - return existing; - } // Process died: clean up before restarting killSession(id); } diff --git a/bin/services/upstream_slots.dart b/bin/services/upstream_slots.dart new file mode 100644 index 0000000..148d062 --- /dev/null +++ b/bin/services/upstream_slots.dart @@ -0,0 +1,151 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:http/http.dart' as http; + +import '../models/playlist_config.dart'; + +/// Connexions ouvertes vers le fournisseur Xtream, par compte. +/// +/// POURQUOI : la plupart des abonnements n'autorisent qu'UNE connexion +/// simultanée (`max_connections: "1"`, vérifié en prod). Or un FFmpeg de +/// lecture survit au spectateur : 4 min pour un direct HLS, jusqu'à 15 min +/// ou la fin du téléchargement pour un film, le temps de détecter la +/// déconnexion pour `turbo.ts`. Le flux suivant tombait sur un slot occupé : +/// 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. +class UpstreamSlots { + final DateTime Function() _now; + final Map _slots = {}; + int _nextToken = 0; + + UpstreamSlots({DateTime Function()? now}) : _now = now ?? DateTime.now; + + /// Inscrit la connexion [id] du compte [account]. [release] la coupe. + /// + /// 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. + int register({ + required String id, + required String account, + required void Function() release, + }) { + final token = ++_nextToken; + _slots[id] = _Slot(token, account, _now(), release); + return token; + } + + void unregister(String id, {int? token}) { + final slot = _slots[id]; + if (slot == null) return; + if (token != null && slot.token != token) return; + _slots.remove(id); + } + + int countFor(String account) => + _slots.values.where((s) => s.account == account).length; + + /// Libère assez de connexions de [account] pour qu'une nouvelle tienne + /// sous [max]. [keep] (la session demandée) n'est jamais coupée. + /// + /// [max] <= 0 = quota inconnu : on ne coupe rien plutôt que de risquer + /// d'interrompre un autre spectateur. + List makeRoom(String account, {required int max, required String keep}) { + 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)); + + final excess = others.length - (max - 1); + if (excess <= 0) return const []; + + final freed = []; + for (final entry in others.take(excess)) { + _slots.remove(entry.key); + try { + entry.value.release(); + } catch (_) {} + freed.add(entry.key); + } + return freed; + } +} + +class _Slot { + final int token; + final String account; + final DateTime since; + final void Function() release; + _Slot(this.token, this.account, this.since, this.release); +} + +/// Clé de compte : deux playlists sur les mêmes identifiants partagent le +/// même quota chez le fournisseur. +String accountKeyOf(PlaylistConfig p) => '${p.dns}|${p.username}'; + +/// `user_info.max_connections` d'une réponse `player_api.php`, ou null. +int? parseMaxConnections(Object? json) { + if (json is! Map) return null; + final info = json['user_info']; + if (info is! Map) return null; + final raw = info['max_connections']; + final value = raw is int ? raw : int.tryParse('$raw'); + return (value != null && value > 0) ? value : null; +} + +/// Quota de connexions par compte, lu chez le fournisseur et mis en cache. +class AccountLimits { + final Duration ttl; + final Future Function(PlaylistConfig) _fetch; + final Map _cache = {}; + + /// Valeur retenue quand le panneau ne répond pas : un seul flux, le cas + /// de loin le plus courant chez les fournisseurs Xtream. + static const fallback = 1; + + AccountLimits({ + this.ttl = const Duration(minutes: 10), + Future Function(PlaylistConfig)? fetch, + }) : _fetch = fetch ?? _fetchFromPanel; + + Future maxFor(PlaylistConfig p) async { + final key = accountKeyOf(p); + final cached = _cache[key]; + if (cached != null && DateTime.now().difference(cached.at) < ttl) { + return cached.max; + } + 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; + } + + static Future _fetchFromPanel(PlaylistConfig p) async { + final uri = Uri.parse('${p.dns}/player_api.php').replace( + queryParameters: {'username': p.username, 'password': p.password}, + ); + final client = http.Client(); + try { + final response = await client.get(uri, headers: { + 'User-Agent': 'VLC/3.0.18 LibVLC/3.0.18', + }).timeout(const Duration(seconds: 5)); + if (response.statusCode != 200) return null; + return parseMaxConnections(jsonDecode(response.body)); + } finally { + client.close(); + } + } +} diff --git a/bin/test/upstream_slots_test.dart b/bin/test/upstream_slots_test.dart new file mode 100644 index 0000000..3bc5002 --- /dev/null +++ b/bin/test/upstream_slots_test.dart @@ -0,0 +1,92 @@ +import 'package:test/test.dart'; +import '../services/upstream_slots.dart'; + +void main() { + group('UpstreamSlots.makeRoom', () { + late UpstreamSlots slots; + late List released; + var clock = DateTime(2026, 1, 1); + + void add(String id, String account) { + clock = clock.add(const Duration(seconds: 1)); + slots.register(id: id, account: account, release: () => released.add(id)); + } + + setUp(() { + slots = UpstreamSlots(now: () => clock); + released = []; + }); + + test('compte à 1 connexion : le flux précédent est coupé', () { + add('live_1_source', 'acc'); + final freed = slots.makeRoom('acc', max: 1, keep: 'live_2_source'); + expect(freed, ['live_1_source']); + expect(released, ['live_1_source']); + expect(slots.countFor('acc'), 0); + }); + + test('la session demandée elle-même n\'est jamais coupée', () { + add('live_1_source', 'acc'); + expect(slots.makeRoom('acc', max: 1, keep: 'live_1_source'), isEmpty); + expect(released, isEmpty); + }); + + test('les plus anciennes partent d\'abord, dans la limite du quota', () { + add('a', 'acc'); + add('b', 'acc'); + add('c', 'acc'); + // 3 connexions max : il faut une place pour la nouvelle → 1 seule coupée. + expect(slots.makeRoom('acc', max: 3, keep: 'new'), ['a']); + expect(slots.countFor('acc'), 2); + }); + + test('un autre compte n\'est jamais touché', () { + add('x', 'other'); + expect(slots.makeRoom('acc', max: 1, keep: 'new'), isEmpty); + expect(slots.countFor('other'), 1); + }); + + test('unregister avec un jeton périmé ne libère pas la session relancée', () { + final old = slots.register(id: 's', account: 'acc', release: () {}); + slots.register(id: 's', account: 'acc', release: () {}); + slots.unregister('s', token: old); + expect(slots.countFor('acc'), 1); + }); + + test('quota inconnu ou nul : rien n\'est coupé', () { + add('a', 'acc'); + expect(slots.makeRoom('acc', max: 0, keep: 'new'), isEmpty); + }); + }); + + group('parseMaxConnections', () { + test('lit la valeur texte renvoyée par player_api', () { + expect( + parseMaxConnections({ + 'user_info': {'max_connections': '1'}, + }), + 1, + ); + }); + + test('accepte un entier', () { + expect( + parseMaxConnections({ + 'user_info': {'max_connections': 3}, + }), + 3, + ); + }); + + test('réponse inexploitable → null', () { + expect(parseMaxConnections({'user_info': {}}), isNull); + expect(parseMaxConnections('oops'), isNull); + expect( + parseMaxConnections({ + 'user_info': {'max_connections': '0'}, + }), + isNull, + ); + }); + }); +}