v2.41 dual simultaneous YouTube ingest (primary+backup tee fan-out)

ingest.py

@@ -6,7 +6,7 @@ import os, threading, time
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
-APP_VERSION = 'relay-v2.40-avfill-splice'
+APP_VERSION = 'relay-v2.41-dual-ingest'
 
 
 def env_rtmp_destination():

relay.py

@@ -572,7 +572,38 @@ class Sender:
             return []
         return ['-rtmp_enhanced_codecs', 'hvc1'] if self.hevc_payload(candidate) else []
 
-    def command(self,destination,candidate=None):
+    @staticmethod
+    def _dual_legs(destination,extra_destinations):
+        """وجهات النشر المتزامن الآمنة للمروحة: الأساسي ثم الاحتياطي، بلا مكرّر ولا
+        فواصل tee داخل أي عنوان. النشر المزدوج هو آلية يوتيوب الرسمية للوفرة:
+        بثّان متزامنان لنفس الحدث، فإذا مات الأساسي أكمل الاحتياطي أمام المشاهدين
+        بدل أن يبدأ الناشر من الصفر فيُرفض عند الفتح (187) لدقائق — حادثة
+        2026-10-21 21:19Z: موت 152 ثم ست جلسات 187 وبطاقة noData لـ3د42ث."""
+        if str(os.getenv('YOUTUBE_DUAL_INGEST','1')).strip().lower() in ('0','off','false','no'):
+            return []
+        legs=[]
+        for url in [destination]+list(extra_destinations or []):
+            url=str(url or '')
+            if (url.startswith(('rtmp://','rtmps://')) and url not in legs
+                    and '|' not in url and ']' not in url):
+                legs.append(url)
+        return legs if len(legs)>=2 else []
+
+    @staticmethod
+    def _tee_spec(legs,enhanced):
+        """مواصفة tee بساق FLV لكل وجهة: onfail=ignore يعني أن ساقًا ميتة (رفض فتح
+        أو قطع منتصف البث) تُسقَط وحدها وتكمل أختها — مُختبَر محليًا على ffmpeg
+        7.0.2: قتل خادم RTMP بعد 4ث من بث 8ث فأكملت ساق الملف 8.02ث كاملة ورجع
+        الناشر 0. مهلة كتابة قصيرة لكل ساق (8ث بدل 20) حتى لا تُسمّم ساق محجوبة
+        أختها عبر خيط المروحة المشترك أكثر من ثوانٍ قبل أن تُسقَط."""
+        try:rw=str(int(os.getenv('YOUTUBE_DUAL_RW_TIMEOUT','8000000')))
+        except (TypeError,ValueError):rw='8000000'
+        opts='f=flv:onfail=ignore:flvflags=no_duration_filesize'
+        if enhanced:opts+=':rtmp_enhanced_codecs=hvc1'
+        opts+=':rw_timeout='+rw
+        return '|'.join('[%s]%s'%(opts,url) for url in legs)
+
+    def command(self,destination,candidate=None,extra_destinations=None):
         # إعلان Enhanced-RTMP يُقاس على الترميز المدفوع فعلًا؛ ملفات الاختبار المحلية لا بروتوكول لها.
         mode=self.send_mode()
         self.active_mode=mode
@@ -665,11 +696,22 @@ class Sender:
                     '-recovery_wait_time',os.getenv('YOUTUBE_RECOVERY_WAIT','1'),
                     '-queue_size',str(q),'-timeshift',str(ts)]
             else:
-                # N26: خرج مباشر بلا مخزن يكذب على الحارس — تقدم out_time هنا يعني تسليمًا فعليًا.
-                output=['-flvflags','no_duration_filesize','-f','flv']
+                # v2.41: بلا timeshift ومع وجهة احتياطية معروفة، الخرج مروحة tee إلى
+                # الاستقبالين معًا (انظر _dual_legs) — الوفرة عند يوتيوب نفسه لا عند
+                # إعادة فتحنا. بلا احتياطي يبقى خرج N26 المباشر كما هو: تقدم out_time
+                # يعني تسليمًا فعليًا، بلا مخزن يكذب على الحارس.
+                legs=self._dual_legs(destination,extra_destinations)
+                if legs:
+                    output=['-f','tee',self._tee_spec(legs,enhanced)]
+                else:
+                    output=['-flvflags','no_duration_filesize','-f','flv']
         else:
             output=['-flvflags','no_duration_filesize','-f','flv']
         ta=self.input_args(candidate)
+        # مع المروحة تسكن الوجهات وإعلان hvc1 ومهلة الكتابة داخل مواصفة كل ساق؛
+        # إلحاق الوجهة المفردة أو الخيارات العامة بعد tee يُفسد الأمر كله.
+        tail=list(output) if (network and output[:1]==['-f'] and 'tee' in output) else \
+            [*enhanced,'-rw_timeout','20000000',*output,destination]
         return [self.ffmpeg,'-hide_banner','-loglevel',('debug' if network and self.trace_rtmp else os.getenv('YOUTUBE_FFMPEG_LOGLEVEL','error')),
             '-nostats','-progress','pipe:1',
             *ta['input'],'-map','0:v:0','-map','0:a:0?',
@@ -680,7 +722,7 @@ class Sender:
             *video,'-c:a','aac','-profile:a',os.getenv('YOUTUBE_AUDIO_PROFILE','aac_low'),
             '-b:a',os.getenv('YOUTUBE_AUDIO_BITRATE','128k'),'-ar','48000','-ac',os.getenv('YOUTUBE_AUDIO_CHANNELS','2'),
             '-af','aresample=async=1:first_pts=0','-avoid_negative_ts','make_zero',
-            *enhanced,'-rw_timeout','20000000',*output,destination]
+            *tail]
     # ------- مهل المصدر (كلها من البيئة؛ القيم الافتراضية مبرَّرة أدناه) -------
     @property
     def stall_limit(self):
@@ -1115,7 +1157,7 @@ class Sender:
         codes = [c for c in codes if c]
         self.log(user, '[CLEANUP] أُوقف FFmpeg (%s%s)' % (reason, '؛ رمز الخروج ' + ','.join(map(str, codes)) if codes else ''))
 
-    def send(self, user, candidate, destination, stop, refresh=None):
+    def send(self, user, candidate, destination, stop, refresh=None, extra_destinations=None):
         """Push one rendition until it dies, the source stalls, or `stop` is set.
 
         الإشراف هنا على **البيانات** لا على العمليات: FFmpeg يبقى حيًا ومسرورًا وهو جالس
@@ -1176,7 +1218,7 @@ class Sender:
             if self.telemetry_key==tele_key:self.telemetry_key=None
         try:
             sess['src_metered'] = not hls          # HLS: FFmpeg يقرأ القائمة، فلا عدّاد مصدر لنا
-            args = self.command(destination, candidate)
+            args = self.command(destination, candidate, extra_destinations=extra_destinations)
             telemetry_observe(tele_key,{'mode':self.active_mode or self.send_mode(),
                                         'source_width':candidate.get('width'),
                                         'source_height':candidate.get('height'),
@@ -1210,8 +1252,10 @@ class Sender:
             with self.lock:
                 self.process = p
             self.log(user, '[RELAY] بدء FFmpeg ← %s (%s؛ mode=%s)' % (candidate['label'],
-                     'إعلان Enhanced-RTMP: hvc1' if '-rtmp_enhanced_codecs' in ' '.join(args)
+                     'إعلان Enhanced-RTMP: hvc1' if 'rtmp_enhanced_codecs' in ' '.join(args)
                      else 'FLV تقليدي: لا إعلان hvc1', self.active_mode or 'unknown'))   # لا إعادة ترميز في هذا الوضع: انقطاع copy لا يترك عملية ورائها
+            if extra_destinations and '-f' in args and 'tee' in args:
+                self.log(user, '[RELAY] نشر مزدوج متزامن: استقبال يوتيوب الأساسي والاحتياطي معًا على نفس الحدث — موت ساق واحدة لا يوقف البث أمام المشاهدين')
 
             def read_errors():
                 # يصنّف أخطاء FFmpeg المختارة؛ لا يُسرّب stderr خامًا ولا روابط البث — الذيل

test_ingest_routes.py

@@ -187,4 +187,60 @@ class PublishDisciplineTests(unittest.TestCase):
         self.mgr._note_early_exit('alice',s,11.0)
         self.assertNotIn('alice',self.mgr._publish_wait)  # رجوع المصدر سريع دائمًا
 
+class DualIngestTests(unittest.TestCase):
+    """v2.41: النشر المتزامن على الاستقبالين هو الوفرة الرسمية. موت ساق واحدة
+    (152 ثم عاصفة 187، 2026-10-21) لا يجوز أن يوقف البث ما دامت الأخرى حيّة."""
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+        self.mgr=ManagedYouTube('/bin/true',lambda u:None,lambda *a:None,lambda:['alice'],
+                               pathlib.Path(self.tmp.name)/'journal.json')
+        self.mgr.journal.change(lambda d:d.update(channel={'id':'CHANNEL'},streams={'alice':{'id':'STREAM'}}))
+        self.primary='rtmps://a.rtmps.youtube.com/live2/test-ingest-only'
+        self.backup='rtmps://b.rtmps.youtube.com:443/live2/test-ingest-only?backup=1'
+        self.mgr._ingest_routes['alice']={'urls':[self.primary,self.backup],
+            'channel_id':'CHANNEL','stream_id':'STREAM'}
+        self.mgr.destinations['alice']=self.primary
+
+    def test_destinations_are_simultaneous_pair_in_preference_order(self):
+        self.assertEqual(self.mgr.ingest_destinations('alice'),[self.primary,self.backup])
+        self.assertTrue(self.mgr.switch_ingest_route('alice'))
+        self.assertEqual(self.mgr.ingest_destinations('alice'),[self.backup,self.primary])
+
+    def test_destinations_fall_back_to_single_after_ownership_change(self):
+        self.mgr.journal.change(lambda d:d.update(channel={'id':'OTHER'}))
+        self.assertEqual(self.mgr.ingest_destinations('alice'),[self.primary])
+
+    def test_destinations_single_when_dual_disabled(self):
+        with patch.dict(os.environ,{'YOUTUBE_DUAL_INGEST':'off'}):
+            self.assertEqual(self.mgr.ingest_destinations('alice'),[self.primary])
+
+    def test_command_tee_fans_out_to_both_legs(self):
+        s=Sender('/bin/true',lambda *a:None);s.mode_override='copy'
+        args=s.command(self.primary,{'codec':'h264'},extra_destinations=[self.backup])
+        joined=' '.join(args)
+        self.assertIn('tee',args)
+        self.assertIn(self.primary,joined);self.assertIn(self.backup,joined)
+        self.assertEqual(joined.count('onfail=ignore'),2)
+        self.assertTrue(args[-1].endswith(self.backup))  # الوجهة داخل المواصفة لا بعدها
+        self.assertNotIn('rtmp_enhanced_codecs',joined)  # h264: لا إعلان معزّز
+
+    def test_command_tee_advertises_hvc1_on_each_leg_for_hevc(self):
+        s=Sender('/bin/true',lambda *a:None);s.mode_override='copy'
+        args=s.command(self.primary,{'codec':'h265'},extra_destinations=[self.backup])
+        self.assertEqual(' '.join(args).count('rtmp_enhanced_codecs=hvc1'),2)
+
+    def test_command_single_destination_is_unchanged(self):
+        s=Sender('/bin/true',lambda *a:None);s.mode_override='copy'
+        args=s.command(self.primary,{'codec':'h264'})
+        self.assertEqual(args[-1],self.primary)
+        self.assertNotIn('tee',args)
+
+    def test_dual_legs_need_two_distinct_safe_urls(self):
+        self.assertEqual(Sender._dual_legs(self.primary,[]),[])
+        self.assertEqual(Sender._dual_legs(self.primary,[self.primary]),[])
+        self.assertEqual(Sender._dual_legs(self.primary,[self.backup]),
+                         [self.primary,self.backup])
+        with patch.dict(os.environ,{'YOUTUBE_DUAL_INGEST':'0'}):
+            self.assertEqual(Sender._dual_legs(self.primary,[self.backup]),[])
+
 if __name__=='__main__':unittest.main(argv=[sys.argv[0]],verbosity=2)

test_relay.py

@@ -428,7 +428,7 @@ last_i=pump.find("sess['last'] = time.monotonic()", pump.find('def _pump_source'
 check('empty FLV chunks do not refresh last', empty_i!=-1 and last_i!=-1 and empty_i<last_i, 'empty=%s last=%s'%(empty_i,last_i))
 td=pump.split('def _teardown')[1].split('def send')[0]
 check('teardown does not stop supervisor Event', 'stop.set()' not in td and "sess['force_stop']" in td)
-check('send keeps FFmpeg via refresh reattach', 'def _try_reattach' in pump and 'def send(self, user, candidate, destination, stop, refresh=None)' in pump)
+check('send keeps FFmpeg via refresh reattach', 'def _try_reattach' in pump and 'def send(self, user, candidate, destination, stop, refresh=None, extra_destinations=None)' in pump)
 check('sender_loop pins codec family during refresh',
       "refresh=lambda: self._refresh_flv(user,getattr(sender,'current_candidate',None))" in mgr_src and
       'def _refresh_flv(self,user,current=None)' in mgr_src)

youtube_manager.py

@@ -531,6 +531,27 @@ class ManagedYouTube:
         # not after the query string, and neither endpoint nor key is persisted.
         return urlunparse((p.scheme,p.netloc,p.path.rstrip('/')+'/'+quote(str(key),safe=''),'',p.query,''))
 
+    def ingest_destinations(self,user):
+        """وجهات النشر المتزامن [المفضّلة الآن، ثم الباقي] من مسارات YouTube API
+        المحفوظة لهذا (قناة، مصدر) حصرًا — نفس حارس الملكية في switch_ingest_route
+        حتى لا تُنشر وجهة قناة/مصدر سابق على حدث جديد. النشر على الاستقبالين معًا
+        (لا التنقّل بينهما بعد الموت) هو وفرة يوتيوب الرسمية: الاحتياطي يصل قبل
+        الحاجة إليه، فموت الأساسي لا يمرّ بالمشاهدين. وجهة البيئة (env) تبقى
+        مفردة: لا زوج احتياطيًا موثّقًا لها. YOUTUBE_DUAL_INGEST=off يرجع للمفرد."""
+        single=self.destinations.get(user)
+        if str(os.getenv('YOUTUBE_DUAL_INGEST','1')).strip().lower() in ('0','off','false','no'):
+            return [single] if single else []
+        routes=self._ingest_routes.get(user) or {}
+        urls=list(routes.get('urls') or [])
+        d=self.journal.read()
+        if (len(urls)>=2 and routes.get('channel_id')==(d.get('channel') or {}).get('id')
+                and routes.get('stream_id')==(d.get('streams',{}).get(user) or {}).get('id')):
+            pref=int(self._ingest_preferred.get(user,0) or 0)
+            if 0<=pref<len(urls):
+                return [urls[pref]]+[u for i,u in enumerate(urls) if i!=pref]
+            return urls
+        return [single] if single else []
+
     def switch_ingest_route(self,user):
         routes=self._ingest_routes.get(user) or {}
         d=self.journal.read()
@@ -1234,12 +1255,15 @@ class ManagedYouTube:
                     # الوجهة تُقرأ في كل دورة: لو تغيّر مفتاح الاستقبال (إعادة استخدام المصدر أو
                     # إعادة ربط للحدث) بقي هذا الخيط يطرق الباب القديم بمفتاح قديم إلى الأبد.
                     destination = self.destinations.get(user) or destination
+                    dests = self.ingest_destinations(user) or ([destination] if destination else [])
+                    if dests: destination = dests[0]
                     try:
                         self.log(user,'[RELAY] @%s starting (%s)'%(user,self._source_brief(candidate)))
                         sender.telemetry_context={'event_key':key,'room_id':room_id,
                                                   'broadcast_id':(self.journal.read()['events'].get(key) or {}).get('broadcast_id','')}
                         if sender.send(user,candidate,destination,stop,
-                                       refresh=lambda: self._refresh_flv(user,getattr(sender,'current_candidate',None))):
+                                       refresh=lambda: self._refresh_flv(user,getattr(sender,'current_candidate',None)),
+                                       extra_destinations=dests[1:]):
                             sent=True
                             self._rendition_pin[user]=self._source_pin(getattr(sender,'current_candidate',None) or candidate)
                             sender.lock_gold(candidate)