Persistent room format lock + fill/swap-aware write deadline (v2.33-format-lock-fillguard)

ingest.py

@@ -6,7 +6,7 @@ import os, threading, time
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
-APP_VERSION = 'relay-v2.32-gap-filler'
+APP_VERSION = 'relay-v2.33-format-lock-fillguard'
 
 
 def env_rtmp_destination():

relay.py

@@ -701,6 +701,18 @@ class Sender:
             raw = os.getenv('YOUTUBE_SOURCE_STALL_SECONDS', '12')
         return max(0.0, float(raw))
 
+    @staticmethod
+    def _deadline_base(sess):
+        """مرجع مهلة كتابة stdin: آخر بايت مصدر عادةً. أثناء حشو فجوة أو تبديل مصدر
+        فالصمت مخطّط له والضخّ قد يكون من خيط الحشو — المرجع يصير آخر تقدّم فعلي
+        من FFmpeg (ff_last)، فلا يُقتل خرج يعمل بتهمة «عالق» بينما المُرمِّز يطحن
+        إطارات مجمّدة أو يخيط مصدرًا جديدًا (مقيس: talolil 00:21:50 و00:30:49).
+        خرج محجوب حقًا يجمّد التقدّم أيضًا، فتمضي المهلة ويموت بنظافة كالعادة."""
+        base=float(sess.get('last') or 0)
+        if sess.get('filling') or sess.get('swapping'):
+            base=max(base,float(sess.get('ff_last') or 0))
+        return base
+
     def _stream_tuning(self):
         """ما يعادله في هذه الشجرة لخيارات Streamlink (المشروع لا يستعمل Streamlink أصلًا):
 
@@ -835,7 +847,7 @@ class Sender:
                 try:
                     with sess['stdin_lock']:
                         if stall_limit:
-                            self._write_deadline(p.stdin, buf, sess['last'] + stall_limit)
+                            self._write_deadline(p.stdin, buf, self._deadline_base(sess) + stall_limit)
                         else:
                             view = memoryview(buf)
                             while view and not stop.is_set() and not sess['force_stop']:

test_smart_mode.py

@@ -209,5 +209,75 @@ class FormatLockTests(unittest.TestCase):
             self.assertEqual(sender.send_mode(), 'copy')
 
 
+class _ExitSender:
+    """أصغر مرسل يراه _note_early_exit عند نهاية جلسة."""
+    def __init__(self, unstable=False, mode='matchgop', out=0.0):
+        self.format_unstable = unstable
+        self.last_exit_code = 187
+        self.last_failure_side = 'output'
+        self.last_source_gap = 0.0
+        self._snap = {'output_seconds': out, 'rtmp_phase': 'accepted',
+                      'source_bytes': 4096, 'send_mode': mode}
+    def snapshot(self): return dict(self._snap)
+    def send_mode(self): return self._snap['send_mode']
+    def request_sticky(self, *a, **k): return False
+
+
+class ManagerFormatLockTests(unittest.TestCase):
+    """قفل التطبيع محفوظ في السجلّ على مستوى (حساب، غرفة): إعادة بناء المرسل
+    لا تمحوه، وجلسة COPY مستقرة على نفس الغرفة وحدها تفكّه."""
+
+    def make_manager(self):
+        tmp = tempfile.TemporaryDirectory(prefix='fmt-lock-')
+        self.addCleanup(tmp.cleanup)
+        manager = ManagedYouTube('/bin/true', lambda u: None, lambda *a: None,
+                                 lambda: ['alice'], pathlib.Path(tmp.name) / 'journal.json')
+        manager.journal.change(lambda d: d['events'].update(
+            {'alice:R1': {'username': 'alice', 'room_id': 'R1'}}))
+        return manager
+
+    def test_instability_is_journaled_for_the_room(self):
+        manager = self.make_manager()
+        manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0)
+        self.assertEqual(manager._format_lock_room('alice'), 'R1')
+
+    def test_fresh_sender_is_armed_from_the_journal(self):
+        manager = self.make_manager()
+        manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0)
+        sender = Sender('/bin/true', lambda *a: None)
+        self.assertFalse(sender.format_unstable)
+        self.assertTrue(manager._arm_format_lock('alice', sender, 'R1'))
+        self.assertTrue(sender.format_unstable)
+
+    def test_lock_does_not_leak_into_a_new_room(self):
+        manager = self.make_manager()
+        manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0)
+        sender = Sender('/bin/true', lambda *a: None)
+        self.assertFalse(manager._arm_format_lock('alice', sender, 'R2'))
+        self.assertFalse(sender.format_unstable)
+
+    def test_verified_copy_session_releases_the_lock(self):
+        manager = self.make_manager()
+        manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0)
+        self.assertEqual(manager._format_lock_room('alice'), 'R1')
+        manager._note_early_exit('alice', _ExitSender(unstable=False, mode='copy', out=60.0), 65.0)
+        self.assertEqual(manager._format_lock_room('alice'), '')
+
+
+class WriteDeadlineTests(unittest.TestCase):
+    """مهلة كتابة stdin أثناء حشو/تبديل تُقاس على تقدّم FFmpeg لا على صمت المصدر."""
+
+    def test_normal_base_is_last_source_byte(self):
+        self.assertEqual(Sender._deadline_base({'last': 100.0, 'ff_last': 150.0}), 100.0)
+
+    def test_filling_base_follows_ffmpeg_progress(self):
+        self.assertEqual(Sender._deadline_base(
+            {'last': 100.0, 'ff_last': 150.0, 'filling': True}), 150.0)
+
+    def test_swapping_base_never_moves_backwards(self):
+        self.assertEqual(Sender._deadline_base(
+            {'last': 200.0, 'ff_last': 150.0, 'swapping': True}), 200.0)
+
+
 if __name__ == '__main__':
     unittest.main(argv=[sys.argv[0]], verbosity=2)

youtube_manager.py

@@ -1020,6 +1020,7 @@ class ManagedYouTube:
         sender.on_sticky=self._remember_sender_mode
         sender.on_sticky_pending=self._remember_pending_sender_mode
         sender.on_sticky_clear=self._clear_sender_mode
+        self._arm_format_lock(u,sender,ev.get('room_id'))
         forced=self.forced_mode(u)
         saved_mode=str(ev.get('sender_mode') or self._sender_sticky.get(u) or '')
         pending=self._sender_pending.get(u)
@@ -1321,6 +1322,42 @@ class ManagedYouTube:
             self.log(user, '[WATCHDOG] الاستقبال يقبل التدفق؛ مُسح تأجيل العقاب وأثره من اللوحة (النوافذ: %d)' % cleared)
         return cleared
 
+    def _format_lock_room(self,user):
+        try:return str((self.journal.read().get('streams',{}).get(user) or {}).get('format_lock_room') or '')
+        except Exception:return ''
+
+    def _note_format_instability(self,user,sender):
+        """قفل تطبيع محفوظ على مستوى (حساب، غرفة): الجسر وثّق GOP/DTS مضطربًا على
+        هذا المرسل، لكن start_sender يبني مرسلًا جديدًا عند كل إعادة إشراف فيُنسى
+        قفل مثيله (مقيس: 18_x200 رجع copy بعد إعادة بناء فماتت جلسته). الحفظ في
+        سجلّ القنوات يجعل أول جلسة بعد أي إعادة بناء تبدأ مطبّعة. الفكّ مشروط:
+        جلسة COPY نظيفة ≥25 ث على نفس الغرفة (فحص حاكم البطء عادة) تعني أن المصدر
+        تصالح. غرفة جديدة = صفحة بيضاء دائمًا."""
+        try:unstable=bool(getattr(sender,'format_unstable',False))
+        except Exception:unstable=False
+        room=''
+        for ev in self.journal.read().get('events',{}).values():
+            if ev.get('username')==user and ev.get('room_id'):room=str(ev['room_id'])
+        if not room:return
+        if unstable:
+            if self._format_lock_room(user)!=room:
+                self.streams_change(user,format_lock_room=room)
+                self.log(user,'[FORMAT] وُثّق اضطراب صيغة على هذه الغرفة؛ حُفظ قفل التطبيع في السجلّ فلا يُنسى بإعادة بناء الناشر')
+            return
+        if self._format_lock_room(user)==room:
+            try:snap=sender.snapshot() or {}
+            except Exception:snap={}
+            if str(snap.get('send_mode') or '')=='copy' and float(snap.get('output_seconds') or 0)>=25:
+                self.streams_change(user,format_lock_room='')
+                self.log(user,'[FORMAT] جلسة COPY مستقرة ≥25 ث على نفس الغرفة؛ فُكّ قفل التطبيع المحفوظ')
+
+    def _arm_format_lock(self,user,sender,room):
+        if room and self._format_lock_room(user)==str(room):
+            sender.format_unstable=True
+            self.log(user,'[FORMAT] قفل تطبيع محفوظ لهذه الغرفة من اضطراب سابق؛ نبدأ مطبّعًا بدل إعادة اكتشافه بموت جلسات')
+            return True
+        return False
+
     def _note_early_exit(self, user, sender, lasted):
         """وتيرة إعادة الاتصال عند رفض الوجهة: يوتيوب يقفل الاتصال بعد ثوانٍ من كل محاولة
         (رمز 187 أو قطعه من الخادم) — هذا رفض عند حافة الاستقبال لا صمت مصدر، والنقر كل 15ث
@@ -1330,6 +1367,7 @@ class ManagedYouTube:
         output_side = getattr(sender, 'last_failure_side', '') == 'output'
         if code is None and not output_side:
             return                              # لم تخرج العملية وحدَها: لا حكم على الوجهة
+        self._note_format_instability(user, sender)
         gap = float(getattr(sender, 'last_source_gap', 0.0) or 0.0)
         if gap >= float(os.getenv('YOUTUBE_SOURCE_GAP_SUSPECT_SECONDS', '8')) and not output_side:
             # جوع مصدر لا رفض وجهة: التأجيل هنا يطيل أمد الصمت الذي يراه المشاهدون. نُسرّع