feat(transport): Windows DSCP via qWAVE flows — PUNKTFUNK_DSCP now real on the wire there

Networking-audit deferred plan §4 (the qos.rs follow-up). On Windows
set_tos_v4 succeeds but the stack strips the mark without a qWAVE flow, so
PUNKTFUNK_DSCP=1 was a silent wire no-op there. Now (Apollo/Sunshine's
approach): QOSCreateHandle once per process; QOSAddSocketToFlow per
connected media socket — video → QOSTrafficTypeAudioVideo, audio →
QOSTrafficTypeVoice (QOS_NON_ADAPTIVE_FLOW) — then best-effort
QOSSetFlow(QOSSetOutgoingDSCPValue, 40/48) to pin the exact CS5/CS6 the
other platforms mark. The pin lands for elevated processes (the host runs
as the SYSTEM service — exactly where the video egress is) or under the
"allow non-admin DSCP" policy; otherwise the traffic-type default marking
stands (still WMM-useful). Gating + contract unchanged: opt-in via
dscp_enabled(), every step debug-logs and continues.

set_media_qos now returns an RAII QosFlow guard (QOSRemoveSocketFromFlow on
drop) that must outlive the socket's traffic: stored in UdpTransport
(declared before the socket, so drop order removes the flow first) and held
for the stream's scope by the GameStream video/audio senders — whose
tagging moved after connect(), since qWAVE derives the flow's 5-tuple from
the connected socket (behavior-neutral on Linux). Off-Windows the guard is
inert and never constructed.

Validated: cargo check -p punktfunk-core --target x86_64-pc-windows-msvc
green (the full host can't cross-check from Linux — aws-lc-sys needs MSVC
tooling; it builds on-box via deploy-host.ps1). Remaining on the next
Windows pass per plan: deploy to the RTX box and pktmon/Wireshark the
client side — DSCP ≠ 0 on video egress with PUNKTFUNK_DSCP=1, 0 without.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/punktfunk-core/src/transport/mod.rs

@@ -3,10 +3,12 @@
 
 mod loopback;
 mod qos;
+#[cfg(windows)]
+mod qos_windows;
 mod udp;
 
 pub use loopback::{loopback_pair, LoopbackTransport};
-pub use qos::{grow_socket_buffers, set_dscp_default, set_media_qos, MediaClass};
+pub use qos::{grow_socket_buffers, set_dscp_default, set_media_qos, MediaClass, QosFlow};
 /// Windows-only: reusable USO (UDP Send Offload) batch send for callers that own their own connected
 /// socket (the GameStream video sender) rather than going through [`UdpTransport`].
 #[cfg(target_os = "windows")]

crates/punktfunk-core/src/transport/qos.rs

@@ -8,9 +8,10 @@
 //! QoS-aware path (Wi-Fi WMM access categories, a managed switch, a shaped uplink) can prioritize it
 //! over bulk flows. Mirrors what Apollo/Sunshine tag — DSCP **CS5** for video, **CS6** for audio. It
 //! is **opt-in** (`PUNKTFUNK_DSCP=1`, or [`set_dscp_default`] from an embedder — the Android client
-//! ties it to its experimental low-latency mode): DSCP can interact badly with some consumer ISPs/routers, and on
-//! Windows a plain `IP_TOS` is silently stripped unless a qWAVE policy is active (Apollo uses the
-//! qWAVE API there — that port is a follow-up; today this is a no-op on the wire on Windows).
+//! ties it to its experimental low-latency mode): DSCP can interact badly with some consumer
+//! ISPs/routers. On Windows a plain `IP_TOS` is silently stripped from the wire, so the marking
+//! goes through qWAVE flows instead (see [`super::qos_windows`]) — the caller holds the returned
+//! [`QosFlow`] guard for as long as the socket sends media.
 
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -60,7 +61,7 @@ pub enum MediaClass {
 
 impl MediaClass {
     /// DSCP code point (the high 6 bits of the IPv4 TOS / IPv6 traffic-class byte).
-    const fn dscp(self) -> u32 {
+    pub(super) const fn dscp(self) -> u32 {
         match self {
             MediaClass::Video => 40, // CS5
             MediaClass::Audio => 48, // CS6
@@ -92,20 +93,43 @@ pub(crate) fn dscp_enabled() -> bool {
     }
 }
 
+/// RAII token for a socket's QoS marking. On Windows it is the qWAVE flow membership
+/// ([`super::qos_windows::QosFlow`]) — dropping it removes the marking, so hold it for as long
+/// as the socket sends media. Elsewhere DSCP rides the socket option itself and the token is
+/// inert (and never constructed — [`set_media_qos`] returns `None`).
+#[cfg(windows)]
+pub use super::qos_windows::QosFlow;
+#[cfg(not(windows))]
+pub struct QosFlow {
+    _never: std::convert::Infallible,
+}
+
 /// Best-effort: tag `socket`'s outgoing packets for prioritized delivery of its media class. A no-op
 /// unless `PUNKTFUNK_DSCP=1`. Every step is best-effort (failures logged at debug, never fatal) — QoS
 /// is a nicety, not required for correctness.
 ///
-/// IPv4 only (all current media sockets bind `0.0.0.0`); a v6 socket simply isn't tagged. On Windows
-/// the `IP_TOS` set succeeds but the OS doesn't tag the wire without a qWAVE policy (follow-up).
-pub fn set_media_qos(socket: &UdpSocket, class: MediaClass) {
-    if dscp_enabled() {
+/// The socket must already be `connect`ed (Windows derives the qWAVE flow from the connected
+/// 5-tuple). IPv4 only (all current media sockets bind `0.0.0.0`); a v6 socket simply isn't
+/// tagged. Returns the [`QosFlow`] guard on Windows — keep it alive with the socket; `None`
+/// elsewhere (the marking is a plain socket option) and whenever a step refused.
+pub fn set_media_qos(socket: &UdpSocket, class: MediaClass) -> Option<QosFlow> {
+    if !dscp_enabled() {
+        return None;
+    }
+    #[cfg(windows)]
+    {
+        super::qos_windows::add_media_flow(socket, class)
+    }
+    #[cfg(not(windows))]
+    {
         apply_media_qos(socket, class);
+        None
     }
 }
 
 /// The unconditional QoS application, factored out of [`set_media_qos`] so it is directly testable
 /// without touching the process-global `PUNKTFUNK_DSCP` env. Best-effort (every step logs-and-continues).
+#[cfg_attr(windows, allow(dead_code))]
 fn apply_media_qos(socket: &UdpSocket, class: MediaClass) {
     let sock = socket2::SockRef::from(socket);
     // DSCP occupies the high 6 bits of the TOS byte → shift left 2.
@@ -143,8 +167,8 @@ mod tests {
     fn qos_and_buffer_growth_are_best_effort_and_never_panic() {
         let sock = UdpSocket::bind("127.0.0.1:0").unwrap();
         // No PUNKTFUNK_DSCP in the test env → early return; must not panic regardless.
-        set_media_qos(&sock, MediaClass::Video);
-        set_media_qos(&sock, MediaClass::Audio);
+        assert!(set_media_qos(&sock, MediaClass::Video).is_none());
+        assert!(set_media_qos(&sock, MediaClass::Audio).is_none());
         grow_socket_buffers(&sock);
     }
 

crates/punktfunk-core/src/transport/qos_windows.rs

@@ -0,0 +1,134 @@
+//! qWAVE (qos2.h) DSCP marking — the Windows path of [`super::qos::set_media_qos`].
+//!
+//! On Windows a plain `IP_TOS` setsockopt succeeds but the stack strips the mark from the wire:
+//! marking requires membership in a qWAVE flow, which is how Apollo/Sunshine tag (qwave.dll).
+//! [`QOSAddSocketToFlow`] with a traffic type yields the OS default marking (AudioVideo → DSCP
+//! 40, Voice → 56, both WMM-mapped); the follow-up `QOSSetFlow(QOSSetOutgoingDSCPValue)` pins
+//! the exact CS5/CS6 code points the other platforms mark. The pin needs an elevated process or
+//! the "allow non-admin DSCP" group policy and silently keeps the traffic-type default
+//! otherwise — the host runs as the SYSTEM service (`PunktfunkHost`), so the exact-DSCP path
+//! applies exactly where it matters (the video egress); user-mode clients keep traffic-type
+//! defaults (still WMM-useful).
+//!
+//! Same contract as the rest of [`super::qos`]: opt-in (`dscp_enabled`), and every step
+//! debug-logs and continues — QoS is a nicety, never required for correctness.
+
+use super::qos::MediaClass;
+use std::net::UdpSocket;
+use std::os::windows::io::AsRawSocket;
+use std::sync::OnceLock;
+use windows_sys::Win32::Foundation::{GetLastError, HANDLE};
+use windows_sys::Win32::NetworkManagement::QoS::{
+    QOSAddSocketToFlow, QOSCreateHandle, QOSRemoveSocketFromFlow, QOSSetFlow,
+    QOSSetOutgoingDSCPValue, QOSTrafficTypeAudioVideo, QOSTrafficTypeVoice, QOS_NON_ADAPTIVE_FLOW,
+    QOS_VERSION,
+};
+
+/// The process-wide qWAVE handle (`QOSCreateHandle` once, cached). `None` = qWAVE unavailable
+/// (e.g. the QWAVE service is disabled) — every flow request then no-ops. Deliberately never
+/// closed: it lives as long as the process, like the media sockets whose flows it carries.
+fn qos_handle() -> Option<HANDLE> {
+    static SLOT: OnceLock<Option<usize>> = OnceLock::new();
+    SLOT.get_or_init(|| {
+        let version = QOS_VERSION {
+            MajorVersion: 1,
+            MinorVersion: 0,
+        };
+        let mut handle: HANDLE = std::ptr::null_mut();
+        // SAFETY: both pointers are valid for the duration of the synchronous call.
+        if unsafe { QOSCreateHandle(&version, &mut handle) } == 0 {
+            tracing::debug!(
+                err = unsafe { GetLastError() },
+                "QOSCreateHandle failed — qWAVE DSCP marking unavailable"
+            );
+            None
+        } else {
+            Some(handle as usize)
+        }
+    })
+    .map(|h| h as HANDLE)
+}
+
+/// RAII qWAVE flow membership: while held, the socket's egress carries the flow's marking;
+/// dropping removes the socket from the flow. (Closing the socket also removes it implicitly —
+/// the guard makes teardown explicit and ordered, and must outlive the socket's traffic.)
+pub struct QosFlow {
+    /// Raw `SOCKET` value only — never dereferenced; qWAVE tolerates an already-closed socket
+    /// on remove (the error is deliberately ignored).
+    socket: u64,
+    flow_id: u32,
+}
+
+impl Drop for QosFlow {
+    fn drop(&mut self) {
+        if let Some(handle) = qos_handle() {
+            // SAFETY: handle/flow_id came from the successful add; a stale socket just errors.
+            unsafe { QOSRemoveSocketFromFlow(handle, self.socket as _, self.flow_id, 0) };
+        }
+    }
+}
+
+/// Put a **connected** media socket on a qWAVE flow of its class (video →
+/// `QOSTrafficTypeAudioVideo`, audio → `QOSTrafficTypeVoice`), then best-effort pin the exact
+/// DSCP the other platforms mark (CS5 = 40 / CS6 = 48). Returns the flow guard, or `None` when
+/// a required step refused (logged at debug).
+pub(super) fn add_media_flow(socket: &UdpSocket, class: MediaClass) -> Option<QosFlow> {
+    let handle = qos_handle()?;
+    let traffic_type = match class {
+        MediaClass::Video => QOSTrafficTypeAudioVideo,
+        MediaClass::Audio => QOSTrafficTypeVoice,
+    };
+    let raw = socket.as_raw_socket();
+    let mut flow_id = 0u32;
+    // NULL destination = derive the flow's 5-tuple from the connected socket (every media
+    // socket is `connect`ed before it is tagged).
+    // SAFETY: the socket is live for the call; `flow_id` is a valid out-pointer.
+    let ok = unsafe {
+        QOSAddSocketToFlow(
+            handle,
+            raw as _,
+            std::ptr::null(),
+            traffic_type,
+            QOS_NON_ADAPTIVE_FLOW,
+            &mut flow_id,
+        )
+    };
+    if ok == 0 {
+        tracing::debug!(
+            err = unsafe { GetLastError() },
+            ?class,
+            "QOSAddSocketToFlow failed — DSCP marking skipped"
+        );
+        return None;
+    }
+    // Construct the guard FIRST so an early return below still removes the flow membership.
+    let flow = QosFlow {
+        socket: raw as u64,
+        flow_id,
+    };
+    // Pin the exact code point. Succeeds for elevated processes or under the "allow non-admin
+    // DSCP" policy; otherwise the traffic-type default marking stands (40 / 56 — WMM-useful).
+    let dscp: u32 = class.dscp();
+    // SAFETY: `buffer` points at 4 valid bytes for the synchronous (no OVERLAPPED) call.
+    let ok = unsafe {
+        QOSSetFlow(
+            handle,
+            flow_id,
+            QOSSetOutgoingDSCPValue,
+            4,
+            &dscp as *const u32 as *const _,
+            0,
+            std::ptr::null_mut(),
+        )
+    };
+    if ok == 0 {
+        tracing::debug!(
+            err = unsafe { GetLastError() },
+            ?class,
+            "QOSSetFlow(OutgoingDSCPValue) refused — traffic-type default marking stands"
+        );
+    } else {
+        tracing::debug!(?class, dscp, flow_id, "qWAVE flow pinned to exact DSCP");
+    }
+    Some(flow)
+}

crates/punktfunk-core/src/transport/udp.rs

@@ -446,6 +446,9 @@ pub fn spawn_data_punch(sock: UdpSocket, stop: std::sync::Arc<std::sync::atomic:
 }
 
 pub struct UdpTransport {
+    /// qWAVE flow guard (Windows, opt-in DSCP): declared before `socket` so drop order removes
+    /// the flow membership before the socket closes. Always `None` off-Windows.
+    _qos_flow: Option<super::qos::QosFlow>,
     socket: UdpSocket,
 }
 
@@ -464,10 +467,14 @@ impl UdpTransport {
         socket.connect(peer)?;
         super::qos::grow_socket_buffers(&socket);
         // The native data plane is video-dominant — tag it as the video class (opt-in via
-        // PUNKTFUNK_DSCP). Each end marks its own egress.
-        super::qos::set_media_qos(&socket, super::qos::MediaClass::Video);
+        // PUNKTFUNK_DSCP). Each end marks its own egress; the socket is connected by now, as
+        // the Windows qWAVE flow requires.
+        let qos_flow = super::qos::set_media_qos(&socket, super::qos::MediaClass::Video);
         socket.set_nonblocking(true)?;
-        Ok(UdpTransport { socket })
+        Ok(UdpTransport {
+            _qos_flow: qos_flow,
+            socket,
+        })
     }
 
     /// Host side of the data plane for clients that may sit behind NAT / a stateful inter-VLAN
@@ -524,9 +531,15 @@ impl UdpTransport {
         socket.connect(target.as_deref().unwrap_or(fallback_peer))?;
         socket.set_read_timeout(None)?;
         super::qos::grow_socket_buffers(&socket);
-        super::qos::set_media_qos(&socket, super::qos::MediaClass::Video);
+        let qos_flow = super::qos::set_media_qos(&socket, super::qos::MediaClass::Video);
         socket.set_nonblocking(true)?;
-        Ok((UdpTransport { socket }, punched))
+        Ok((
+            UdpTransport {
+                _qos_flow: qos_flow,
+                socket,
+            },
+            punched,
+        ))
     }
 
     /// A second handle to the data socket, for sending hole-punch keepalives ([`PUNCH_MAGIC`])