fix: close QUIC streams after receiving FIN

Author: richard-ramos

lsquic.nimble

@@ -1,5 +1,5 @@
 packageName = "lsquic"
-version = "0.5.5"
+version = "0.6.0"
 author = "Status Research & Development GmbH"
 description = "Nim wrapper around the lsquic library"
 license = "MIT"

lsquic/context/stream.nim

@@ -104,6 +104,10 @@ proc onRead*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl.
 
   streamCtx.toRead = Opt.none(ReadTask)
 
+  if n == 0 and not streamCtx.closeIfDone():
+    error "could not close stream after EOF", streamId = lsquic_stream_id(stream)
+    streamCtx.abort()
+
 proc onWrite*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl.} =
   trace "onWrite"
 

lsquic/stream.nim

@@ -21,6 +21,7 @@ type Stream* = ref object
   quicStream*: ptr lsquic_stream_t
   closedByEngine*: bool
   closeWrite*: bool
+  closeRequested: bool
   # This is called when on_close callback is executed
   closed*: AsyncEvent
   # Reuse a single closed-event waiter to minimize allocations on hot paths.
@@ -132,23 +133,39 @@ template raiseIfWriteReset(stream: Stream) =
   if stream.writeResetByPeer():
     raise stream.newStreamResetError("stream write")
 
-proc abort*(stream: Stream) =
+proc requestClose(stream: Stream): bool {.raises: [].} =
+  if stream.closedByEngine or stream.quicStream.isNil or stream.closeRequested:
+    return true
+
+  stream.closeRequested = true
+  let ret = lsquic_stream_close(stream.quicStream)
+  if ret != 0:
+    let closeErrno = errno
+    if closeErrno == EBADF:
+      stream.doProcess()
+      return true
+
+    stream.closeRequested = false
+    trace "could not close stream",
+      streamId = lsquic_stream_id(stream.quicStream), errno = closeErrno
+    return false
+
+  stream.doProcess()
+  true
+
+proc closeIfDone*(stream: Stream): bool {.raises: [].} =
   if stream.closeWrite and stream.isEof:
-    if not stream.closed.isSet():
-      stream.closed.fire()
-    stream.abortPendingWrites("stream aborted")
-    return
+    return stream.requestClose()
 
-  if not stream.closedByEngine:
-    let ret = lsquic_stream_close(stream.quicStream)
-    if ret != 0:
-      trace "could not abort stream", streamId = lsquic_stream_id(stream.quicStream)
-    stream.doProcess()
+  true
 
+proc abort*(stream: Stream) =
   stream.closeWrite = true
   stream.isEof = true
+  discard stream.requestClose()
   stream.abortPendingWrites("stream aborted")
-  stream.closed.fire()
+  if not stream.closed.isSet():
+    stream.closed.fire()
 
 proc close*(stream: Stream) {.async: (raises: [StreamError, CancelledError]).} =
   if stream.closeWrite or stream.closedByEngine:
@@ -157,14 +174,13 @@ proc close*(stream: Stream) {.async: (raises: [StreamError, CancelledError]).} =
   # Closing only the write side
   let ret = lsquic_stream_shutdown(stream.quicStream, 1)
   if ret == 0:
-    if stream.isEof:
-      if lsquic_stream_close(stream.quicStream) != 0:
-        stream.abort()
-        raise newException(StreamError, "could not close the stream")
-
     stream.abortPendingWrites("stream closed")
     stream.closeWrite = true
-    stream.doProcess()
+    if not stream.closeIfDone():
+      stream.abort()
+      raise newException(StreamError, "could not close the stream")
+    if not stream.isEof:
+      stream.doProcess()
   else:
     raise newException(StreamError, "could not close the stream")
 
@@ -200,6 +216,9 @@ proc readOnce*(
 
   if n == 0:
     stream.isEof = true
+    if not stream.closeIfDone():
+      stream.abort()
+      raise newException(StreamError, "could not close the stream")
     return 0
   elif n > 0:
     return n
@@ -231,6 +250,7 @@ proc readOnce*(
       raiseIfReadReset(stream)
       stream.isEof = true
       stream.closeWrite = true
+      discard stream.closeIfDone()
       return 0
 
     return await doneFut

tests/test_lifecycle.nim

@@ -267,6 +267,48 @@ suite "lifecycle":
     check (await reading) == 0
     await incomingStream.close()
 
+  asyncTest "close then EOF retires peer-initiated stream credit":
+    const StreamCount = 120
+    const StreamCreditTimeout = 30.seconds
+    let peers = await connectPeers()
+    defer:
+      await stopPeers(peers)
+
+    proc openStreamWithTimeout(conn: Connection): Future[Stream] {.async:(raises:[CancelledError, ConnectionError]).} =
+      let opening = conn.openStream()
+      if not await opening.withTimeout(StreamCreditTimeout):
+        await opening.cancelAndWait()
+        doAssert false, "timed out opening stream; close+EOF should retire stream credit"
+      await opening
+
+    proc serve() {.async.} =
+      for i in 0 ..< StreamCount:
+        let stream = await peers.incoming.incomingStream()
+        var buf = newSeq[byte](1)
+        check (await stream.readOnce(buf)) == 1
+        check buf[0] == byte(i mod 256)
+
+        await stream.write(@[byte((i + 1) mod 256)])
+        await stream.close()
+        check (await stream.readOnce(buf)) == 0
+
+    let serving = serve()
+
+    for i in 0 ..< StreamCount:
+      let stream = await peers.outgoing.openStreamWithTimeout()
+      await stream.write(@[byte(i mod 256)])
+      await stream.close()
+
+      var buf = newSeq[byte](1)
+      check (await stream.readOnce(buf)) == 1
+      check buf[0] == byte((i + 1) mod 256)
+      check (await stream.readOnce(buf)) == 0
+
+    if not await serving.withTimeout(StreamCreditTimeout):
+      await serving.cancelAndWait()
+      doAssert false, "timed out serving streams; close+EOF should not leave stream handlers blocked"
+    await serving
+
   asyncTest "cancelled blocked read clears pending read":
     let peers = await connectPeers()
     defer: