SocketPool: throw CancellationError when cancelled while waiting

FlyingFox/Sources/HTTPServer.swift

@@ -101,7 +101,9 @@ public final actor HTTPServer {
             state = (socket: socket, task: task)
             try await task.getValue(cancelling: .whenParentIsCancelled)
         } catch {
-            logger.logCritical("server error: \(error.localizedDescription)")
+            if !(error is CancellationError) {
+                logger.logCritical("server error: \(error.localizedDescription)")
+            }
             if let state = self.state {
                 try? state.socket.close()
             }

FlyingFox/Tests/HTTPServerTests.swift

@@ -121,11 +121,28 @@ actor HTTPServerTests {
 
         task.cancel()
 
-        await #expect(throws: (any Error).self) {
+        await #expect(throws: CancellationError.self) {
             try await task.value
         }
     }
 
+    @Test
+    func taskCanBeCancelled_WhileWaitingForConnections() async throws {
+        // Once the socket pool waits on its event queue, cancelling it stops the queue under
+        // that wait. The server should still end with CancellationError rather than the
+        // wait's failure (EBADF from kevent). The two race, so try several times.
+        for _ in 0..<5 {
+            let server = HTTPServer.make(address: .loopback(port: 0))
+            let task = try await startServer(server)
+            try await Task.sleep(nanoseconds: 200_000_000)
+            task.cancel()
+
+            await #expect(throws: CancellationError.self) {
+                try await task.value
+            }
+        }
+    }
+
     @Test
     func taskCanBeCancelled_AfterServerIsStopped() async throws {
         let server = HTTPServer.make()

FlyingSocks/Sources/Socket.swift

@@ -765,16 +765,20 @@ extension Socket {
                     // match getPacketInfoControl above.
                     header.pointee.cmsg_level = Socket.ipproto_ip
                     header.pointee.cmsg_type = Socket.ip_pktinfo
+                    // ManagedBuffer leaves the element uninitialized: zero it, so that no
+                    // field is sent holding whatever was in memory.
+                    element.initialize(to: in_pktinfo())
                     element.pointee.ipi_ifindex = IPv4InterfaceIndexType(interfaceIndex ?? 0)
+                    // ip(7): "If IP_PKTINFO is passed to sendmsg(2) and ipi_spec_dst is not
+                    // zero, then it is used as the local source address for the routing
+                    // table lookup". ipi_addr is the destination of a received packet.
                     if let address {
-                        var address = address
-                        withUnsafePointer(to: &address) {
-                            $0.withMemoryRebound(to: sockaddr_in.self, capacity: 1) {
-                                element.pointee.ipi_addr = $0.pointee.sin_addr
+                        let storage = address.makeStorage()
+                        if storage.ss_family == sa_family_t(AF_INET) {
+                            element.pointee.ipi_spec_dst = withUnsafeBytes(of: storage) {
+                                $0.load(as: sockaddr_in.self).sin_addr
                             }
                         }
-                    } else {
-                        element.pointee.ipi_addr.s_addr = 0
                     }
 
                     return header.pointee
@@ -791,16 +795,20 @@ extension Socket {
                     // match getPacketInfoControl above.
                     header.pointee.cmsg_level = Socket.ipproto_ipv6
                     header.pointee.cmsg_type = Socket.ipv6_pktinfo
+                    // memberwise: Socket+Glibc declares its own in6_pktinfo, which has no
+                    // zero initializer
+                    element.initialize(to: in6_pktinfo(ipi6_addr: in6_addr(), ipi6_ifindex: 0))
                     element.pointee.ipi6_ifindex = IPv6InterfaceIndexType(interfaceIndex ?? 0)
+                    // RFC 3542 §6.1: on output, ipi6_addr is the source address. Read it
+                    // from the address's storage: a sockaddr_in6 is too large to be held
+                    // inline in an existential, so the existential's own bytes are not it.
                     if let address {
-                        var address = address
-                        withUnsafePointer(to: &address) {
-                            $0.withMemoryRebound(to: sockaddr_in6.self, capacity: 1) {
-                                element.pointee.ipi6_addr = $0.pointee.sin6_addr
+                        let storage = address.makeStorage()
+                        if storage.ss_family == sa_family_t(AF_INET6) {
+                            element.pointee.ipi6_addr = withUnsafeBytes(of: storage) {
+                                $0.load(as: sockaddr_in6.self).sin6_addr
                             }
                         }
-                    } else {
-                        element.pointee.ipi6_addr = in6_addr()
                     }
 
                     return header.pointee

FlyingSocks/Sources/SocketPool.swift

@@ -139,15 +139,21 @@ public final actor SocketPool<Queue: EventQueue>: AsyncSocketPool {
     private func getNotifications() async throws -> [EventNotification] {
         try Task.checkCancellation()
         nonisolated(unsafe) let queue = queue
-        return try await withIdentifiableThrowingContinuation { continuation in
-            dispatchQueue.async {
-                let result = Result {
-                    try queue.getNotifications()
+        do {
+            return try await withIdentifiableThrowingContinuation { continuation in
+                dispatchQueue.async {
+                    let result = Result {
+                        try queue.getNotifications()
+                    }
+                    continuation.resume(with: result)
                 }
-                continuation.resume(with: result)
+            } onCancel: { _ in
+                Task { await self.stopQueue() }
             }
-        } onCancel: { _ in
-            Task { await self.stopQueue() }
+        } catch where Task.isCancelled {
+            // cancelling stops the queue under the wait in progress, which then fails
+            // (with EBADF from kevent) rather than as a cancellation
+            throw CancellationError()
         }
     }
 

FlyingSocks/Tests/SocketTests.swift

@@ -419,6 +419,52 @@ struct SocketTests {
         #expect(type == Int32(IPV6_PKTINFO))
         #endif
     }
+
+    @Test
+    func withPacketInfoControl_IP4_SetsSourceAddressInSpecDst() throws {
+        // ip(7): sendmsg(2) takes the source address from ipi_spec_dst; ipi_addr is the
+        // destination of a received packet. The address is also passed as a
+        // sockaddr_storage, which an existential holds boxed rather than inline.
+        let loopback = try sockaddr_in.inet(ip4: "127.0.0.1", port: 0)
+        for address in [loopback as any SocketAddress, loopback.makeStorage()] {
+            let info = Socket.withPacketInfoControl(
+                family: sa_family_t(AF_INET),
+                interfaceIndex: 7,
+                address: address
+            ) { header, _ in
+                UnsafeRawPointer(header!)
+                    .advanced(by: MemoryLayout<cmsghdr>.size)
+                    .loadUnaligned(as: in_pktinfo.self)
+            }
+            #expect(info.ipi_spec_dst.s_addr == loopback.sin_addr.s_addr)
+            #expect(info.ipi_addr.s_addr == 0)
+            #expect(info.ipi_ifindex == 7)
+        }
+    }
+
+    @Test
+    func withPacketInfoControl_IP6_SetsSourceAddress() {
+        // RFC 3542 §6.1: on output, ipi6_addr is the source address. A sockaddr_in6 is too
+        // large to be held inline in an existential, so this also checks the address is
+        // read from the value, not from the existential's own storage.
+        let loopback = sockaddr_in6.loopback(port: 0)
+        for address in [loopback as any SocketAddress, loopback.makeStorage()] {
+            let info = Socket.withPacketInfoControl(
+                family: sa_family_t(AF_INET6),
+                interfaceIndex: 7,
+                address: address
+            ) { header, _ in
+                UnsafeRawPointer(header!)
+                    .advanced(by: MemoryLayout<cmsghdr>.size)
+                    .loadUnaligned(as: in6_pktinfo.self)
+            }
+            #expect(
+                withUnsafeBytes(of: info.ipi6_addr) { Array($0) } ==
+                    withUnsafeBytes(of: loopback.sin6_addr) { Array($0) }
+            )
+            #expect(info.ipi6_ifindex == 7)
+        }
+    }
     #endif
 }