fix(video): retry an rtsp source over tcp after a udp address-family failure

docs/architecture/video.md

@@ -36,6 +36,7 @@ Each plugin declares:
 - `is_available()` → `(bool, reason)`; defaults to `(True, "")`. Override when the plugin needs an OS or GStreamer element that isn't universally present (e.g. `v4l2src` on Linux, `avfvideosrc` on macOS, the `ndisrc` plugin from gst-plugin-ndi).
 - `source_element_name` → the **instance name** of the element that first produces bytes from the source (`"rtspsrc"`, `"udpsrc"`, `"media_source"`, …), or `None` for an input with no such boundary. The receiver probes it for the first buffer → `DATA_ARRIVING`. `TestSourceElementDeclaration` in [`tests/test_video_input_plugins.py`](../../tests/test_video_input_plugins.py) builds each plugin's pipeline and asserts the name resolves, so a rename inside `create_pipeline` can't silently detach the probe
 - `source_kind` → `SourceKind.REMOTE` (dials a host: rtsp, srt), `NAMED` (finds a source by name: ndi), `LISTENER` (waits at an address: rtp) or `LOCAL` (hardware or a file on this box). It picks the **wording** of the sentence and the action, so an NDI source is not told to check an address and port it does not have, and a USB camera is not told to check a sender. Defaults to `LOCAL`, which names no address, port or sender - a plugin that forgets is vague, never wrong
+- `on_bus_error(error)` → optional; the failed attempt's `BusError`, handed over in `_schedule_reconnect` before the next attempt is built. RTSP reads it to change the transport the retry asks for
 - `observe_progress(pipeline, report)` → optional; report protocol progress only that plugin can see. RTSP wires `rtspsrc::on-sdp` → `TRANSPORT_UP` + `STREAM_DESCRIBED`. The receiver keeps the furthest phase reported, so a duplicate or out-of-order report is harmless
 
 ## Receiver (`video/receiver.py`) – generic orchestrator
@@ -83,7 +84,9 @@ udpsrc → rtpjitterbuffer → pre_queue → decodebin → post_queue → videoc
 ```
 rtspsrc → decodebin → post_queue (leaky) → videoconvert → shared_videosink
 ```
-- `protocols` allows TCP + UDP + multicast, **except for a name dialled over IPv6, which gets TCP interleaved only**. `rtspsrc` opens its UDP ports for the family of the literal in the URL, so a name reads as IPv4 even when the control connection lands on IPv6 (a dual-stack server, `localhost`), and `udpsink` then refuses the server's address with `Invalid address family`. `_pin.dials_ipv6` asks the bounded resolver (`_RESOLVE_WAIT_S`) and reads the family of its first answer, the one the system's address selection (RFC 6724, against its own routes) puts first and the connection dials, so a dual-stack camera reached over IPv4 keeps UDP and multicast; a name still resolving keeps every transport this attempt and the next attempt reads the answer. An IPv6 literal is left to the element, which reads its family itself
+- `protocols` allows TCP + UDP + multicast, **except for a name dialled over IPv6 and a host whose attempt failed on the address family, which get TCP interleaved only**. `rtspsrc` opens a stream's UDP ports for the address type of the server's session description (`c=IN IP4` / `IP6`, IPv4 when there is none), not for the family the control connection reached the server on, so a server reached over IPv6 that describes IPv4 (or nothing) gets IPv4 ports, and the RTCP `udpsink` then refuses its address: `gst-resource-error-quark:13`, `Invalid address family` in the debug string, posted as a pipeline error after `on-sdp`. Verified against 1.28.7 on loopback with a stub server, for a name and an IPv6 literal alike; the 1.26 source picks the family the same way (`gst_rtspsrc_do_stream_connection`)
+  - **Up front:** `_pin.dials_ipv6` asks the bounded resolver (`_RESOLVE_WAIT_S`) and reads the family of its first answer, the one the system's address selection (RFC 6724, against its own routes) puts first and the connection dials, so a dual-stack camera reached over IPv4 keeps UDP and multicast; a name still resolving keeps every transport this attempt and the next attempt reads the answer. An address literal is not looked up
+  - **After the fact:** the connection can land on the other family anyway (the first address refused or unreachable, the next one dialled; an IPv6 literal whose server describes IPv4). `on_bus_error` reads `Invalid address family` in the failed attempt's debug string and keeps that host on TCP interleaved for every later attempt and heal of this input; a changed URL host, or an input rebuilt by a saved change to the source's own settings or type, tries every transport again
 - `tcp-timeout` 10 s so the element gives up before the connection timeout, `latency` 0, `drop-on-latency`, `buffer-mode` none; only the video track is linked into `decodebin`, since an NVR may list its audio track first
 
 ## Stream credentials (RTSP / SRT)
@@ -179,7 +182,10 @@ against.
 bad RTSP path comes back as
 `gst-resource-error-quark:13` with `SDP contains no streams`, not
 `RESOURCE_NOT_FOUND`. Check what the element actually emits before adding a
-mapping - a code that looks obvious from the enum may never be sent.
+mapping - a code that looks obvious from the enum may never be sent. RTSP's
+UDP address-family mismatch is that same code with `Invalid address family` in
+the debug string; it reads by the phase like an open failure (after `on-sdp`,
+`NO_DATA`), and the retry runs over TCP interleaved (see the RTSP plugin).
 
 **`INTERFACE_DOWN` and `WRONG_INTERFACE` never come from the bus.** Only the
 video input pin's preflight produces them, before anything is dialled, so

openfollow/video/failure.py

@@ -15,6 +15,7 @@
     "ConnectionPhase",
     "SourceKind",
     "VideoFailure",
+    "address_family_mismatch",
     "classify_failure",
     "failure_action",
     "failure_chip",
@@ -107,6 +108,11 @@ class VideoFailure(Enum):
 # settings failure. The wording is gstrtspsrc's own, not a camera's.
 _EMPTY_SDP_MARKERS = ("sdp contains no streams", "no streams in sdp")
 
+# A UDP sink refusing a destination of the other IP family than its socket's:
+# rtspsrc opened its ports for the family the server's session description gave,
+# and the connection reached the server on the other.
+_ADDRESS_FAMILY_MARKERS = ("invalid address family",)
+
 # A source asked for a format, size or frame rate it cannot deliver stops with
 # the generic stream failure; only the flow return in the debug string says so.
 _NOT_NEGOTIATED_MARKERS = ("not-negotiated",)
@@ -292,6 +298,12 @@ def failure_action(failure: VideoFailure, *, kind: SourceKind = SourceKind.REMOT
     return _ACTIONS.get(failure, _ACTIONS[VideoFailure.UNKNOWN])
 
 
+def address_family_mismatch(text: str) -> bool:
+    """Whether an error's *text* (message or debug string) names a UDP address-family mismatch."""
+    lowered = text.lower()
+    return any(marker in lowered for marker in _ADDRESS_FAMILY_MARKERS)
+
+
 def classify_failure(
     *,
     phase: ConnectionPhase,
@@ -326,6 +338,8 @@ def classify_failure(
             return VideoFailure.DEVICE_UNAVAILABLE
         if code == RESOURCE_SETTINGS and any(marker in text for marker in _EMPTY_SDP_MARKERS):
             return VideoFailure.STREAM_NOT_FOUND
+        if code == RESOURCE_SETTINGS and address_family_mismatch(text):
+            return _silence_verdict(phase, saw_video)
         if code in _OPEN_CODES:
             if any(marker in text for marker in _REFUSED_MARKERS):
                 return VideoFailure.REFUSED

openfollow/video/inputs/_base.py

@@ -20,6 +20,7 @@
 
 from openfollow.net_egress import is_loopback_host
 from openfollow.net_utils import plane_source_iface
+from openfollow.runtime.receiver_bus import BusError
 from openfollow.video.failure import ConnectionPhase, SourceKind
 from openfollow.video.inputs._pin import PinRefusal
 
@@ -243,6 +244,9 @@ def on_bus_segment_done(self, pipeline: Any) -> bool:
         next iteration of a seamless loop. The default reports it unhandled."""
         return False
 
+    def on_bus_error(self, error: BusError) -> None:  # noqa: B027 – intentional optional hook
+        """Called with a failed attempt's error before the next attempt is built."""
+
     def on_caps_changed(self, width: int, height: int) -> None:  # noqa: B027 – intentional optional hook
         """Called when video resolution is detected from caps-changed."""
 

openfollow/video/inputs/_pin.py

@@ -85,9 +85,9 @@ def dials_ipv6(host: str) -> bool:
 
     The connection dials the resolver's first answer, which the system orders
     against its own routes, so a dual-stack name preferred over IPv4 is not one.
-    An IP literal is never one: the element reads a literal's family itself. A
-    name whose lookup is still running reads as not one this attempt; the answer
-    serves the next one.
+    An IP literal is never one: it is not looked up, and the RTSP input moves a
+    host to TCP once an attempt fails on the address family. A name whose lookup
+    is still running reads as not one this attempt; the answer serves the next one.
     """
     if not host:
         return False

openfollow/video/inputs/rtsp.py

@@ -9,8 +9,9 @@
 from typing import Any
 from urllib.parse import urlsplit
 
+from openfollow.runtime.receiver_bus import BusError
 from openfollow.uri_redaction import redact_uri, strip_uri_userinfo
-from openfollow.video.failure import ConnectionPhase, SourceKind
+from openfollow.video.failure import ConnectionPhase, SourceKind, address_family_mismatch
 from openfollow.video.inputs._base import (
     ConfigField,
     InputCapabilities,
@@ -75,6 +76,13 @@ class RtspInput(VideoInputBase):
     source_kind = SourceKind.REMOTE
     pins_interface = True
 
+    def __init__(self) -> None:
+        super().__init__()
+        # The host the last build dialled, and one whose UDP ports met a server
+        # address of the other IP family, which stays on TCP interleaved.
+        self._dialled_host = ""
+        self._interleaved_host = ""
+
     # -- Declarations ---------------------------------------------------------
 
     @classmethod
@@ -182,13 +190,20 @@ def create_pipeline(
         rtspsrc.set_property("latency", 0)
         rtspsrc.set_property("drop-on-latency", True)
         rtspsrc.set_property("buffer-mode", 0)  # none – lowest latency
-        # rtspsrc opens its UDP ports for the family of the literal in the URL, so a
-        # name reads as IPv4 even when the connection lands on IPv6 and udpsink then
-        # refuses the server's address. A name dialled over IPv6 takes the interleaved
-        # transport; everything else may negotiate TCP, UDP or multicast.
+        # rtspsrc opens its UDP ports for the address type of the server's session
+        # description (IPv4 when it names none), whatever family the connection
+        # reached the server on, and udpsink then refuses an address of the other
+        # family. A name dialled over IPv6 takes the interleaved transport up front,
+        # a host whose attempt failed that way from then on; everything else may
+        # negotiate TCP, UDP or multicast.
         endpoint = self.source_endpoint(config)
-        tcp_only = endpoint is not None and dials_ipv6(endpoint.host)
-        rtspsrc.set_property("protocols", _TRANSPORTS_TCP if tcp_only else _TRANSPORTS_ALL)
+        self._dialled_host = endpoint.host if endpoint is not None else ""
+        tcp_reason = ""
+        if self._dialled_host and self._dialled_host == self._interleaved_host:
+            tcp_reason = "an earlier attempt's UDP ports were of the other IP family"
+        elif endpoint is not None and dials_ipv6(endpoint.host):
+            tcp_reason = "the name is dialled over IPv6"
+        rtspsrc.set_property("protocols", _TRANSPORTS_TCP if tcp_reason else _TRANSPORTS_ALL)
         pin = config_pin(config)
         if pin:
             # Only the multicast membership can be pinned; the preflight checks
@@ -197,7 +212,7 @@ def create_pipeline(
         logger.info(
             "RTSP source: %s (latency=0, %s, login=%s)",
             redact_uri(location),
-            "tcp interleaved: the name is dialled over IPv6" if tcp_only else "tcp+udp+multicast",
+            f"tcp interleaved: {tcp_reason}" if tcp_reason else "tcp+udp+multicast",
             "set" if (user or password) else "none",
         )
 
@@ -285,6 +300,12 @@ def on_bus_async_done(self, pipeline: Any) -> None:
         pipeline.set_latency(0)
         logger.info("Pipeline ASYNC_DONE (RTSP) -- latency forced to 0")
 
+    def on_bus_error(self, error: BusError) -> None:
+        """Keep the dialled host on TCP interleaved once its UDP ports met an address of the other family."""
+        if address_family_mismatch(error.debug):
+            self._interleaved_host = self._dialled_host
+            logger.warning("RTSP UDP ports did not match the server's IP family; using TCP interleaved from now on")
+
     def observe_progress(self, pipeline: Any, report: PhaseReporter) -> None:
         """Report the SDP: the server answered and described its media."""
         rtspsrc = pipeline.get_by_name("rtspsrc")

openfollow/video/receiver.py

@@ -1149,6 +1149,11 @@ def _schedule_reconnect(
         # reason, and published with the verdict it explains.
         failure = self._classify(error) if failure is None else failure
         phase = self._state.phase
+        if error is not None:
+            try:
+                self._input.on_bus_error(error)
+            except Exception:
+                logger.exception("Error handing the failure to the %s input", self._input.display_name)
         self._cancel_connection_timeout()
         self._cancel_reconnect()
         self._cancel_heal()

tests/_fake_gst.py

@@ -26,6 +26,7 @@ def __init__(self, name: str = "pad", *, link_returns: str = "ok") -> None:
         self._link_returns = link_returns
         self.linked_to: FakePad | None = None
         self._linked_into = False
+        self.probes: list[tuple[Any, Any]] = []
 
     def get_name(self) -> str:
         return self._name
@@ -39,6 +40,10 @@ def link(self, other: FakePad) -> str:
             other._linked_into = True
         return self._link_returns
 
+    def add_probe(self, probe_type: Any, callback: Any) -> int:
+        self.probes.append((probe_type, callback))
+        return len(self.probes)
+
     def get_current_caps(self) -> FakeCaps:
         return FakeCaps(f"video/{self._name}")
 

tests/test_video_failure_classification.py

@@ -346,6 +346,53 @@ def test_it_reads_the_path_not_a_dropped_feed(self) -> None:
         )
 
 
+class TestAnAddressFamilyMismatchIsNotUnknown:
+    """rtspsrc's UDP ports opened for one IP family while the server's address
+    is the other: the RTCP udpsink refuses it with the generic settings code,
+    and only the debug string names the cause."""
+
+    _MESSAGE = "Could not get/set settings from/on resource."
+    _DEBUG = (
+        "../gst/udp/gstmultiudpsink.c(1229): gst_multiudpsink_configure_client (): "
+        "/GstPipeline:rtsp-sink/GstRTSPSrc:rtspsrc/GstUDPSink:udpsink0:\nInvalid address family (got 10)"
+    )
+
+    @pytest.mark.parametrize(
+        ("phase", "expected"),
+        [
+            (ConnectionPhase.STARTING, VideoFailure.UNREACHABLE),
+            (ConnectionPhase.STREAM_DESCRIBED, VideoFailure.NO_DATA),
+            (ConnectionPhase.DATA_ARRIVING, VideoFailure.NO_DATA),
+            (ConnectionPhase.DECODING, VideoFailure.STALLED),
+        ],
+    )
+    def test_it_reads_by_how_far_the_feed_got(self, phase: ConnectionPhase, expected: VideoFailure) -> None:
+        verdict = classify_failure(
+            phase=phase, domain=RESOURCE_DOMAIN, code=RESOURCE_SETTINGS, message=self._MESSAGE, debug=self._DEBUG
+        )
+        assert verdict == expected
+
+    def test_the_marker_is_matched_case_insensitively(self) -> None:
+        verdict = classify_failure(
+            phase=ConnectionPhase.STREAM_DESCRIBED,
+            domain=RESOURCE_DOMAIN,
+            code=RESOURCE_SETTINGS,
+            message=self._MESSAGE,
+            debug="INVALID ADDRESS FAMILY (got 30)",
+        )
+        assert verdict == VideoFailure.NO_DATA
+
+    @pytest.mark.parametrize(
+        ("domain", "code"),
+        [(STREAM_DOMAIN, STREAM_FAILED), ("some-other-quark", RESOURCE_SETTINGS), (RESOURCE_DOMAIN, 1)],
+    )
+    def test_the_wording_under_another_code_is_not_claimed(self, domain: str, code: int) -> None:
+        verdict = classify_failure(
+            phase=ConnectionPhase.STREAM_DESCRIBED, domain=domain, code=code, message=self._MESSAGE, debug=self._DEBUG
+        )
+        assert verdict == VideoFailure.UNKNOWN
+
+
 class TestSourceBytesAreNotDecodedVideo:
     """``DATA_ARRIVING`` is bytes out of the source element. A feed carrying a
     payload this station cannot decode produces them just as readily as a