Fix live source handover without interrupting the YouTube publisher

Start the replacement reader after releasing its handover flag; cancel and join old readers exclusively. Handle CDN EOF/read failures inside the same FFmpeg, allow bounded first-byte grace, reuse recent compatible live observations and ignore 16x16 placeholder canvas dimensions. Preserve channel, user settings, events and Secrets. Regression coverage includes real HTTP/FFmpeg continuity; 367 checks passed.

Dockerfile

@@ -6,7 +6,7 @@ WORKDIR /app
 COPY requirements.txt .
 RUN pip install --no-cache-dir -U -r requirements.txt
 COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py tiktok_auth.py ingest.py index.html ./
-COPY verify.sh verify_tokens.py probe_live.py test_watchdog.py test_lifecycle.py test_relay.py test_concurrency.py test_flv_bridge.py test_regressions.py test_completion.py test_channel_switch.py test_dashboard.js ./
+COPY verify.sh verify_tokens.py probe_live.py test_watchdog.py test_lifecycle.py test_relay.py test_source_reconnect.py test_concurrency.py test_flv_bridge.py test_regressions.py test_completion.py test_channel_switch.py test_dashboard.js ./
 
 # TikTok Live Monitor → Direct Relay → YouTube. No local video files.
 ENV TZ=Asia/Baghdad \
@@ -47,6 +47,7 @@ ENV TZ=Asia/Baghdad \
     YOUTUBE_FFMPEG_LOGLEVEL=info \
     YOUTUBE_SOURCE_READ_TIMEOUT=15 \
     YOUTUBE_SOURCE_READ_ATTEMPTS=3 \
+    YOUTUBE_REATTACH_GRACE_SECONDS=2 \
     YOUTUBE_PROGRESS_STALL_SECONDS=60 \
     YOUTUBE_KILL_GRACE_SECONDS=10 \
     YOUTUBE_PUMP_JOIN_SECONDS=2 \

README.md

@@ -8,7 +8,19 @@ pinned: false
 app_port: 7860
 ---
 
-# Reoo88/On — relay-v2.22-channel-confirmation
+# Reoo88/On — relay-v2.23-source-handover
+
+## إصلاح انقطاع المصدر في v2.23
+
+كان القارئ البديل يبدأ قبل مسح `swap_source`، فيخرج أحيانًا دون قراءة أي بايت، ثم يُعاد فتح المصدر مرارًا بينما يوتيوب يعرض `noData`. الآن يُوقف القارئ القديم أولًا، ولا يُعاد استعمال جسر FLV إذا بقي حيًا؛ تُمسح علامة التسليم قبل تشغيل القارئ الجديد. لكل قارئ إلغاء مستقل لمنع بايت متأخر من الجلسة السابقة.
+
+- إغلاق CDN للاستجابة أو خطأ القراءة يجرّب مصدرًا بديلًا داخل **نفس FFmpeg ونفس حدث يوتيوب** قبل إعادة تشغيل الناشر.
+- تُمنح الاستجابة الجديدة مهلة أول بايت قدرها ثانيتان (`YOUTUBE_REATTACH_GRACE_SECONDS`، حد أقصى 5 ث ونصف مهلة الصمت). لا يُحدّث عدّاد وصول البيانات وهميًا، ولا تُخفى أعطال خرج RTMP.
+- أثناء إعادة التوصيل نجرّب رابطًا متوافق الترميز من رصد LIVE حديث (بحد أقصى 45 ث) بدل تجويع الناشر بانتظار طلبات بيانات تيك توك الطويلة. الرصد القديم أو غير المتوافق لا يُعتمد، وفحص انتهاء البث يبقى كما هو.
+- ترويسة البداية `16×16` لا تثبّت كانفس التحويل الخفيف. عند غياب مقاس صالح يستعمل `matchgop` كانفس ثابتًا `720×1280`؛ أما مسار نسخ AVC/HEVC الأصلي فلا يُعاد ترميزه بسبب هذه الرقعة.
+
+`test_source_reconnect.py` يشمل جدولة يبدأ فيها القارئ فورًا، وقارئًا قديمًا لم يتوقف، وإيقاف المستخدم، وعدم تزوير العمر، والترويسة الصغيرة، واختبار HTTP + FFmpeg حقيقيين: إغلاقات متكررة وتأخير أول بايت، مع استمرار الساعة بنفس عملية الإرسال في وضعي `copy` و`matchgop`. لا يستعمل الاختبار بثوثًا حقيقية.
+
 
 رصد TikTok LIVE → FFmpeg → YouTube RTMPS.
 **لا تسجيل محلي، لا أرشيف فيديو، ولا ملفات فيديو على القرص.**

ingest.py

@@ -6,7 +6,7 @@ import os, threading, time
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
-APP_VERSION = 'relay-v2.22-channel-confirmation'
+APP_VERSION = 'relay-v2.23-source-handover'
 
 
 def env_rtmp_destination():

relay.py

@@ -351,7 +351,7 @@ class Sender:
         an already-open live event is exactly the flash that only a YouTube-app restart
         clears for existing viewers.
         """
-        if int(getattr(self, '_gop_w', 0) or 0) >= 16:
+        if int(getattr(self, '_gop_w', 0) or 0) >= 32:
             return (self._gop_w, self._gop_h)
         if candidate and not (w and h):
             try:
@@ -364,7 +364,7 @@ class Sender:
         except (TypeError, ValueError):
             w = h = 0
         w, h = w - w % 2, h - h % 2
-        if w >= 16 and h >= 16:
+        if w >= 32 and h >= 32:
             w, h = self.youtube_safe_size(w, h)
             self._gop_w, self._gop_h = w, h
         return (int(getattr(self, '_gop_w', 0) or 0), int(getattr(self, '_gop_h', 0) or 0))
@@ -381,8 +381,8 @@ class Sender:
             w, h = int(w or 0), int(h or 0)
         except (TypeError, ValueError):
             return 0, 0
-        if w < 16 or h < 16:
-            return w, h
+        if w < 32 or h < 32:
+            return 0, 0  # TikTok may begin with a 16x16 placeholder configuration.
         short, long_ = min(w, h), max(w, h)
         portrait = h >= w
         if short <= 720 and long_ >= 1280:
@@ -468,15 +468,12 @@ class Sender:
             video=['-c:v','copy']
         else:
             if mode=='matchgop':
-                w=int((candidate or {}).get('width') or 0); h=int((candidate or {}).get('height') or 0)
-                w,h=w-w%2,h-h%2
-                if not getattr(self,'_gop_w',0) and w>=16 and h>=16:
-                    self._gop_w,self._gop_h=w,h
-                cw,ch=int(getattr(self,'_gop_w',0) or w), int(getattr(self,'_gop_h',0) or h)
-                if cw>=16 and ch>=16:
-                    vf=self.canvas(cw,ch)
-                else:
-                    vf='scale=trunc(iw/2)*2:trunc(ih/2)*2,setsar=1,format=yuv420p'
+                self.lock_gold(candidate)
+                if not self._gop_w or not self._gop_h:
+                    # A 16x16 startup header is not the player's real canvas. Unknown
+                    # dimensions still need a constant output, not a reinitializing scale.
+                    self._gop_w,self._gop_h=720,1280
+                vf=self.canvas(self._gop_w,self._gop_h)
                 bitrate=os.getenv('VIDEO_BITRATE','').strip() or '4500k'
             else:
                 width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k')
@@ -638,7 +635,7 @@ class Sender:
         except BaseException as e:          # تُعاد رفعها في خيط الضخّ لتُصنَّف هناك
             out.put(e)
 
-    def _pump_source(self, response, p, bridge, stop, sess, read_timeout, retries):
+    def _pump_source(self, response, p, bridge, stop, sess, read_timeout, retries, cancel=None):
         """نقل بايتات تيك توك إلى FFmpeg في خيط مستقل، مع تسجيل زمن آخر وصول فعلي.
 
         القراءة هنا هي مصدر الحقيقة: `poll()` الخاص بـ FFmpeg يبقى ناجحًا وهو جالس ينتظر
@@ -649,7 +646,8 @@ class Sender:
         chunk_wait = min(read_timeout, max(1.0, self._stall_seconds() / 3.0)) if self._stall_seconds() else read_timeout
         pending = None
         reader = None
-        while not stop.is_set() and not sess['force_stop'] and not sess.get('swap_source'):
+        while (not stop.is_set() and not sess['force_stop'] and not sess.get('swap_source')
+               and not (cancel and cancel.is_set())):
             if time.monotonic() > deadline:
                 sess['error'] = 'انتهت مهلة الجلسة القصوى (%ds)' % int(sess['max_seconds'])
                 return
@@ -665,12 +663,17 @@ class Sender:
             try:
                 got = pending.get(timeout=chunk_wait)
             except queue.Empty:
+                if stop.is_set() or sess['force_stop'] or sess.get('swap_source') or (cancel and cancel.is_set()):
+                    return
                 silent = time.monotonic() - sess['last']
                 if self._stall_seconds() and silent > self._stall_seconds():
                     sess['error'] = 'صمت من تيك توك (%.0f ث بلا بايت)' % silent
                     return
                 continue
             pending = None
+            # A late result from the old HTTP reader cannot mutate the replacement pump.
+            if stop.is_set() or sess['force_stop'] or sess.get('swap_source') or (cancel and cancel.is_set()):
+                return
             if isinstance(got, BaseException):
                 e = got
                 if self._is_timeout(e) and not sess['force_stop']:
@@ -728,13 +731,17 @@ class Sender:
         Returns (new_response, new_pump) or None. Viewer cuts come from killing FFmpeg, not from
         a 1–2 s TikTok URL refresh into the same process.
         """
-        if not refresh or p is None or p.stdin is None or p.poll() is not None:
+        if (not refresh or p is None or p.stdin is None or p.poll() is not None
+                or stop.is_set() or sess.get('force_stop')):
             return None
         if int(sess.get('reattaches') or 0) >= 24:
             return None
         sess['swapping'] = True
         sess['swap_source'] = True
         sess['swap_started'] = time.monotonic()
+        old_cancel = sess.get('pump_cancel')
+        if old_cancel is not None:
+            old_cancel.set()
         try:
             if response:
                 try:
@@ -743,12 +750,18 @@ class Sender:
                     pass
             if pump is not None:
                 pump.join(timeout=float(os.getenv('YOUTUBE_PUMP_JOIN_SECONDS', '2')))
+                if pump.is_alive():
+                    self.log(user, '[RELAY] قارئ المصدر السابق لم يتوقف؛ لا نشارك معه الجسر أو أنبوب FFmpeg')
+                    return None
+            if stop.is_set() or sess.get('force_stop'):
+                return None
             try:
                 cand = refresh()
             except Exception as e:
                 self.log(user, '[RELAY] تعذر رصد مصدر بديل دون قطع يوتيوب (%s)' % type(e).__name__)
                 return None
-            if not cand or cand.get('media') == 'hls' or not cand.get('url'):
+            if (not cand or cand.get('media') == 'hls' or not cand.get('url')
+                    or stop.is_set() or sess.get('force_stop') or p.poll() is not None):
                 return None
             try:
                 new_resp = requests.get(cand['url'],
@@ -759,19 +772,34 @@ class Sender:
             except Exception as e:
                 self.log(user, '[RELAY] فشل فتح مصدر تيك توك البديل (%s)' % type(e).__name__)
                 return None
+            if stop.is_set() or sess.get('force_stop') or p.poll() is not None:
+                new_resp.close()
+                return None
             bridge.reconnect()
             sess['ended'] = False
             sess['error'] = ''
             sess['stall_reason'] = ''
-            sess['force_stop'] = False
             sess['timeouts'] = 0
             # Do not bump sess['last'] here: that hid 40–57s of silence from the
             # 12s stall watchdog (measured @vimtoo.2: out frozen, age 42–57, yh still good).
             sess['reattaches'] = int(sess.get('reattaches') or 0) + 1
             self.current_candidate = cand
             self.set(source=cand.get('label') or cand.get('brief') or '', stalled=False)
+            try:
+                grace = max(0.5, min(5.0, float(os.getenv('YOUTUBE_REATTACH_GRACE_SECONDS', '2'))))
+            except (TypeError, ValueError):
+                grace = 2.0
+            limit = self._stall_seconds()
+            if limit:
+                grace = min(grace, max(0.5, limit / 2.0))
+            sess['reattach_not_before'] = time.monotonic() + grace
+            new_cancel = threading.Event()
+            sess['pump_cancel'] = new_cancel
+            # Clear the OLD reader's handover flag BEFORE starting its replacement. If
+            # start() schedules the new target immediately, a true flag kills it unread.
+            sess['swap_source'] = False
             new_pump = threading.Thread(target=self._pump_source,
-                                        args=(new_resp, p, bridge, stop, sess, read_timeout, retries),
+                                        args=(new_resp, p, bridge, stop, sess, read_timeout, retries, new_cancel),
                                         daemon=True)
             new_pump.start()
             self.log(user, '[RELAY] أُعيد توصيل تيك توك على نفس جلسة يوتيوب — بلا قتل FFmpeg')
@@ -970,8 +998,9 @@ class Sender:
             threading.Thread(target=read_progress, daemon=True).start()
             pump = None
             if response is not None:
+                sess['pump_cancel'] = threading.Event()
                 pump = threading.Thread(target=self._pump_source,
-                                        args=(response, p, bridge, stop, sess, read_timeout, retries),
+                                        args=(response, p, bridge, stop, sess, read_timeout, retries, sess['pump_cancel']),
                                         daemon=True)
                 pump.start()
             # ---- الحارس: لا تعتمد على poll() ولا على رمز الخروج ----
@@ -986,6 +1015,20 @@ class Sender:
                 # في HLS يستحيل القياس إلا على FFmpeg (هو من يفتح القائمة)، فنقبل تقدّمه.
                 _fresh = (sess['ff_last'] if response is None else sess['last'])
                 swap_for = (now - float(sess.get('swap_started') or 0)) if sess.get('swapping') else 0.0
+                # Give a newly opened HTTP source time to deliver its first actual bytes.
+                # Do not alter sess['last']: source age remains honest while we wait.
+                recent_attach = now < float(sess.get('reattach_not_before') or 0)
+                tried_reattach = False
+                source_done = response is not None and (sess.get('ended') or sess.get('error') or
+                              (pump is not None and not pump.is_alive()))
+                if (source_done and not recent_attach and not sess.get('force_stop')
+                        and not sess.get('writing') and not sess.get('output_failure')):
+                    tried_reattach = True
+                    attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
+                                                  read_timeout, retries, response, pump)
+                    if attached:
+                        response, pump = attached
+                        continue
                 # swapping must not disable the stall watchdog for longer than the stall
                 # limit: refresh()+HTTP open was measured at 40–57s of frozen viewers.
                 if self._recovering_copy:
@@ -1001,21 +1044,26 @@ class Sender:
                         self.sticky_mode=None
                         self.sticky_since=0.0
                         self._recovering_copy=False
-                if (stall_limit and now - _fresh > 3 and not sess.get('swapping')
+                if (stall_limit and now - _fresh > 3 and not recent_attach and not tried_reattach
+                        and not sess.get('force_stop')
+                        and not sess.get('swapping')
                         and not sess.get('swap_source') and not sess.get('writing')
                         and not sess.get('output_failure')):
+                    tried_reattach = True
                     attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
                                                   read_timeout, retries, response, pump)
                     if attached:
                         response, pump = attached
                         continue
-                if stall_limit and now - _fresh > stall_limit and (not sess.get('swapping') or swap_for > stall_limit):
+                if (stall_limit and now - _fresh > stall_limit and
+                        (not recent_attach or sess.get('output_failure') or sess.get('writing')) and
+                        (not sess.get('swapping') or swap_for > stall_limit)):
                     # A full stdin pipe proves the source delivered a chunk but FFmpeg stopped
                     # consuming it. Reopening TikTok cannot repair a blocked RTMPS output.
                     if sess.get('writing') or sess.get('output_failure'):
                         self.last_failure_side = 'output'
                         sess['output_failure'] = True
-                    else:
+                    elif not tried_reattach:
                         attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
                                                       read_timeout, retries, response, pump)
                         if attached:
@@ -1051,10 +1099,10 @@ class Sender:
                     break
                 if now - progress['last'] > progress_limit:
                     raise SourceStalled('لم يتقدّم FFmpeg منذ %.0f ث' % (now - progress['last']))
-                if sess['ended'] and not sess['error']:
+                if sess['ended'] and not sess['error'] and not recent_attach:
                     self.log(user, '[MONITOR] أغلق تيك توك الاستجابة؛ الجلسة تُعاد من جديد')
                     break
-                if sess['error']:
+                if sess['error'] and not recent_attach:
                     raise SourceStalled(sess['error'])
                 if response is not None and now >= next_report:
                     # بمعدل مضبوط: لا logging لكل بايت (كان سيملأ السجل في دقائق).
@@ -1062,7 +1110,7 @@ class Sender:
                     self.log(user, '[WATCHDOG] المصدر يعمل؛ آخر بايت قبل %.0f ث، خرج %.1f ث'
                              % (now - sess['last'], max(0, progress['seconds'])))
                 if pump is not None and not pump.is_alive() and response is not None and not stop.is_set():
-                    if sess['force_stop'] or sess['error'] or sess['ended']:
+                    if sess['force_stop'] or ((sess['error'] or sess['ended']) and not recent_attach):
                         break
                 time.sleep(1)
         except (OSError, ValueError) as e:

test_source_reconnect.py

@@ -0,0 +1,295 @@
+#!/usr/bin/env python3
+"""Regression for live source handover; no real TikTok/YouTube requests.
+
+Starting a replacement pump while swap_source=True made the new thread exit before
+reading a single byte. FFmpeg was kept alive but YouTube received no replacement media.
+"""
+import io
+import http.server
+import json
+import shutil
+import subprocess
+import tempfile
+import os
+import pathlib
+import sys
+import threading
+import time
+import unittest
+from unittest.mock import Mock, patch
+
+ROOT = pathlib.Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else pathlib.Path(__file__).parent
+sys.path.insert(0, str(ROOT))
+os.environ.setdefault('DATA_DIR', tempfile.mkdtemp(prefix='handover-test-data-'))
+from relay import Sender
+from youtube_manager import ManagedYouTube
+
+
+class InlineThread:
+    """Deterministic scheduler: the new pump runs immediately when start() is called."""
+    def __init__(self, target, args=(), daemon=False):
+        self.target = target
+        self.args = args
+        self.daemon = daemon
+        self.started = False
+
+    def start(self):
+        self.started = True
+        self.target(*self.args)
+
+    def join(self, timeout=None):
+        pass
+
+    def is_alive(self):
+        return False
+
+
+class Response:
+    def __init__(self, chunks=(b'replacement media',)):
+        self.chunks = chunks
+        self.closed = False
+        self.reads = 0
+
+    def raise_for_status(self):
+        pass
+
+    def iter_content(self, chunk_size=None):
+        for chunk in self.chunks:
+            self.reads += 1
+            yield chunk
+
+    def close(self):
+        self.closed = True
+
+
+class HandoverTests(unittest.TestCase):
+    def setUp(self):
+        self.env = patch.dict(os.environ, {'SOURCE_STALL_TIMEOUT': '0'})
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.sender = Sender('/bin/true', lambda *a: None)
+        self.p = Mock()
+        self.p.poll.return_value = None
+        self.p.stdin = io.BytesIO()
+        self.old_pump = Mock()
+        self.old_pump.is_alive.return_value = False
+        self.old_response = Response()
+        self.new_response = Response()
+        self.stop = threading.Event()
+        self.sess = {'last': time.monotonic() - 4, 'max_seconds': 180,
+                     'force_stop': False, 'timeouts': 0, 'error': '', 'ended': True}
+        self.bridge = Mock()
+        self.bridge.feed.side_effect = lambda chunk: chunk
+        self.refresh = Mock(return_value={'url': 'https://example.invalid/source.flv', 'codec': 'h264', 'label': 'hd'})
+
+    def attach(self, refresh=None):
+        with patch('relay.requests.get', return_value=self.new_response), patch('relay.threading.Thread', InlineThread):
+            return self.sender._try_reattach('alice', self.p, self.bridge, self.sess,
+                self.stop, refresh or self.refresh, 10, 3, self.old_response, self.old_pump)
+
+    def test_new_pump_reads_media_even_when_scheduled_immediately(self):
+        attached = self.attach()
+        self.assertIsNotNone(attached)
+        self.assertEqual(self.new_response.reads, 1)
+        self.assertEqual(self.p.stdin.getvalue(), b'replacement media')
+        self.assertEqual(self.sender.snapshot()['source_bytes'], len(b'replacement media'))
+        self.assertFalse(self.sess['swap_source'])
+
+    def test_old_pump_must_stop_before_bridge_is_reused(self):
+        self.old_pump.is_alive.return_value = True
+        self.assertIsNone(self.attach())
+        self.bridge.reconnect.assert_not_called()
+        self.refresh.assert_not_called()
+        self.assertEqual(self.new_response.reads, 0)
+
+    def test_stop_during_resolution_does_not_resurrect_the_reader(self):
+        def stop_then_refresh():
+            self.stop.set()
+            return self.refresh()
+        self.assertIsNone(self.attach(stop_then_refresh))
+        self.bridge.reconnect.assert_not_called()
+        self.assertEqual(self.new_response.reads, 0)
+
+    def test_refresh_does_not_fabricate_source_heartbeat(self):
+        self.new_response = Response(chunks=())
+        previous = self.sess['last']
+        self.assertIsNotNone(self.attach())
+        self.assertEqual(self.sess['last'], previous)
+        self.assertEqual(self.sender.snapshot()['source_bytes'], 0)
+
+    def test_opened_source_has_bounded_first_byte_grace(self):
+        previous = self.sess['last']
+        now = time.monotonic()
+        self.new_response = Response(chunks=())
+        self.attach()
+        self.assertGreater(self.sess['reattach_not_before'], now)
+        self.assertLessEqual(self.sess['reattach_not_before'], time.monotonic() + 5)
+        self.assertEqual(self.sess['last'], previous)
+
+    def test_cancelled_reader_cannot_deliver_a_late_chunk(self):
+        cancel = threading.Event()
+        response = Response()
+        def late_chunk(_size=None):
+            cancel.set()
+            yield b'old reader late packet'
+        response.iter_content = late_chunk
+        with patch('relay.threading.Thread', InlineThread):
+            self.sender._pump_source(response, self.p, self.bridge, self.stop,
+                self.sess, 10, 3, cancel)
+        self.assertEqual(self.p.stdin.getvalue(), b'')
+        self.assertEqual(self.sender.snapshot()['source_bytes'], 0)
+
+    def test_no_source_is_a_failed_handover_not_a_fake_success(self):
+        self.refresh.return_value = None
+        self.assertIsNone(self.attach())
+        self.bridge.reconnect.assert_not_called()
+
+
+class CanvasTests(unittest.TestCase):
+    def test_placeholder_16x16_does_not_pin_the_live_canvas(self):
+        sender = Sender('/bin/true', lambda *a: None)
+        self.assertEqual(sender.lock_gold({'width': 16, 'height': 16}), (0, 0))
+        self.assertEqual(sender.lock_gold({'width': 720, 'height': 1440}), (720, 1280))
+
+    def test_unknown_or_placeholder_matchgop_uses_safe_constant_canvas(self):
+        for candidate in ({}, {'width': 16, 'height': 16}):
+            sender = Sender('/bin/true', lambda *a: None)
+            sender.mode_override = 'matchgop'
+            args = sender.command('test.flv', candidate)
+            vf = args[args.index('-vf') + 1]
+            self.assertIn('scale=w=720:h=1280:', vf)
+            # A later real configuration does not mutate the encoder canvas.
+            sender.lock_gold({'width': 960, 'height': 1920})
+            self.assertEqual((sender._gop_w, sender._gop_h), (720, 1280))
+
+    def test_normal_and_real_1080_source_dimensions_remain_supported(self):
+        for size in ((160, 120), (320, 240), (1080, 1920)):
+            sender = Sender('/bin/true', lambda *a: None)
+            self.assertEqual(sender.lock_gold({'width': size[0], 'height': size[1]}), size)
+
+
+class CachedSourceTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.manager = ManagedYouTube('/bin/true', lambda u, force=False: None,
+            lambda *a: None, lambda: ['alice'], pathlib.Path(self.tmp.name) / 'journal.json')
+        self.manager.video_sources = lambda sources: list(sources)
+        self.hd = {'url': 'https://example.invalid/hd.flv', 'media': 'flv', 'codec': 'h264',
+                   'quality_name': 'hd', 'width': 720, 'height': 1280, 'label': 'hd'}
+        self.sd = dict(self.hd, url='https://example.invalid/sd.flv', quality_name='sd',
+                       width=480, height=854, label='sd')
+        self.obs = {'status': 'live', 'room_id': 'same-room', 'sources': [self.hd, self.sd]}
+        self.manager._observe = Mock(return_value=self.obs)
+
+    def test_recent_compatible_source_is_reused_without_blocking_metadata_lookup(self):
+        self.manager._held['alice'] = (time.monotonic(), self.obs)
+        chosen = self.manager._refresh_flv('alice', self.hd)
+        self.assertEqual(chosen['url'], self.sd['url'])
+        self.manager._observe.assert_not_called()
+
+    def test_stale_source_snapshot_requires_fresh_lookup(self):
+        self.manager._held['alice'] = (time.monotonic() - 60, self.obs)
+        self.assertEqual(self.manager._refresh_flv('alice', self.hd)['url'], self.sd['url'])
+        self.manager._observe.assert_called_once_with('alice', force=True, bypass_throttle=True, allow_hold=False)
+
+    def test_cache_with_wrong_codec_cannot_be_spliced_into_open_publisher(self):
+        self.manager._held['alice'] = (time.monotonic(), dict(self.obs, sources=[dict(self.sd, codec='h265')]))
+        chosen = self.manager._refresh_flv('alice', self.hd)
+        self.assertEqual(chosen['codec'], 'h264')
+        self.manager._observe.assert_called_once()
+
+
+@unittest.skipUnless(shutil.which('ffmpeg') and shutil.which('ffprobe'), 'FFmpeg required')
+class RealHTTPContinuityTests(unittest.TestCase):
+    def run_continuity(self, mode):
+        with tempfile.TemporaryDirectory(prefix='source-handover-media-') as tmp:
+            root = pathlib.Path(tmp)
+            source = root / 'generated.flv'
+            output = root / 'output.flv'
+            subprocess.run([shutil.which('ffmpeg'), '-v', 'error', '-y', '-f', 'lavfi',
+                '-i', 'testsrc2=size=160x120:rate=20', '-f', 'lavfi', '-i', 'sine=sample_rate=48000',
+                '-t', '2', '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p',
+                '-g', '20', '-c:a', 'aac', '-f', 'flv', str(source)], check=True, capture_output=True)
+            media = source.read_bytes()
+            calls = []
+            class Handler(http.server.BaseHTTPRequestHandler):
+                def log_message(self, *a):
+                    pass
+                def do_GET(self):
+                    calls.append(time.monotonic())
+                    self.send_response(200)
+                    self.send_header('Content-Type', 'video/x-flv')
+                    self.send_header('Content-Length', str(len(media)))
+                    self.end_headers()
+                    # Header arrives immediately; replacements wait before their first byte.
+                    # Without startup grace the supervisor closes them before this media arrives.
+                    if len(calls) > 1:
+                        time.sleep(1)
+                    try:
+                        self.wfile.write(media)
+                        self.wfile.flush()
+                    except (OSError, BrokenPipeError):
+                        pass
+            server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+            server.daemon_threads = True
+            server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+            server_thread.start()
+            stop = threading.Event()
+            errors = []
+            candidate = {'url': 'http://127.0.0.1:%d/source.flv' % server.server_port,
+                         'media': 'flv', 'codec': 'h264', 'label': 'fixture', 'width': 160, 'height': 120}
+            sender = Sender(shutil.which('ffmpeg'), lambda *a: None)
+            sender.mode_override = mode
+            spawned = []
+            from relay import subprocess as relay_subprocess
+            real_popen = relay_subprocess.Popen
+            def track_process(*a, **kw):
+                child = real_popen(*a, **kw)
+                spawned.append(child)
+                return child
+            def send():
+                try:
+                    sender.send('fixture', candidate, str(output), stop, refresh=lambda: dict(candidate))
+                except Exception as exc:
+                    errors.append(exc)
+            env = {'SOURCE_STALL_TIMEOUT': '3', 'YOUTUBE_REATTACH_GRACE_SECONDS': '2',
+                   'YOUTUBE_SOURCE_READ_TIMEOUT': '5', 'YOUTUBE_PUMP_JOIN_SECONDS': '2',
+                   'YOUTUBE_SEND_MODE': mode, 'YOUTUBE_VIDEO_MODE': 'original',
+                   'YOUTUBE_AUTO_TRANSCODE': '0', 'YOUTUBE_X264_PRESET': 'ultrafast'}
+            thread = None
+            try:
+                with patch.dict(os.environ, env), patch('relay.subprocess.Popen', side_effect=track_process):
+                    thread = threading.Thread(target=send, daemon=True)
+                    thread.start()
+                    time.sleep(7.5)
+                    alive = thread.is_alive() and sender.alive()
+                    snapshot = sender.snapshot()
+                    stop.set()
+                    thread.join(6)
+                self.assertTrue(alive, 'Publisher ended instead of reusing the delayed replacement source: %r' % errors)
+                self.assertFalse(errors, repr(errors))
+                self.assertEqual(len(spawned), 1, 'Source reconnect must keep the same FFmpeg process')
+                self.assertGreaterEqual(len(calls), 3)
+                self.assertGreater(snapshot.get('output_seconds', 0), 4)
+                self.assertFalse(thread.is_alive(), 'Sender did not stop cleanly')
+                probe = subprocess.run([shutil.which('ffprobe'), '-v', 'error', '-show_entries',
+                    'format=duration', '-of', 'json', str(output)], check=True, capture_output=True, text=True)
+                self.assertGreater(float(json.loads(probe.stdout)['format']['duration']), 4)
+            finally:
+                stop.set()
+                sender.terminate()
+                if thread:
+                    thread.join(5)
+                server.shutdown()
+                server.server_close()
+
+    def test_eof_and_delayed_replacements_keep_same_copy_publisher(self):
+        self.run_continuity('copy')
+
+    def test_eof_and_delayed_replacements_keep_same_matchgop_publisher(self):
+        self.run_continuity('matchgop')
+
+
+if __name__ == '__main__':
+    unittest.main(argv=[sys.argv[0]], verbosity=2)

verify.sh

@@ -26,6 +26,7 @@ else
 fi
 python3 test_regressions.py "$ROOT"
 python3 test_relay.py "$ROOT"
+python3 test_source_reconnect.py "$ROOT"
 python3 test_flv_bridge.py "$ROOT"
 python3 test_completion.py "$ROOT"
 # Optional heavy tests when ffmpeg exists

youtube_manager.py

@@ -1215,12 +1215,27 @@ class ManagedYouTube:
         YouTube. After a long stable interval, a later natural refresh may restore the preferred
         rendition. HLS and H.264↔H.265 changes cannot be spliced into this open FFmpeg map.
         """
-        self._held.pop(user, None)
-        self._skip_hold.add(user)
-        try:
-            obs=self._observe(user, force=True, bypass_throttle=True, allow_hold=False)
-        except Exception:
-            return None
+        # A normal CDN EOF is not evidence that its signed URL expired. The manager
+        # already refreshes live observations continuously. Try a recent verified same-codec
+        # source first instead of blocking this publisher on 20s metadata requests while
+        # YouTube starves. Stale/absent/incompatible snapshots still force a fresh lookup.
+        obs=None
+        held=self._held.get(user)
+        family=self._codec_family(current)
+        if held and self.hold_seconds>0 and time.monotonic()-held[0]<=min(45.0,self.hold_seconds):
+            cached=held[1] or {}
+            if cached.get('status')=='live' and any(
+                    c.get('url') and c.get('media')!='hls' and
+                    (not family or self._codec_family(c)==family)
+                    for c in self.video_sources(cached.get('sources') or [])):
+                obs=cached
+        if obs is None:
+            self._held.pop(user, None)
+            self._skip_hold.add(user)
+            try:
+                obs=self._observe(user, force=True, bypass_throttle=True, allow_hold=False)
+            except Exception:
+                return None
         if (obs or {}).get('status')!='live':
             return None
         # We are already inside a three-second source gap. Probing every rendition serially can