From a9a18749fd3b5223f5685e6172719b212dd4e8f3 Mon Sep 17 00:00:00 2001 From: Rick Newton-Rogers Date: Fri, 21 Aug 2026 13:59:27 -0400 Subject: [PATCH 1/3] `FrameArray.connectionComplete` is about the final frame The property reported true if *any* frame carried `connectionComplete`, which makes it a question about the array's contents rather than about the stream. A stream ends after its last byte, so a flag on a frame with bytes behind it does not describe a stream that has ended, and reading it that way reports the end early. This matches QUIC and TCP, which both put the end-of-stream marker on the last byte and allow nothing after it. Nothing was relying on the any-frame reading; there was simply nothing stating where the flag may sit. The only place that marks an inbound frame already puts it on the last one. Reading the final frame states the rule, and makes a stray or stale mark irrelevant rather than harmful, so no writer has to be policed. `markEndOfStream` no longer needs to explain itself, and `ProtocolStreamHandlers.receiveStreamData`, which gates on the array flag to release data below the caller's minimum, can no longer be tricked into releasing early. --- .../SwiftNetwork/Protocols/FrameArray.swift | 12 +++------- .../SwiftNetworkFrameArrayTests.swift | 22 +++++++++++++++++++ 2 files changed, 25 insertions(+), 9 deletions(-) diff --git a/Sources/SwiftNetwork/Protocols/FrameArray.swift b/Sources/SwiftNetwork/Protocols/FrameArray.swift index dd0bd45..1f1285a 100644 --- a/Sources/SwiftNetwork/Protocols/FrameArray.swift +++ b/Sources/SwiftNetwork/Protocols/FrameArray.swift @@ -267,15 +267,9 @@ public struct FrameArray: ~Copyable { return length } + /// Whether this array ends the stream, which is a question about its final frame. public var connectionComplete: Bool { - var connectionComplete = false - iterateImmutableFrames { frame in - if frame.connectionComplete { - connectionComplete = true - return false - } - return true - } - return connectionComplete + if frames.isEmpty { return false } + return frames[frames.count - 1].connectionComplete } } diff --git a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift index 4572631..de3ab59 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift @@ -664,4 +664,26 @@ final class SwiftNetworkFrameArrayTests: NetTestCase { XCTAssertEqual(collectBytes(array), [4, 5, 6, 1, 2, 3]) array.finalizeAllFramesAsFailed() } + func testConnectionCompleteIsAboutTheFinalFrame() { + var head = Frame(copyBuffer: [1, 2, 3] as [UInt8]) + head.connectionComplete = true + var array = FrameArray(frame: head) + XCTAssertTrue(array.connectionComplete) + + array.add(frame: Frame(copyBuffer: [4, 5, 6] as [UInt8])) + XCTAssertFalse(array.connectionComplete, "bytes follow the marked frame, so the stream has not ended") + + var tail = Frame(copyBuffer: [7, 8, 9] as [UInt8]) + tail.connectionComplete = true + array.add(frame: tail) + XCTAssertTrue(array.connectionComplete) + + array.finalizeAllFramesAsFailed() + } + + func testConnectionCompleteIsFalseWhenEmpty() { + var array = FrameArray() + XCTAssertFalse(array.connectionComplete) + array.finalizeAllFramesAsFailed() + } } From f9cd7285d56f0601c49d6660411d37e0b95e1ba5 Mon Sep 17 00:00:00 2001 From: Rick Newton-Rogers Date: Thu, 20 Aug 2026 10:24:20 -0400 Subject: [PATCH 2/3] `FrameArray` keeps the end-of-stream flag with the last byte With the array reporting the flag from its final frame, whatever writes the flag has to put it there, and a byte-limited drain has to keep it there. `markEndOfStream()` marks the final frame; the socket bottom previously marked every queued frame, which reported the end of the stream while bytes were still queued. `drainArray` moves the flag to the tail it retains when it splits a frame: one branch swaps the original out as the returned prefix, and since `swap` exchanges whole frame values the flag would otherwise leave with the prefix and be lost from the bytes that remain. Which frame carries the flag is an invariant of the container rather than of whatever fills it, and it is the invariant `drainArray` already relies on, so both belong here. --- .../SwiftNetwork/Protocols/FrameArray.swift | 14 +++++++ .../SwiftNetworkFrameArrayTests.swift | 41 +++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/Sources/SwiftNetwork/Protocols/FrameArray.swift b/Sources/SwiftNetwork/Protocols/FrameArray.swift index 1f1285a..1618a48 100644 --- a/Sources/SwiftNetwork/Protocols/FrameArray.swift +++ b/Sources/SwiftNetwork/Protocols/FrameArray.swift @@ -238,6 +238,14 @@ public struct FrameArray: ~Copyable { let claimed = frames[0].claim(fromStart: 0, fromEnd: partialBytesToKeep) precondition(claimed) + // The flag belongs with the last byte, which is the tail being kept. `swap` + // exchanges whole frame values, so it would carry the flag out with the prefix + // instead. Move it first, while each name still refers to what it says. + if frames[0].connectionComplete { + frames[0].connectionComplete = false + splitFrame.connectionComplete = true + } + // Swap the new frame with the original frame swap(&splitFrame, &frames[0]) @@ -267,6 +275,12 @@ public struct FrameArray: ~Copyable { return length } + /// Marks the end of the stream, on the last frame only. + public mutating func markEndOfStream() { + if frames.isEmpty { return } + frames[frames.count - 1].connectionComplete = true + } + /// Whether this array ends the stream, which is a question about its final frame. public var connectionComplete: Bool { if frames.isEmpty { return false } diff --git a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift index de3ab59..f2e9a6c 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift @@ -569,6 +569,47 @@ final class SwiftNetworkFrameArrayTests: NetTestCase { array.finalizeAllFramesAsFailed() } + func testMarkEndOfStreamOnlyMarksTheFinalFrame() { + var array = FrameArray(frame: Frame(copyBuffer: Array(repeating: 1, count: 100))) + array.add(frame: Frame(copyBuffer: Array(repeating: 2, count: 100))) + array.add(frame: Frame(copyBuffer: Array(repeating: 3, count: 100))) + array.markEndOfStream() + + var first = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(first.unclaimedLength, 100) + XCTAssertFalse(first.connectionComplete, "end of stream reported with 200 bytes still queued") + + var second = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(second.unclaimedLength, 100) + XCTAssertFalse(second.connectionComplete, "end of stream reported with 100 bytes still queued") + + var third = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(third.unclaimedLength, 100) + XCTAssertTrue(third.connectionComplete, "end of stream not reported with the last bytes") + + first.finalizeAllFramesAsFailed() + second.finalizeAllFramesAsFailed() + third.finalizeAllFramesAsFailed() + } + + func testDrainByteCountSplitLeavesEndOfStreamWithTheTrailingBytes() { + var frame = Frame(copyBuffer: Array(0..<20)) + frame.connectionComplete = true + var array = FrameArray(frame: frame) + + var drained = array.drainArray(maximumByteCount: 12) + XCTAssertEqual(drained.unclaimedLength, 12) + XCTAssertFalse(drained.connectionComplete, "end of stream reported before the last byte") + XCTAssertEqual(array.unclaimedLength, 8) + XCTAssertTrue(array.connectionComplete, "end of stream lost from the remaining bytes") + + var rest = array.drainArray(maximumByteCount: 8) + XCTAssertTrue(rest.connectionComplete) + drained.finalizeAllFramesAsFailed() + rest.finalizeAllFramesAsFailed() + array.finalizeAllFramesAsFailed() + } + func testDrainByteCountSplitMajoritySentToNewArray() { // Split where the majority of the split frame's bytes go to the new array // Drain 18 of 20 bytes from a single frame From 48b4a9d8c8ed2e2f5970481d8af03f1475a748b2 Mon Sep 17 00:00:00 2001 From: Rick Newton-Rogers Date: Thu, 20 Aug 2026 10:24:20 -0400 Subject: [PATCH 3/3] Stream receives report the end of the stream At the moment `Message.isComplete` is always false for a stream receive, whatever the sender did: the flag travels on a frame as `connectionComplete` and the socket bottom sets it, but the stream read path accumulated only the bytes and dropped the flag before any consumer could see it. That leaves a consumer inferring half-closure from a receive completing with no content, which is a different mechanism and an unreliable one. It works only when the FIN arrives in a read event of its own; when it arrives alongside data, `drainArray` hands the sentinel over with the payload, so there is no empty receive either and the next one never completes. Nothing in the library reads the flag, which is why no test caught it. `read(minimumBytes:maximumBytes:)` now returns whether the stream ended alongside the content, and the socket bottom marks the queued bytes rather than appending a sentinel. That means no frame has to be allocated at EOF, and keying the decision on the half-closed state rather than on a queued frame is what lets the marker be synthesized once the queue has already drained. Delivery is recorded from the drained result as well, so the common path where the flag rides out on real data does not then report a second, empty end of stream. --- .../EndpointFlow/EndpointFlow.swift | 10 +++- .../EndpointFlow/EndpointFlowProtocols.swift | 15 ++++- Sources/SwiftNetwork/Protocols/Frame.swift | 7 +++ .../Protocols/SocketProtocol.swift | 57 +++++++++++++------ .../SwiftNetworkConnectionTests.swift | 56 ++++++++++++++++++ 5 files changed, 124 insertions(+), 21 deletions(-) diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift index 95bf494..e70eac0 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift @@ -235,12 +235,16 @@ final class EndpointFlow: CustomDebugStringConvertible { case .stream(let flow): while true { if let readRequest = self.readRequests.first { - if let content = flow.read( + if let streamRead = flow.read( minimumBytes: readRequest.minimumBytes, maximumBytes: readRequest.maximumBytes ) { - // TODO: Get the actual metadata - readRequest.complete(content: content, isComplete: false, isFinal: true) + // TODO: Get the remaining per-frame metadata + readRequest.complete( + content: streamRead.content, + isComplete: streamRead.isComplete, + isFinal: true + ) // TODO: This is not efficient. Probably better to use an ArraySlice here self.readRequests.removeFirst() } else { diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift index 0dd8bf6..ea0b5c1 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift @@ -528,7 +528,8 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol [UInt8]? { + /// Reads buffered stream bytes, reporting whether the read reached the end of the stream. + func read(minimumBytes: Int, maximumBytes: Int) -> (content: [UInt8], isComplete: Bool)? { fromExternal { do throws(NetworkError) { guard @@ -542,9 +543,16 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol 0 { _ = Deserializer.deserialize(&frame, claim: false) { read throws(DeserializationError) in try read.buffer(&buffer, length: length) @@ -558,7 +566,10 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol) { self._bytes = bytes self.buffer = .bytes diff --git a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift index c598c82..e94a650 100644 --- a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift +++ b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift @@ -429,6 +429,8 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC private var isConnecting = false private var inputSourceSuspended = false private var inputFinished = false + /// Whether the end-of-stream marker has been delivered to the consumer. + private var inputFinishedDelivered = false private var outputFinished = false private var pendingDisconnect = false private var incomingFrames = FrameArray() @@ -571,20 +573,33 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC // MARK: - BottomStreamProtocol public func receiveStreamData(minimumBytes: Int, maximumBytes: Int) throws(NetworkError) -> FrameArray? { - guard !incomingFrames.isEmpty, - incomingFrames.unclaimedLength >= minimumBytes || incomingFrames.connectionComplete - else { - // We don't have enough buffered to satisfy the consumer yet. If we - // had suspended on the high-water mark, resume — the consumer needs - // more than we're currently holding, so reading must continue even - // past the soft cap. Otherwise a large minimum would deadlock. - if inputSourceSuspended && !inputFinished { - inputSourceSuspended = false - dispatchReadSource?.resume() + if incomingFrames.isEmpty { + // Nothing buffered and nothing more coming, so report the end of the stream. + // It is synthesized rather than carried on a queued frame, so nothing has to + // be allocated to hold the flag once the queue has drained. + if inputFinished, !inputFinishedDelivered { + inputFinishedDelivered = true + return FrameArray(frame: Frame(count: 0, connectionComplete: true)) } + + resumeReadingIfSuspended() + return nil + } + + // Either hold the consumer until its minimum can be met or if the peer has + // half-closed no more will arrive, so what is buffered is all it is ever going to get. + guard incomingFrames.unclaimedLength >= minimumBytes || inputFinished else { + resumeReadingIfSuspended() return nil } + let result = incomingFrames.drainArray(maximumByteCount: maximumBytes) + // The flag sits on whichever frame carries the last byte, so a drain that takes it counts + // as delivery. Without this the branch above would synthesize a second, empty end of stream + // on the next receive. + if result.connectionComplete { + inputFinishedDelivered = true + } if inputSourceSuspended, incomingFrames.unclaimedLength < maximumInputSize { inputSourceSuspended = false dispatchReadSource?.resume() @@ -592,6 +607,17 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC return result } + /// Resumes the read source if it was suspended on the high-water mark. + /// + /// The consumer wants more than is buffered, so reading has to continue past the soft cap or a + /// large minimum would deadlock. Not once the peer has half-closed: the source is cancelled by + /// then and no further event can arrive. + private func resumeReadingIfSuspended() { + guard inputSourceSuspended, !inputFinished else { return } + inputSourceSuspended = false + dispatchReadSource?.resume() + } + public func getOutboundStreamDataRoomAvailable() throws(NetworkError) -> Int { let pending = pendingOutputFrames.unclaimedLength if pending >= maximumOutputSize { return 0 } @@ -756,7 +782,7 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC switch result { case .processed(let bytesRead): if bytesRead == 0 { - // Stream EOF — mark the next frame as connectionComplete. + // Stream EOF — the queued bytes are marked below. reachedEOF = true } else { let frame = Frame(copyBuffer: UnsafeRawBufferPointer(start: readBuffer, count: bytesRead)) @@ -773,11 +799,10 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC if reachedEOF { inputFinished = true - // Tag the last incoming frame (or an empty one) with connectionComplete - // so the upper protocol sees stream completion. - var sentinel = Frame(count: 0) - sentinel.connectionComplete = true - incomingFrames.add(frame: sentinel) + // Mark the bytes already queued, so the read that returns the last of them also + // reports the end of the stream. With nothing queued, `receiveStreamData` + // synthesizes the marker instead. + incomingFrames.markEndOfStream() receivedAny = true // The stream is finished; stop the read source. EOF keeps the // descriptor readable, so leaving it armed would spin forever. diff --git a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift index 603e5a0..81535e3 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift @@ -657,6 +657,62 @@ final class SwiftNetworkConnectionTests: NetTestCase { ) } + // A stream send marked `isComplete` must arrive marked complete. + func testStreamReceiveReportsEndOfStream() { + let group = DispatchGroup() + group.enter() + let c1 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7878), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7877)) + ) + .onStateUpdate { _, state in + if case .cancelled = state { group.leave() } + } + + group.enter() + let c2 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7877), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7878)) + ) + .onStateUpdate { _, state in + if case .cancelled = state { group.leave() } + } + + c1.start() + c2.start() + + c1.send(.message(content: [1, 2, 3], isComplete: true)) { result in + if case .failure(let error) = result { + XCTFail("send failed with error \(error)") + } + } + + c2.receive(atLeast: 1, atMost: Int.max) { result in + switch result { + case .success(let message): + XCTAssertEqual(message.content, [1, 2, 3]) + XCTAssertTrue(message.isComplete, "end of stream was not reported to the receiver") + case .failure(let error): + XCTFail("receive failed with error \(error)") + } + c1.cancel() + c2.cancel() + } + + XCTAssertEqual( + group.wait(timeout: DispatchTime.now() + .seconds(5)), + DispatchTimeoutResult.success + ) + } + #if HAS_SWIFTTLS_RECORD func testTLSNoTransportDataPath() { let group = DispatchGroup()