tst-udp: fire-and-forget unconnected sender + IPv4-first hostname resolution

CHANGELOG.md

@@ -41,9 +41,25 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   non-ascending pair, companion overflow, or token count (target
   `tst_rtp::rtsp::transport_negotiation`, `debug!` level). Wire-supplied
   values are Debug-escaped before logging.
+- **`TransportError::is_connection_refused()`** — portable refused-
+  classification without hard-coding platform errnos (111 Linux / 61 BSD /
+  10061 Windows); classifies via the std `io::ErrorKind` mapping over
+  `Backpressure`/`Broken`'s `errno_code`. `std`-gated (the type stays
+  no_std-buildable).
 
 ### Changed
 
+- **`UdpTransport` now sends on an unconnected socket (`send_to`)**, from
+  an integrator field report reproduced against a live media server: a
+  transient ICMP port-unreachable from a restarting or idle receiver no
+  longer kills the sender with a fatal `Broken`/ECONNREFUSED. Fire-and-
+  forget datagram semantics, matching every TS-over-UDP sender (ffmpeg,
+  VLC, mediamtx). No knob — this is the only mode.
+- **`udp://` hostname resolution now prefers IPv4 among reachable-family
+  candidates.** A UDP connect-probe can reject an unconfigured/unroutable
+  address family but cannot detect an absent listener, so `localhost`
+  resolving `[::1, 127.0.0.1]` on a dual-stack host previously picked
+  `::1` by resolver order and died against an IPv4-only listener.
 - **Breaking (Rust, pre-1.0): `UdpUrlError::BadHost` is removed**, superseded
   by the hostname-resolution work above. Unresolvable or junk hosts now
   surface as `UdpUrlError::HostResolve { host, detail }`, and passing an

crates/tst-core/public-api.txt

@@ -6851,6 +6851,8 @@ pub tst_core::transport::TransportError::ExplicitClose
 pub tst_core::transport::TransportError::TooLarge
 pub tst_core::transport::TransportError::TooLarge::len: usize
 pub tst_core::transport::TransportError::TooLarge::max: usize
+impl tst_core::transport::TransportError
+pub fn tst_core::transport::TransportError::is_connection_refused(&self) -> bool
 impl core::clone::Clone for tst_core::transport::TransportError
 pub fn tst_core::transport::TransportError::clone(&self) -> tst_core::transport::TransportError
 impl core::cmp::Eq for tst_core::transport::TransportError
@@ -7407,6 +7409,8 @@ pub tst_core::TransportError::ExplicitClose
 pub tst_core::TransportError::TooLarge
 pub tst_core::TransportError::TooLarge::len: usize
 pub tst_core::TransportError::TooLarge::max: usize
+impl tst_core::transport::TransportError
+pub fn tst_core::transport::TransportError::is_connection_refused(&self) -> bool
 impl core::clone::Clone for tst_core::transport::TransportError
 pub fn tst_core::transport::TransportError::clone(&self) -> tst_core::transport::TransportError
 impl core::cmp::Eq for tst_core::transport::TransportError

crates/tst-core/src/transport.rs

@@ -183,6 +183,50 @@ pub enum TransportError {
     ExplicitClose,
 }
 
+#[cfg(feature = "std")]
+impl TransportError {
+    /// True when this error carries an OS errno identifying a refused
+    /// connection (ICMP port-unreachable surfaced as `ECONNREFUSED`).
+    /// Uses the std `io::ErrorKind` mapping so callers never hard-code the
+    /// platform errno split (111 Linux / 61 BSD / 10061 Windows).
+    #[must_use]
+    pub fn is_connection_refused(&self) -> bool {
+        let errno = match self {
+            Self::Backpressure { errno_code, .. } | Self::Broken { errno_code, .. } => *errno_code,
+            _ => None,
+        };
+        errno.is_some_and(|c| {
+            std::io::Error::from_raw_os_error(c).kind() == std::io::ErrorKind::ConnectionRefused
+        })
+    }
+}
+
+#[cfg(all(test, feature = "std"))]
+mod error_tests {
+    use super::*;
+
+    #[test]
+    fn is_connection_refused_classifies_portably() {
+        #[cfg(target_os = "linux")]
+        let refused: i32 = 111; // ECONNREFUSED
+        #[cfg(any(target_os = "macos", target_os = "freebsd", target_os = "openbsd"))]
+        let refused: i32 = 61;
+        #[cfg(windows)]
+        let refused: i32 = 10061; // WSAECONNREFUSED
+        let e = TransportError::Broken {
+            msg: "send error".into(),
+            errno_code: Some(refused),
+        };
+        assert!(e.is_connection_refused());
+        let none = TransportError::Broken {
+            msg: "x".into(),
+            errno_code: None,
+        };
+        assert!(!none.is_connection_refused());
+        assert!(!TransportError::Closed.is_connection_refused());
+    }
+}
+
 /// One-shot byte transport. Each `send_bytes` call sends exactly one
 /// outbound message.
 ///

crates/tst-udp/src/transport.rs

@@ -18,8 +18,14 @@ use crate::url::UdpUrl;
 /// [`crate::builder::UdpTransportBuilder`] (added in a later phase) for full
 /// control over knobs.
 ///
-/// Always represents a single peer (set via `connect()` on the underlying
-/// std socket). To send to a different peer, build a new transport.
+/// Always sends to a single fixed peer via `send_to` on an
+/// **unconnected** socket — fire-and-forget datagram semantics, matching
+/// what TS-over-UDP receivers (ffmpeg, VLC, mediamtx) expect of a
+/// sender. Deliberately NOT a connected socket: on Linux a connected UDP
+/// socket surfaces ICMP port-unreachable as a fatal `ECONNREFUSED` on a
+/// later `send`, which turns a receiver's restart/idle-rebind window
+/// into a dead sender. To send to a different peer, build a new
+/// transport.
 pub struct UdpTransport {
     socket: UdpSocket,
     pkt_size: usize,
@@ -66,7 +72,6 @@ impl UdpTransport {
         apply_socket2_knobs(&socket, cfg).map_err(UdpError::Io)?;
 
         let peer = SocketAddr::new(url.addr, url.port);
-        socket.connect(peer).map_err(UdpError::Io)?;
 
         Ok(Self {
             socket,
@@ -110,7 +115,7 @@ impl Transport for UdpTransport {
                 max: self.pkt_size,
             });
         }
-        match self.socket.send(msg) {
+        match self.socket.send_to(msg, self.peer) {
             Ok(_n) => {
                 self.stats.datagrams_sent = self.stats.datagrams_sent.saturating_add(1);
                 self.stats.bytes_sent = self.stats.bytes_sent.saturating_add(msg.len() as u64);

crates/tst-udp/src/url.rs

@@ -75,8 +75,9 @@ impl UdpUrl {
     ///
     /// The host may be an IP literal or a DNS hostname. Non-literal
     /// hosts are resolved here via the system resolver
-    /// ([`std::net::ToSocketAddrs`]) and the first returned address is
-    /// used — on dual-stack hosts that may be the IPv6 address. The
+    /// ([`std::net::ToSocketAddrs`]); among multiple results, IPv4 is
+    /// preferred among the candidates that probe clean (see the internal
+    /// `resolve_host` doc comment for the full tiebreak rationale). The
     /// `?localaddr=` and IPv4 `?iface=` values stay literal-only
     /// (resolving a local NIC selector through DNS is meaningless).
     pub fn parse(s: &str) -> Result<Self, UdpUrlError> {
@@ -170,10 +171,14 @@ impl UdpUrl {
 /// `bind` + `connect` (no packets are sent; a UDP `connect` only sets the
 /// default destination), which fails fast for an address family that is
 /// unconfigured or unroutable on this host (e.g. an AAAA record arriving
-/// first while IPv6 is disabled). The first candidate that probes clean
-/// wins; if every probe fails, fall back to the first resolved address so
-/// the real send path surfaces the OS error — never worse than not
-/// probing at all.
+/// first while IPv6 is disabled). The probe only rejects unconfigured or
+/// unroutable families though — it cannot detect an absent listener — so
+/// among probe-clean candidates IPv4 is preferred (the dual-stack
+/// `localhost` trap: `[::1, 127.0.0.1]` both probe clean, but picking
+/// `::1` dies against an IPv4-only listener; this matches the dominant
+/// TS-over-UDP tooling). If every probe fails, fall back to the first
+/// resolved address so the real send path surfaces the OS error — never
+/// worse than not probing at all.
 fn resolve_host(host: &str, port: u16) -> Result<IpAddr, UdpUrlError> {
     use std::net::{SocketAddr, ToSocketAddrs, UdpSocket};
 
@@ -191,18 +196,26 @@ fn resolve_host(host: &str, port: u16) -> Result<IpAddr, UdpUrlError> {
             host: host.to_string(),
             detail: "resolved to no addresses".to_string(),
         })?;
+    let mut first_clean: Option<SocketAddr> = None;
     for sa in &candidates {
         let unspec: SocketAddr = if sa.is_ipv4() {
             (std::net::Ipv4Addr::UNSPECIFIED, 0).into()
         } else {
             (std::net::Ipv6Addr::UNSPECIFIED, 0).into()
         };
-        if let Ok(probe) = UdpSocket::bind(unspec) {
-            if probe.connect(sa).is_ok() {
-                return Ok(sa.ip());
+        let Ok(probe) = UdpSocket::bind(unspec) else {
+            continue;
+        };
+        if probe.connect(sa).is_ok() {
+            if sa.is_ipv4() {
+                return Ok(sa.ip()); // documented preference: first clean IPv4
             }
+            first_clean.get_or_insert(*sa);
         }
     }
+    if let Some(sa) = first_clean {
+        return Ok(sa.ip());
+    }
     Ok(first.ip())
 }
 
@@ -343,6 +356,35 @@ mod tests {
         assert!(!u.recv_bind);
     }
 
+    /// P3 residual: `localhost` resolves `[::1, 127.0.0.1]` on dual-stack
+    /// hosts and a UDP connect-probe cannot detect an absent listener, so
+    /// without an explicit preference the sender picks `::1` and dies
+    /// against an IPv4-only listener. Documented preference: IPv4 first
+    /// among probe-clean candidates.
+    #[test]
+    fn hostname_resolution_prefers_ipv4() {
+        // IPv6-only hosts have no IPv4 localhost record to prefer — the
+        // contract is "IPv4 first when present", not "IPv4 required".
+        use std::net::ToSocketAddrs;
+        let has_ipv4 = ("localhost", 5004u16)
+            .to_socket_addrs()
+            .map(|mut addrs| addrs.any(|a| a.is_ipv4()))
+            .unwrap_or(false);
+        if !has_ipv4 {
+            eprintln!(
+                "skipping hostname_resolution_prefers_ipv4: host has no IPv4 localhost record"
+            );
+            return;
+        }
+
+        let u = UdpUrl::parse("udp://localhost:5004").unwrap();
+        assert!(
+            u.addr.is_ipv4(),
+            "expected IPv4-first for localhost, got {}",
+            u.addr
+        );
+    }
+
     #[test]
     fn hostname_recv_bind_resolves() {
         let u = UdpUrl::parse("udp://@localhost:5004").unwrap();

crates/tst-udp/tests/loopback_unicast.rs

@@ -40,17 +40,23 @@ fn unicast_loopback_sends_payload_to_std_recv() {
 #[test]
 fn unicast_loopback_sends_via_hostname_url() {
     // The containerized-consumer path: `udp://<name>:<port>` where <name>
-    // is a resolvable hostname, not an IP literal. Bind the receiver on
-    // whatever address the system resolver returns FIRST for localhost —
-    // the same selection UdpUrl::parse uses — so the test holds whether
-    // the resolver prefers ::1 or 127.0.0.1.
+    // is a resolvable hostname, not an IP literal. `UdpUrl::parse`
+    // deterministically prefers IPv4 among probe-clean candidates (the
+    // dual-stack `localhost` tiebreak — see `resolve_host`'s doc comment),
+    // so bind the receiver the same way: prefer IPv4, but fall back to the
+    // first resolved candidate on an IPv6-only host (production doesn't
+    // require IPv4, only prefer it).
     use std::net::ToSocketAddrs;
-    let first = ("localhost", 0u16)
+    let candidates: Vec<_> = ("localhost", 0u16)
         .to_socket_addrs()
         .expect("resolve localhost")
-        .next()
-        .expect("localhost resolved to no addresses");
-    let recv = UdpSocket::bind(first).expect("bind recv");
+        .collect();
+    let addr = candidates
+        .iter()
+        .find(|a| a.is_ipv4())
+        .copied()
+        .unwrap_or(candidates[0]);
+    let recv = UdpSocket::bind(addr).expect("bind recv");
     recv.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
     let port = recv.local_addr().unwrap().port();
 
@@ -74,6 +80,27 @@ fn unicast_loopback_sends_via_hostname_url() {
     assert_eq!(got.as_slice(), &payload[..]);
 }
 
+/// P1 regression (integrator field report): a transient ICMP
+/// port-unreachable must never kill the sender. With the old connected
+/// socket, Linux surfaced it as a fatal ECONNREFUSED on the next send.
+#[test]
+fn send_to_absent_peer_never_errors() {
+    // Bind + drop to obtain a loopback port with nothing behind it.
+    let probe = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    let port = probe.local_addr().unwrap().port();
+    drop(probe);
+    let mut t = UdpTransport::connect(&format!("udp://127.0.0.1:{port}")).unwrap();
+    let payload = vec![0x47u8; 188];
+    for i in 0..8 {
+        // The sleep gives the kernel's ICMP reply time to arrive between
+        // sends — with a connected socket that made send #2+ fail.
+        t.send_bytes(&payload)
+            .unwrap_or_else(|e| panic!("send {i} failed: {e}"));
+        std::thread::sleep(std::time::Duration::from_millis(20));
+    }
+    assert_eq!(t.stats().datagrams_sent, 8);
+}
+
 #[test]
 fn unicast_loopback_recvs_payload_from_std_send() {
     let mut recv = UdpRecvTransport::listen("udp://@127.0.0.1:0").expect("build recv");

crates/tst-udp/tests/pipeline_round_trip.rs

@@ -113,15 +113,16 @@ fn mux_via_udp_demux_round_trip_recovers_program_map() {
     for i in 0i64..5 {
         let pts = Pts90khz::new(i * 9001);
         // The receiver thread returns the instant it sees a ProgramMap, which
-        // drops its socket. On fast runners that can happen before this loop
-        // finishes; Linux then surfaces the ICMP port-unreachable as
-        // ECONNREFUSED on this connected UDP socket's next send. That's benign
-        // here — the receiver already has what it needs. Stop sending on the
-        // first broken-transport error; the real assertion is the channel
-        // result below, which reports `false` (or times out) if the receiver
-        // never recovered a ProgramMap. Non-transport error kinds still panic
-        // so a real mux/config regression fails loudly. Mirrors the tst-tcp
-        // sibling test.
+        // drops its socket. The sender's UDP socket is unconnected
+        // (fire-and-forget `send_to`), so a receiver dropping mid-stream no
+        // longer surfaces as a fatal ECONNREFUSED here — sends to an absent
+        // peer simply succeed at the OS level. The broken-transport arm
+        // stays defensive for a genuine transport failure during the loop;
+        // the real assertion is the channel result below, which reports
+        // `false` (or times out) if the receiver never recovered a
+        // ProgramMap. Non-transport error kinds still panic so a real
+        // mux/config regression fails loudly. Mirrors the tst-tcp sibling
+        // test.
         match sender.send_video(&au, pts, true) {
             Ok(()) => {}
             Err(e)

tests/coverage/surface-manifest.toml

@@ -5839,6 +5839,15 @@ reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asse
 item = "tst_core::transport::TransportCancel::cancel"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
 
+[[surface]]
+kind = "rust"
+crate = "tst-core"
+item = "tst_core::transport::TransportError::is_connection_refused"
+owning_tests = ["crates/tst-core/src/transport.rs"]
+bindings = []
+scenario_ids = []
+tier = "A"
+
 [[exempt]]
 item = "tst_core::transport::TransportError::clone"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"