Harden peer connectivity framing

Author: 1amageek

Package.resolved

@@ -34,13 +34,7 @@ let package = Package(
         .library(name: "PeerConnectivityBonjour", targets: ["PeerConnectivityBonjour"]),
     ],
     dependencies: [
-        // `embedded`-branch only: resolve swift-libp2p against the local working
-        // tree (its `embedded` branch) so this package builds against the
-        // Embedded-clean libp2p cores under development. This matches the
-        // established cross-package wiring on the `embedded` branch.
-        // RESTORE the URL ref (`from: "0.2.0"`) before release — a local-path ref
-        // must never be tagged, or downstream dependency resolution fails.
-        .package(path: "../swift-libp2p"),
+        .package(url: "https://github.com/1amageek/swift-libp2p.git", from: "0.2.1"),
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.91.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.8.0"),
     ],
@@ -49,13 +43,10 @@ let package = Package(
         // No Foundation, no NIO, no `any`, no Mutex/ContinuousClock/key paths.
         // Owns the resource-transfer header framing (`"<name>\0<size>\0"`) over
         // `[UInt8]`; the async stream loop / file I/O / chunk transfer stay in
-        // the `PeerConnectivityLibP2P` adapter. Depends on the libp2p
-        // Embedded-clean core for `decodeUTF8Strict`.
+        // the `PeerConnectivityLibP2P` adapter.
         .target(
             name: "PeerConnectivityCore",
-            dependencies: [
-                .product(name: "LibP2PCore", package: "swift-libp2p"),
-            ],
+            dependencies: [],
             exclude: ["CONTEXT.md"],
             swiftSettings: coreSettings
         ),

Package.swift

@@ -2,6 +2,7 @@ import Synchronization
 
 public final class PeerConnectivityEventBroadcaster<T: Sendable>: Sendable {
     private let state: Mutex<BroadcastState>
+    private let bufferingPolicy: AsyncStream<T>.Continuation.BufferingPolicy
 
     private struct Entry: Sendable {
         let id: UInt64
@@ -11,16 +12,24 @@ public final class PeerConnectivityEventBroadcaster<T: Sendable>: Sendable {
     private struct BroadcastState: Sendable {
         var entries: [Entry] = []
         var nextID: UInt64 = 0
+        var isShutdown = false
     }
 
-    public init() {
+    private enum SubscribeAction: Sendable {
+        case registered(UInt64)
+        case finishImmediately
+    }
+
+    public init(bufferingPolicy: AsyncStream<T>.Continuation.BufferingPolicy = .bufferingNewest(1024)) {
+        self.bufferingPolicy = bufferingPolicy
         self.state = Mutex(BroadcastState())
     }
 
     deinit {
         let entries = state.withLock { state in
             let entries = state.entries
             state.entries.removeAll()
+            state.isShutdown = true
             return entries
         }
         for entry in entries {
@@ -29,31 +38,41 @@ public final class PeerConnectivityEventBroadcaster<T: Sendable>: Sendable {
     }
 
     public func subscribe() -> AsyncStream<T> {
-        let (stream, continuation) = AsyncStream<T>.makeStream()
-        let id = state.withLock { state -> UInt64 in
+        let (stream, continuation) = AsyncStream<T>.makeStream(bufferingPolicy: bufferingPolicy)
+        let action = state.withLock { state -> SubscribeAction in
+            guard !state.isShutdown else { return .finishImmediately }
             let id = state.nextID
             state.nextID += 1
             state.entries.append(Entry(id: id, continuation: continuation))
-            return id
+            return .registered(id)
         }
 
-        continuation.onTermination = { [weak self] _ in
-            self?.state.withLock { state in
-                state.entries.removeAll { $0.id == id }
+        switch action {
+        case .registered(let id):
+            continuation.onTermination = { [weak self] _ in
+                self?.state.withLock { state in
+                    state.entries.removeAll { $0.id == id }
+                }
             }
+        case .finishImmediately:
+            continuation.finish()
         }
         return stream
     }
 
     public func emit(_ event: T) {
-        let entries = state.withLock { $0.entries }
+        let entries = state.withLock { state in
+            state.isShutdown ? [] : state.entries
+        }
         for entry in entries {
             entry.continuation.yield(event)
         }
     }
 
     public func shutdown() {
         let entries = state.withLock { state -> [Entry] in
+            guard !state.isShutdown else { return [] }
+            state.isShutdown = true
             let entries = state.entries
             state.entries.removeAll()
             return entries

Sources/PeerConnectivity/EventBroadcaster.swift

@@ -23,18 +23,19 @@ can never be accepted as a complete header.
 - The trailing NUL after `<size>` is REQUIRED: a header missing it is rejected
   (`missingSizeSeparator`), so a truncated header is never accepted as complete.
 - The name is decoded lossy (U+FFFD substitution) — a malformed name never
-  fails the transfer; the size token is decoded strictly (`decodeUTF8Strict`)
-  and parsed as a non-negative base-10 integer, rejecting any non-digit,
-  leading sign, empty token, or overflow (`invalidSize` / `emptySize`).
+  fails the transfer; the size token is parsed directly as ASCII decimal,
+  rejecting any non-digit, leading sign, empty token, or overflow
+  (`invalidSize` / `emptySize`).
 - An empty name is rejected (`missingNameSeparator` requires `nameEnd > 0`).
 - `decodeMaterialized` additionally requires the remaining payload length to
   equal the declared size, failing closed with `payloadSizeMismatch` — sizes
   are never trusted to match silently.
 
 ## Embedded constraints (do not regress)
-- No Foundation, no NIO, no `any`, no `Mutex`, no `ContinuousClock`, no key
-  paths. UTF-8 decoding reuses `LibP2PCore.decodeUTF8Strict`. This module is
-  part of the dual-build (host + Embedded) Embedded-clean core.
+- No Foundation, no NIO, no libp2p dependency, no `any`, no `Mutex`, no
+  `ContinuousClock`, no key paths. Size parsing is ASCII-decimal directly over
+  bytes, and malformed/non-digit bytes fail closed as `invalidSize`. This module
+  is part of the dual-build (host + Embedded) Embedded-clean core.
 
 ## Wire protocol notes
 - Header body: `"<name>\0<size>\0"`. `<size>` is the exact payload byte count

Sources/PeerConnectivityCore/CONTEXT.md

@@ -1,10 +1,8 @@
-import LibP2PCore
-
 /// Resource-transfer header framing for PeerConnectivity (Embedded-clean).
 ///
 /// Embedded-clean: no Foundation, no NIO, no `any`, no Mutex, no
-/// ContinuousClock, no key paths. Operates over `[UInt8]` and uses
-/// `LibP2PCore.decodeUTF8Strict` for strict UTF-8 decoding of the size token.
+/// ContinuousClock, no key paths. Operates over `[UInt8]` without pulling in
+/// the host libp2p facade.
 ///
 /// A resource is sent over a libp2p stream as a sequence of length-prefixed
 /// frames (the length prefix is owned by the muxer's
@@ -68,7 +66,7 @@ public enum ResourceFrameCodec {
     /// - Throws: `ResourceFrameError` describing the specific framing failure.
     public static func decodeHeader(_ bytes: [UInt8]) throws(ResourceFrameError) -> ResourceHeader {
         let layout = try parseHeaderLayout(bytes)
-        let name = lossyUTF8(Array(bytes[0..<layout.nameEnd]))
+        let name = lossyUTF8(bytes[0..<layout.nameEnd])
         return ResourceHeader(name: name, size: layout.size)
     }
 
@@ -103,7 +101,7 @@ public enum ResourceFrameCodec {
         guard availablePayload == layout.size else {
             throw .payloadSizeMismatch(declared: layout.size, available: availablePayload)
         }
-        let name = lossyUTF8(Array(bytes[0..<layout.nameEnd]))
+        let name = lossyUTF8(bytes[0..<layout.nameEnd])
         return MaterializedHeader(
             header: ResourceHeader(name: name, size: layout.size),
             payloadStart: layout.payloadStart
@@ -127,12 +125,10 @@ public enum ResourceFrameCodec {
               let sizeEnd = firstIndex(of: separator, in: bytes, from: sizeStart) else {
             throw .missingSizeSeparator
         }
-        let sizeBytes = Array(bytes[sizeStart..<sizeEnd])
-        guard !sizeBytes.isEmpty else {
+        guard sizeStart < sizeEnd else {
             throw .emptySize
         }
-        guard let sizeString = decodeSizeToken(sizeBytes),
-              let size = parseNonNegativeInt(sizeString) else {
+        guard let size = parseNonNegativeInt(bytes, start: sizeStart, end: sizeEnd) else {
             throw .invalidSize
         }
         return HeaderLayout(nameEnd: nameEnd, payloadStart: sizeEnd + 1, size: size)
@@ -149,23 +145,21 @@ public enum ResourceFrameCodec {
         return nil
     }
 
-    private static func decodeSizeToken(_ bytes: [UInt8]) -> String? {
-        // Strict UTF-8: a non-numeric / malformed size token is a framing error,
-        // never substituted. Reuses the Embedded-clean helper from LibP2PCore.
-        decodeUTF8Strict(bytes)
-    }
-
-    /// Parses a non-negative base-10 integer from `string`, rejecting any
+    /// Parses a non-negative base-10 integer from ASCII bytes, rejecting any
     /// non-digit character, leading sign, or overflow. Returns `nil` on failure.
-    private static func parseNonNegativeInt(_ string: String) -> Int? {
+    private static func parseNonNegativeInt(
+        _ bytes: [UInt8],
+        start: Int,
+        end: Int
+    ) -> Int? {
         var value = 0
-        var sawDigit = false
-        for scalar in string.unicodeScalars {
-            guard scalar.value >= 48, scalar.value <= 57 else {
+        var index = start
+        while index < end {
+            let byte = bytes[index]
+            guard byte >= 48, byte <= 57 else {
                 return nil
             }
-            sawDigit = true
-            let digit = Int(scalar.value - 48)
+            let digit = Int(byte - 48)
             let (multiplied, overflowMul) = value.multipliedReportingOverflow(by: 10)
             guard !overflowMul else {
                 return nil
@@ -175,14 +169,12 @@ public enum ResourceFrameCodec {
                 return nil
             }
             value = added
-        }
-        guard sawDigit else {
-            return nil
+            index += 1
         }
         return value
     }
 
-    private static func lossyUTF8(_ bytes: [UInt8]) -> String {
+    private static func lossyUTF8<Bytes: Collection>(_ bytes: Bytes) -> String where Bytes.Element == UInt8 {
         String(decoding: bytes, as: UTF8.self)
     }
 }

Sources/PeerConnectivityCore/ResourceFrameCodec.swift

@@ -9,7 +9,7 @@ import PeerConnectivityCore
 /// payload), using the muxer's `read/writeLengthPrefixedMessage` helpers:
 ///
 /// ```
-/// [frame] header   : "<name>\0<size>"   (size is the decimal payload byte count)
+/// [frame] header   : "<name>\0<size>\0" (size is the decimal payload byte count)
 /// [frame] chunk     : raw payload bytes (one or more frames)
 /// [frame] chunk     : ...
 /// ```