Deliver RESET_STREAM on a peer-initiated stream's first frame

Sources/SwiftNetwork/QUIC/QUICFrame.swift

@@ -1088,13 +1088,24 @@ struct FrameResetStream: ~Copyable, QUICFrameProtocol {
             connection.close(with: .streamStateError, "RESET_STREAM for send-only stream")
             return false
         }
-        guard let flowID = connection.knownFlows[streamID] else {
-            return true
-        }
-        let potentialFlow = connection.flow(for: flowID)
-        if potentialFlow == nil {
+
+        let stream: QUICStreamInstance
+        if let flowID = connection.knownFlows[streamID] {
+            // The stream id is known. The flow object may still be missing
+            // if the stream was torn down without clearing `knownFlows`; in
+            // that case there is no one to deliver the reset to, so drop it.
+            guard let existing = connection.flow(for: flowID) else {
+                Logger.proto.error(
+                    "stream \(streamID.value) has known flow but no QUICStreamInstance; dropping RESET_STREAM"
+                )
+                return true
+            }
+            stream = existing
+
+        } else {
+            // RESET_STREAM may be the first frame the peer sends on a stream
+            // createInboundStreams will register it.
             let inboundStreamResult = connection.createInboundStreams(streamID: streamID)
-            // Use Int(exactly:) here to make sure that finalSize UInt64 can fit into Int64
             if inboundStreamResult.checkZombie {
                 connection.zombieStreamListFinalSizeReceived(
                     streamID: streamID,
@@ -1103,14 +1114,20 @@ struct FrameResetStream: ~Copyable, QUICFrameProtocol {
                 return true
             }
             if !inboundStreamResult.created {
-                return false
+                // The stream is being ignored. createInboundStreams already
+                // drove any required peer notification or close, so treat the
+                // frame as handled.
+                return true
             }
-        }
-        guard let stream = potentialFlow else {
-            Logger.proto.error(
-                "client state for flow \(flowID.debugDescription) is not a QUICStreamInstance"
-            )
-            return true
+            guard let flowID = connection.knownFlows[streamID],
+                let created = connection.flow(for: flowID)
+            else {
+                Logger.proto.error(
+                    "stream \(streamID.value) is not a QUICStreamInstance after createInboundStreams"
+                )
+                return true
+            }
+            stream = created
         }
         // Set the application error code
         stream.streamMetadata.applicationError = self.code

Tests/SwiftNetworkTests/QUICTestHarness.swift

@@ -1340,6 +1340,65 @@ final class QUICTestHarness {
         stop()
     }
 
+    /// `RESET_STREAM` is the very first frame the server sees for a fresh
+    /// peer-initiated bidi stream — no preceding `STREAM` frame. The server
+    /// must still create the inbound flow and deliver the inbound-aborted
+    /// event for the reset to be observable end-to-end.
+    func runQUICResetStreamFirstFrameOnNewStream(
+        applicationError: UInt64 = 42,
+        timeout: TimeInterval = 4.0
+    ) {
+        do {
+            try quicHandshake()
+        } catch {
+            XCTFail("Handshake failed: \(error)")
+            return
+        }
+
+        guard let clientStream = createNewStream(identifier: "C1") else {
+            XCTFail("Failed to create client stream")
+            return
+        }
+
+        let serverFlowExpectation = XCTestExpectation(description: "Server sees new flow")
+        let serverAbortExpectation = XCTestExpectation(description: "Server sees inbound abort event")
+        var serverStream: StreamUpperHarness?
+
+        context.async {
+            self.state?.serverHarness.waitForNewFlow {
+                guard let stream = self.state?.serverHarness.upperHarnesses.last else {
+                    XCTFail("Server flow missing")
+                    serverFlowExpectation.fulfill()
+                    return
+                }
+                serverStream = stream
+                stream.waitForInboundAborted { error in
+                    XCTAssertNotNil(error, "Server should see inbound abort from RESET_STREAM")
+                    XCTAssertEqual(
+                        error?.quicApplicationError,
+                        Int64(applicationError),
+                        "Application error code should be propagated from RESET_STREAM"
+                    )
+                    serverAbortExpectation.fulfill()
+                }
+                serverFlowExpectation.fulfill()
+            }
+        }
+
+        // Reset with no prior write so the only frame the client emits for
+        // this stream id is RESET_STREAM. There is no STREAM frame to
+        // stride-create the flow ahead of the reset.
+        context.async {
+            clientStream.abortOutbound(error: .init(quicApplicationError: applicationError))
+        }
+
+        wait(for: [serverFlowExpectation], timeout: timeout)
+        wait(for: [serverAbortExpectation], timeout: timeout)
+
+        _ = serverStream  // silence unused-warning; harness keeps strong ref
+        stop()
+    }
+
     func runQUICServerTestForPendingBidirectional(
         identifier: String = #function,
         dataBlock: [UInt8],

Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift

@@ -362,6 +362,10 @@ final class SwiftNetworkQUICHarnessTests: NetTestCase {
         QUICTestHarness().runQUICNewStreamWithImmediateAbort(abortKind: .stopSending)
     }
 
+    func testQUICResetStreamFirstFrameOnNewStream() {
+        QUICTestHarness().runQUICResetStreamFirstFrameOnNewStream()
+    }
+
     func testQUICResetStreamDoesNotAffectOppositeDirection() {
         QUICTestHarness().runQUICTest(
             streamCount: 1,