Warn when an outbound frame push fails

noxfile.py

@@ -63,4 +63,32 @@ def integration(session: nox.Session) -> None:
     notably the ``reactor_webrtc`` wheel — present rather than skipped.
     """
     _install_locked(session)
-    session.run("pytest", "-q", "tests/integration", *session.posargs)
+    # Diagnostic settings while an integration hang is being chased in CI: -v names
+    # each test as it starts, -s keeps output unbuffered so the last line before a
+    # stall is real, --log-cli-level streams the runtime's own logging, and the
+    # faulthandler timeout dumps every thread's stack and aborts rather than
+    # letting the job sit until the runner is reclaimed.
+    # The deadlock probe runs in its own invocation, and first: faulthandler
+    # aborts the whole process on a stall, so anything collected after the
+    # loopback never gets to report. Its own failure must not stop the suite from
+    # running, hence the accepted exit codes.
+    session.run(
+        "pytest",
+        "-v",
+        "-s",
+        "--log-cli-level=INFO",
+        "tests/integration/transport/webrtc/test_gil_signalling_deadlock.py",
+        success_codes=[0, 1, 2, 3, 4, 5],
+    )
+    session.run(
+        "pytest",
+        "-v",
+        "-s",
+        "-o",
+        "faulthandler_timeout=120",
+        "-o",
+        "faulthandler_exit_on_timeout=true",
+        "--log-cli-level=INFO",
+        "tests/integration",
+        *session.posargs,
+    )

src/reactor_runtime/transport/webrtc/peer.py

@@ -108,6 +108,8 @@
 # Outbound frame queue depth: a bundle that arrives when the drain thread is
 # behind is dropped rather than allowed to grow unbounded latency.
 _FRAME_QUEUE_MAX = 10
+# How often a still-failing outbound push repeats its warning, in frames.
+_PUSH_FAILURE_LOG_EVERY = 300
 
 # Outbound audio is 48 kHz mono, matching the runtime's audio frames and the
 # rate the synthetic audio device plays out at. The device takes one 10 ms frame
@@ -234,6 +236,9 @@ def __init__(self) -> None:
         # audio thread feeds that buffer to this peer's LocalAudioSource track in
         # steady 10 ms frames via track.push_pcm().
         self._frame_queue: queue.Queue[MediaBundle] = queue.Queue(maxsize=_FRAME_QUEUE_MAX)
+        # Outbound pushes that raised, counted so a persistent fault is reported
+        # without one line per frame.
+        self._push_failures = 0
         self._frame_thread: threading.Thread | None = None
         self._audio_track: rw.Track | None = None
         self._audio_buf: npt.NDArray[np.int16] = np.array([], dtype=np.int16)
@@ -519,7 +524,18 @@ def _frame_drain_loop(self) -> None:
             try:
                 self._push_bundle(bundle)
             except Exception:
-                logger.debug("outbound frame push failed", exc_info=True)
+                self._push_failures += 1
+                # A push that fails takes the model's output off the wire while the
+                # device keeps filling the gap, so the session looks alive and only
+                # the content is missing. That is worth a warning. The first one
+                # carries the traceback and the rest are counted, because this runs
+                # per frame — repeating it would bury the log at frame rate.
+                if self._push_failures == 1:
+                    logger.warning("outbound frame push failed", exc_info=True)
+                elif self._push_failures % _PUSH_FAILURE_LOG_EVERY == 0:
+                    logger.warning(
+                        "outbound frame push still failing (%d frames)", self._push_failures
+                    )
 
     def _push_bundle(self, bundle: MediaBundle) -> None:
         """Push a bundle's video and buffer its audio for the feeder thread.

tests/integration/transport/webrtc/test_gil_signalling_deadlock.py

@@ -0,0 +1,86 @@
+"""Confirm on the runner why the loopback stops inside ``create_peer_connection``.
+
+The loopback stalls there and never returns: the last log line before the
+faulthandler dump is the one immediately preceding the call, and the only other
+thread in the dump has a Python thread state with no Python frame — a native
+thread waiting to enter Python.
+
+That is a deadlock between the GIL and libwebrtc's signalling thread. The
+installed ``reactor_webrtc`` holds the GIL while it creates a peer connection,
+and creating one is a proxy call: it is posted to the signalling thread and the
+caller blocks until it finishes. The loopback's stand-in client is gathering ICE
+candidates at that exact moment, and each candidate is delivered by the
+signalling thread into a Python callback. So the creating thread waits for the
+signalling thread while the signalling thread waits for the GIL, and neither can
+be interrupted, because the creating thread is in native code and never yields.
+
+This test drives that collision directly, with no tracks, negotiation, media or
+frame metadata involved — a candidate callback parked in Python, then one call.
+It runs in a subprocess, because the deadlock cannot be observed from inside the
+process that is stuck: a watchdog thread would need the GIL the stuck caller
+holds. A timeout turns the hang into a failure.
+
+Failing here means the deadlock is in the binding rather than in anything this
+package does with it. It is a probe, not a regression test — the fix and its
+permanent test belong to ``reactor-webrtc``, and this file goes away with them.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+pytest.importorskip("reactor_webrtc")
+
+_TIMEOUT_S = 45
+
+_PROBE = textwrap.dedent("""
+    import asyncio, threading, time
+    import reactor_webrtc as rw
+
+    factory = rw.PeerConnectionFactory()
+
+    # Fires on the signalling thread. The sleep releases the GIL, so the main
+    # thread is free to take it, and the callback then needs it back to return
+    # into native code — the state a caller holding the GIL traps it in.
+    entered = threading.Event()
+
+    def hold_the_signalling_thread(_candidate):
+        entered.set()
+        time.sleep(2.0)
+
+    async def main():
+        observer = rw.PeerConnectionObserver()
+        observer.on_ice_candidate = hold_the_signalling_thread
+        first = factory.create_peer_connection(rw.RtcConfiguration(), observer)
+        first.add_transceiver(rw.MediaKind.Video, rw.TransceiverDirection.RecvOnly)
+        offer = await first.create_offer()
+        await first.set_local_description(offer)
+        assert entered.wait(30), "no ICE candidate was delivered"
+        factory.create_peer_connection(rw.RtcConfiguration(), rw.PeerConnectionObserver())
+
+    asyncio.run(main())
+    print("returned")
+""")
+
+
+def test_creating_a_peer_connection_returns_while_a_candidate_callback_is_in_flight() -> None:
+    """Create a peer connection with the signalling thread parked in Python."""
+    try:
+        done = subprocess.run(
+            [sys.executable, "-c", _PROBE],
+            capture_output=True,
+            text=True,
+            timeout=_TIMEOUT_S,
+        )
+    except subprocess.TimeoutExpired:
+        pytest.fail(
+            f"create_peer_connection did not return in {_TIMEOUT_S}s: the binding holds "
+            f"the GIL across the dispatch to the signalling thread, which is waiting "
+            f"for the GIL. This is what stalls the loopback."
+        )
+    assert done.returncode == 0, f"probe exited {done.returncode}:\n{done.stderr}"
+    assert "returned" in done.stdout

tests/unit/transport/webrtc/test_peer.py

@@ -7,6 +7,9 @@
 """
 
 import asyncio
+import logging
+import threading
+import time
 from types import SimpleNamespace
 from typing import Any, cast
 
@@ -425,3 +428,36 @@ async def test_factory_rejects_empty_offer() -> None:
         await libwebrtc_peer_factory(
             ConnId(1), SdpOffer(sdp="   "), TrackMap(), WebRtcConfig(), ProtocolVersion.V0
         )
+
+
+def test_a_failing_outbound_push_is_warned_not_swallowed(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The drain loop reports a push that raises, with its traceback.
+
+    Swallowing it takes the model's output off the wire while the device keeps
+    filling the gap, so the session looks healthy and only the content is
+    missing — the one failure here that a debug line is not enough for.
+    """
+    peer = WebRTCPeer()
+    # A track object with no push_video_frame, so the push raises inside the loop.
+    peer._out_tracks["v"] = cast(Any, object())
+    peer._frame_queue.put_nowait(_video_bundle("v"))
+
+    thread = threading.Thread(target=peer._frame_drain_loop, daemon=True)
+    with caplog.at_level(logging.WARNING):
+        thread.start()
+        try:
+            deadline = time.monotonic() + 5.0
+            while time.monotonic() < deadline and peer._push_failures == 0:
+                time.sleep(0.01)
+        finally:
+            peer._stop_event.set()
+            thread.join(timeout=5.0)
+
+    assert peer._push_failures >= 1, "the drain loop never attempted the push"
+    warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
+    assert any("outbound frame push failed" in r.getMessage() for r in warnings), (
+        f"the failure was not warned about: {[r.getMessage() for r in warnings]}"
+    )
+    assert any(r.exc_info for r in warnings), "the warning carried no traceback"