Neutralize disconnected UDP controllers and bound subscriber processing

Sources/FinallyTheControllerWorks/Output/UDPHub.swift

@@ -8,7 +8,7 @@
 //          | f32 lx,ly,rx,ry | u8 lt,rt | u16 battery_mv
 //          | i16 gyro[3] | i16 accel[3]
 //   Rumble in  (6 bytes):  "S2R1" | u8 strong | u8 weak
-//   Any inbound datagram registers its sender as a subscriber (30 s TTL).
+//   An empty hello or valid rumble registers its sender (30 s TTL).
 
 import Foundation
 import Darwin
@@ -17,6 +17,7 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable {
 
     private static let basePort: UInt16 = 24800
     private static let peerTTL: TimeInterval = 30
+    private static let maxPeers = 64
 
     var onRumble: ((Int, Double, Double) -> Void)?
 
@@ -93,7 +94,8 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable {
     private func drainSocket(slot: Int) {
         guard let s = slots[slot] else { return }
         var buf = [UInt8](repeating: 0, count: 64)
-        while true {
+        // Bound each read dispatch so a noisy local peer cannot starve output.
+        for _ in 0..<256 {
             var from = sockaddr_in()
             var fromLen = socklen_t(MemoryLayout<sockaddr_in>.size)
             let n = withUnsafeMutablePointer(to: &from) {
@@ -102,14 +104,19 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable {
                 }
             }
             if n < 0 { break }  // EWOULDBLOCK: drained
+            let isRumble = n == 6 && buf[0...3].elementsEqual([0x53, 0x32, 0x52, 0x31])
+            guard n == 0 || isRumble else { continue }
+            let now = ProcessInfo.processInfo.systemUptime
+            s.peers = s.peers.filter { now - $0.value <= Self.peerTTL }
             let peer = SockAddr(addr: from.sin_addr.s_addr, port: from.sin_port)
             let isNewPeer = s.peers[peer] == nil
-            s.peers[peer] = CFAbsoluteTimeGetCurrent()
+            guard !isNewPeer || s.peers.count < Self.maxPeers else { continue }
+            s.peers[peer] = now
             if isNewPeer, !s.name.isEmpty {
                 // Late joiners get the name before their first state packet.
                 send(Self.namePacket(s.name), to: peer, via: s.fd)
             }
-            if n >= 6, buf[0] == 0x53, buf[1] == 0x32, buf[2] == 0x52, buf[3] == 0x31 {  // "S2R1"
+            if isRumble {
                 onRumble?(slot, Double(buf[4]) / 255.0, Double(buf[5]) / 255.0)
             }
         }
@@ -154,7 +161,13 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable {
 
     func controllerDisconnected(slot: Int) {
         queue.async { [weak self] in
-            self?.slots[slot]?.seq = 0
+            guard let self, let s = self.slots[slot] else { return }
+            // Release held input immediately on orderly disconnect. The SDL
+            // presence watchdog remains necessary for crashes or packet loss.
+            s.seq &+= 1
+            let neutral = Self.statePacket(seq: s.seq, state: ControllerState())
+            for peer in s.peers.keys { self.send(neutral, to: peer, via: s.fd) }
+            s.name = ""
         }
     }
 
@@ -163,7 +176,7 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable {
             guard let self, let s = self.slots[slot], !s.peers.isEmpty else { return }
             s.seq &+= 1
             let packet = Self.statePacket(seq: s.seq, state: state)
-            let now = CFAbsoluteTimeGetCurrent()
+            let now = ProcessInfo.processInfo.systemUptime
             for (peer, seen) in s.peers {
                 if now - seen > Self.peerTTL {
                     s.peers.removeValue(forKey: peer)

tests/udp/UDPTests.swift

@@ -0,0 +1,112 @@
+import Foundation
+#if os(Linux)
+import Glibc
+let datagram = Int32(SOCK_DGRAM.rawValue)
+#else
+import Darwin
+let datagram = SOCK_DGRAM
+#endif
+
+enum BridgeEngine { static let maxPlayers = 4 }
+enum LogLevel { case info, warning, error, debug }
+func bridgeLog(_ level: LogLevel, _ category: String, _ message: String) {}
+protocol ControllerOutputSink: AnyObject {
+    var onRumble: ((Int, Double, Double) -> Void)? { get set }
+    func controllerConnected(slot: Int, model: Switch2.Model)
+    func controllerDisconnected(slot: Int)
+    func controllerName(slot: Int, name: String)
+    func controllerState(slot: Int, state: ControllerState)
+}
+
+@main
+enum UDPTests {
+    static func client(_ port: UInt16) -> Int32 {
+        let fd = socket(AF_INET, datagram, 0)
+        precondition(fd >= 0)
+        var address = sockaddr_in()
+        address.sin_family = sa_family_t(AF_INET)
+        address.sin_port = port.bigEndian
+        address.sin_addr.s_addr = UInt32(0x7f000001).bigEndian
+        let result = withUnsafePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
+                connect(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
+            }
+        }
+        precondition(result == 0)
+        var timeout = timeval(tv_sec: 1, tv_usec: 0)
+        precondition(setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, socklen_t(MemoryLayout<timeval>.size)) == 0)
+        return fd
+    }
+    static func sendBytes(_ fd: Int32, _ data: [UInt8]) {
+        let n = data.withUnsafeBytes { send(fd, $0.baseAddress, $0.count, 0) }
+        precondition(n == data.count)
+    }
+    static func receive(_ fd: Int32) -> Data {
+        var bytes = [UInt8](repeating: 0, count: 128)
+        let n = recv(fd, &bytes, bytes.count, 0)
+        precondition(n >= 0, "Expected UDP packet was not delivered")
+        return Data(bytes.prefix(n))
+    }
+    static func sendAndDrain(_ hub: UDPHub, slot: Int, fd: Int32, bytes: [UInt8]) {
+        // Hold the hub's serial queue so its dispatch source cannot consume
+        // the packet between poll() and the explicit production drain call.
+        hub.queue.sync {
+            sendBytes(fd, bytes)
+            var ready = pollfd(fd: hub.slots[slot]!.fd, events: Int16(POLLIN), revents: 0)
+            precondition(poll(&ready, 1, 1000) == 1, "Loopback packet did not arrive")
+            hub.drainSocket(slot: slot)
+        }
+    }
+    static func main() {
+        let selected = CommandLine.arguments.last!
+        let hub = UDPHub()
+        hub.queue.sync { precondition(hub.slots.count == 4) }
+        let a = client(24800), b = client(24801)
+        defer { close(a); close(b) }
+        sendBytes(a, []); sendBytes(b, [])
+        for _ in 0..<100 {
+            if hub.queue.sync(execute: { !hub.slots[0]!.peers.isEmpty && !hub.slots[1]!.peers.isEmpty }) { break }
+            Thread.sleep(forTimeInterval: 0.005)
+        }
+        if selected == "all" || selected == "neutral" {
+            var held = ControllerState(); held.buttons = [.a]; held.leftTrigger = 255
+            hub.controllerState(slot: 0, state: held)
+            precondition(Switch2.u32(receive(a), 8) == Switch2.Buttons.a.rawValue)
+            hub.controllerDisconnected(slot: 0)
+            let packet = receive(a)
+            precondition(packet.count == 44 && packet.prefix(4) == Data("S2B1".utf8))
+            precondition(packet.dropFirst(8).allSatisfy { $0 == 0 }, "Disconnect must emit neutral state")
+            hub.controllerState(slot: 1, state: held)
+            precondition(Switch2.u32(receive(b), 8) == Switch2.Buttons.a.rawValue)
+            print("PASS orderly neutralization and unrelated slot")
+        }
+        if selected == "all" || selected == "malformed" {
+            let c = client(24802); defer { close(c) }
+            sendAndDrain(hub, slot: 2, fd: c, bytes: [1, 2, 3])
+            hub.queue.sync {
+                precondition(hub.slots[2]!.peers.isEmpty, "Malformed packets must not allocate subscribers")
+            }
+            print("PASS malformed subscription rejection")
+        }
+        if selected == "all" || selected == "capacity" {
+            hub.queue.sync {
+                let s = hub.slots[3]!
+                for port in 1...64 {
+                    s.peers[UDPHub.SockAddr(addr: 0x0100007f, port: UInt16(port))] = ProcessInfo.processInfo.systemUptime
+                }
+            }
+            let c = client(24803); defer { close(c) }
+            sendAndDrain(hub, slot: 3, fd: c, bytes: [])
+            hub.queue.sync {
+                precondition(hub.slots[3]!.peers.count == 64, "Peer table must stay bounded")
+            }
+            // Expired peers must not block a new subscriber even without state traffic.
+            hub.queue.sync { hub.slots[3]!.peers = hub.slots[3]!.peers.mapValues { _ in -100 } }
+            sendAndDrain(hub, slot: 3, fd: c, bytes: [])
+            hub.queue.sync {
+                precondition(hub.slots[3]!.peers.count == 1)
+            }
+            print("PASS peer cap and expiry without input")
+        }
+    }
+}

tests/udp/run.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+set -euo pipefail
+cd "$(dirname "$0")/../.."
+work=$(mktemp -d)
+trap 'rm -rf "$work"' EXIT
+python3 - "$work" <<'PY'
+from pathlib import Path
+import os,re,sys
+out=Path(sys.argv[1])
+s=Path(os.environ.get('UDP_SOURCE','Sources/FinallyTheControllerWorks/Output/UDPHub.swift')).read_text()
+s=re.sub(r'\bprivate\s+', '', s)
+if sys.platform != 'darwin':
+    s=s.replace('import Darwin','import Glibc\nimport CoreFoundation').replace('SOCK_DGRAM,','Int32(SOCK_DGRAM.rawValue),')
+out.joinpath('UDPHub.swift').write_text(s)
+s=Path('Sources/FinallyTheControllerWorks/Bluetooth/ControllerSession.swift').read_text()
+a=s.index('struct ControllerState:'); b=s.index('/// Called on the Bluetooth queue.',a)
+out.joinpath('State.swift').write_text('import Foundation\n'+s[a:b])
+PY
+swiftc -swift-version 5 Sources/FinallyTheControllerWorks/Protocol/Switch2Protocol.swift \
+ "$work/State.swift" "$work/UDPHub.swift" tests/udp/UDPTests.swift -o "$work/test"
+"$work/test" "${UDP_CASE:-all}"