veza/veza-stream-server/docs/AUDIT_STREAM_PROCESSING.md

🔍 AUDIT INITIAL - Stream Processing Thread

Date: 2025-12-05
Service: veza-stream-server
Mission: P1 - Implémentation complète du thread de Stream Processing

---

1. ÉTAT ACTUEL DU CODE

1.1 Quelles parties sont déjà prêtes ?

✅ Pool FFmpeg (src/core/encoding_pool.rs)

• État: ✅ 100% Fonctionnel
• Pool de workers avec async_channel pour la queue
• Workers qui spawn des processus FFmpeg
• Capture stderr en streaming (lignes 202-232)
• Parsing de progression via FfmpegProgress (lignes 214-224)
• Timeout configurable (10 minutes par défaut)
• Mise à jour DB des statuts (pending, encoding, done, error)
• Parse et stocke les segments après complétion (lignes 281-349)

✅ Command Builder (src/transcoding/ffmpeg/command_builder.rs)

• État: ✅ Fonctionnel
• Builder pattern pour commandes FFmpeg
• Support HLS complet (-f hls, -hls_time, -hls_playlist_type vod)
• Pattern de nommage segments: segment_%05d.ts (ligne 148)
• Isolation CPU: -threads 1 (ligne 105)
• Audio-only: -map 0:a (ligne 108)
• Redirection stdout/stderr: Stdio::piped() (lignes 156-157)

✅ Progress Parser (src/transcoding/ffmpeg/progress_parser.rs)

• État: ✅ Fonctionnel
• Regex pour parser time=HH:MM:SS.mm
• Support audio-only (sans frame/fps)
• Extraction de time, speed, bitrate

✅ Routes API (src/routes/transcode.rs)

• État: ✅ Fonctionnel
• POST /v1/stream/transcode - Soumettre un job
• GET /v1/stream/job/{id} - Statut d'un job
• GET /v1/stream/hls/{job_id}/index.m3u8 - Manifest HLS
• GET /v1/stream/hls/{job_id}/{segment} - Segment .ts

✅ Database Schema

• État: ✅ Prêt
• stream_jobs: id, track_id, status, created_at, updated_at, error_message
• stream_segments: id, track_id, quality, segment_index, path, duration, created_at
• Index optimisés pour requêtes fréquentes

✅ Job Structure (src/core/job.rs)

• État: ✅ Fonctionnel
• EncodeJob: structure complète avec tous les paramètres
• EncodeJobStatus: enum (Pending, Encoding, Done, Error)

❌ Stream Processing Thread (src/core/encoder.rs ligne 459)

• État: ❌ TODO non implémenté
• EncoderPipeline::start_processing() ne fait rien (ligne 459)
• Aucun thread de traitement temps réel
• Aucun monitoring incrémental des segments

---

1.2 Qu'est-ce que le thread doit réellement faire ?

Le thread de Stream Processing doit :

1. Spawn FFmpeg

   • Utiliser tokio::process::Command depuis FfmpegCommandBuilder
   • Rediriger stdout/stderr
   • Enregistrer PID + start_time en DB

2. Monitoring Temps Réel

   • Lire stderr ligne par ligne (déjà fait dans encoding_pool.rs mais de manière basique)
   • Détecter les événements FFmpeg:
     • Ouverture de segments: Opening 'segment_00000.ts'
     • Progression: time=00:00:05.12
     • Erreurs: error, Error, ERROR
     • Fin: muxing overhead
   • Extraire les timestamps et détecter la génération de segments

3. Suivi des Segments

   • Détecter chaque segment généré en temps réel (pas seulement après complétion)
   • Enregistrer immédiatement en DB (stream_segments)
   • Mettre à jour current_duration dans stream_jobs

4. Persistance DB Temps Réel

   • Insérer chaque segment dès qu'il est détecté
   • Mettre à jour stream_jobs.status = 'encoding' avec progression
   • Mettre à jour stream_jobs.current_duration (si colonne existe)

5. Callbacks de Fin

   • on_success: Mettre à jour status = 'done', finaliser playlist HLS
   • on_error: Mettre à jour status = 'error', logger l'erreur
   • on_timeout: Tuer le processus, nettoyer fichiers temporaires

6. Gestion Stop/Retry

   • Support pour arrêter un job en cours
   • Retry automatique en cas d'erreur temporaire

---

1.3 Quels événements FFmpeg sont détectables (stderr lines) ?

Événements Détectables dans stderr FFmpeg:

1. Ouverture de Segment

   Opening 'segment_00000.ts' for writing
   

   • Regex: Opening '([^']+)' for writing
   • Indique qu'un nouveau segment est en cours de création

2. Progression

   frame=  123 fps=0.0 q=-1.0 size=    1024kB time=00:00:05.12 bitrate=1638.4kbits/s speed=10.2x
   

   • Déjà parsé par FfmpegProgress::parse()
   • Contient time (timestamp actuel)

3. Erreurs

   error: Invalid data found when processing input
   Error opening input file
   ERROR: Failed to open output file
   

   • Détection: line.contains("error") || line.contains("Error") || line.contains("ERROR")
   • Déjà détecté dans encoding_pool.rs ligne 227

4. Fin de Muxing

   muxing overhead: 1.234567%
   

   • Indique que FFmpeg a terminé l'encodage
   • Regex: muxing overhead: ([\d.]+)%

5. Segment Complet

   • Pas d'événement direct dans stderr
   • Solution: Détecter quand un nouveau segment apparaît dans le répertoire
   • Ou: Parser le manifest .m3u8 en temps réel (mais il est mis à jour de manière asynchrone)

---

1.4 Comment les segments sont-ils nommés ?

D'après command_builder.rs ligne 148:

let segment_filename = output_path.with_file_name("segment_%05d.ts");
cmd.arg("-hls_segment_filename").arg(segment_filename);

Pattern: segment_%05d.ts où %05d est le numéro de séquence (00000, 00001, 00002, ...)

Exemple:

• segment_00000.ts
• segment_00001.ts
• segment_00002.ts

Répertoire: /data/streams/<track_id>/<quality>/segment_*.ts

---

1.5 Est-ce que la playlist live / vod est déjà générée automatiquement ?

Oui, mais de manière asynchrone:

1. Génération Automatique: FFmpeg génère automatiquement index.m3u8 avec les segments
2. Type VOD: -hls_playlist_type vod (ligne 141 de command_builder.rs)
3. Liste Complète: -hls_list_size 0 (ligne 143) = tous les segments dans la playlist
4. Mise à Jour: La playlist est mise à jour par FFmpeg au fur et à mesure

Problème Actuel:

• La playlist n'est parsée qu'après la complétion du job (ligne 247 de encoding_pool.rs)
• Pas de parsing temps réel pendant l'encodage

Solution:

• Parser le manifest .m3u8 périodiquement (toutes les 2-3 secondes)
• Ou: Surveiller le répertoire pour détecter les nouveaux fichiers .ts

---

1.6 Le système supporte-t-il ABR (multi-qualité) ?

Oui, mais de manière séquentielle:

1. Qualités Supportées: low, medium, high, hi_res (défini dans job.rs)
2. Encodage Multi-Qualité: Chaque qualité est encodée dans un job séparé
3. Master Playlist: Pas de master playlist .m3u8 pour ABR actuellement
4. Structure:

   /data/streams/<track_id>/
     low/index.m3u8 + segments
     medium/index.m3u8 + segments
     high/index.m3u8 + segments
     hi_res/index.m3u8 + segments
   

ABR Support Futur:

• Créer un master.m3u8 qui référence les playlists de chaque qualité
• Format:

  #EXTM3U
  #EXT-X-STREAM-INF:BANDWIDTH=64000,CODECS="mp4a.40.2"
  low/playlist.m3u8
  #EXT-X-STREAM-INF:BANDWIDTH=128000,CODECS="mp4a.40.2"
  medium/playlist.m3u8
  ...
  

---

2. GAPS IDENTIFIÉS

2.1 Ce qui manque pour le Processing Thread

1. Module StreamProcessor

   • Structure pour gérer un job de processing
   • Méthode run() qui orchestre spawn → monitor → finalize

2. Module FFmpegMonitor

   • Lecture stderr ligne par ligne
   • Détection événements (segments, erreurs, fin)
   • Extraction timestamps et progression

3. Module SegmentTracker

   • Suivi des segments détectés
   • Persistance DB temps réel
   • Mise à jour current_duration

4. Module ProcessingCallbacks

   • Callbacks on_success, on_error, on_timeout
   • Finalisation playlist HLS
   • Nettoyage fichiers temporaires

5. API Status Temps Réel

   • Route GET /api/streams/jobs/{id}/status
   • Retourne segments détectés + progression

6. Intégration dans EncoderPipeline

   • Remplacer le TODO ligne 459 par un vrai thread
   • Utiliser StreamProcessor pour le traitement

---

3. DESIGN PROPOSÉ

3.1 Architecture

EncoderPipeline::start_processing()
  └─> StreamProcessor::new(job, db, config)
      └─> StreamProcessor::run()
          ├─> spawn_ffmpeg() → tokio::process::Command
          ├─> monitor_output() → FFmpegMonitor
          │   ├─> Lire stderr ligne par ligne
          │   ├─> Détecter événements
          │   └─> SegmentTracker::register(segment)
          ├─> SegmentTracker::persist() → DB
          └─> finalize() → ProcessingCallbacks
              ├─> on_success → DB status = 'done'
              ├─> on_error → DB status = 'error'
              └─> on_timeout → kill process, cleanup

3.2 Flux de Données

FFmpeg stderr → FFmpegMonitor → SegmentTracker → DB
                                      ↓
                              ProcessingStatus (mpsc)
                                      ↓
                              API Status Endpoint

---

4. PROCHAINES ÉTAPES

1. ✅ Audit initial (ce document)
2. ⏳ Implémenter StreamProcessor
3. ⏳ Implémenter FFmpegMonitor
4. ⏳ Implémenter SegmentTracker
5. ⏳ Implémenter ProcessingCallbacks
6. ⏳ Créer API status endpoint
7. ⏳ Intégrer dans EncoderPipeline
8. ⏳ Tests unitaires et intégration
9. ⏳ Documentation complète

---

Auteur: Veza Stream Server Team
Dernière mise à jour: 2025-12-05