Stream receives report the end of the stream

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 {

Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift

@@ -528,7 +528,8 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol<InboundStreamLinkag
         }
     }
 
-    func read(minimumBytes: Int, maximumBytes: Int) -> [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<InboundStreamLinkag
                     return nil
                 }
                 var returnBuffer: [UInt8]? = nil
+                var reachedEndOfStream = false
                 frames.iterateMutableFrames { frame in
                     var buffer = [UInt8]()
                     let length = frame.unclaimedLength
+                    // The FIN rides on the frames, set by the socket bottom on EOF and by QUIC on
+                    // its last data frame. They are finalized below, so it has to leave with the
+                    // content or the caller cannot recover it.
+                    if frame.connectionComplete {
+                        reachedEndOfStream = true
+                    }
                     if length > 0 {
                         _ = Deserializer.deserialize(&frame, claim: false) { read throws(DeserializationError) in
                             try read.buffer(&buffer, length: length)
@@ -558,7 +566,10 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol<InboundStreamLinkag
                     frame.finalize(success: true)
                     return true
                 }
-                return returnBuffer
+                guard let returnBuffer else {
+                    return nil
+                }
+                return (content: returnBuffer, isComplete: reachedEndOfStream)
             } catch {
                 return nil
             }

Sources/SwiftNetwork/Protocols/Frame.swift

@@ -125,6 +125,13 @@ public struct Frame: ~Copyable {
         self.effectiveBufferLength = self.bufferLength
     }
 
+    /// A frame of `count` zero bytes carrying `connectionComplete`, for the marker a stream bottom
+    /// hands over once the peer has half-closed and its queue has drained.
+    init(count: Int, connectionComplete: Bool) {
+        self.init(count: count)
+        self.connectionComplete = connectionComplete
+    }
+
     init(bytes: consuming NetworkUniqueArray<UInt8>) {
         self._bytes = bytes
         self.buffer = .bytes

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,15 +275,15 @@ 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 {
-        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
     }
 }

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,27 +573,51 @@ 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()
         }
         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.

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()

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
@@ -664,4 +705,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()
+    }
 }