fix: clear pending stream I/O tasks on cancellation

lsquic/stream.nim

@@ -90,6 +90,40 @@ proc abortPendingWrites*(stream: Stream, error: ref StreamError) {.raises: [].}
 proc abortPendingWrites*(stream: Stream, reason: string = "") {.raises: [].} =
   stream.abortPendingWrites(newException(StreamError, reason))
 
+proc clearPendingRead(
+    stream: Stream, doneFut: Future[int].Raising([CancelledError, StreamError])
+) {.raises: [].} =
+  let task = stream.toRead.valueOr:
+    return
+  if task.doneFut != doneFut:
+    return
+
+  stream.toRead = Opt.none(ReadTask)
+
+  if stream.closedByEngine or stream.quicStream.isNil:
+    return
+
+  if lsquic_stream_wantread(stream.quicStream, 0) == -1:
+    error "could not set stream wantread",
+      streamId = lsquic_stream_id(stream.quicStream)
+
+proc clearPendingWrite(
+    stream: Stream, doneFut: Future[void].Raising([CancelledError, StreamError])
+) {.raises: [].} =
+  let task = stream.toWrite.valueOr:
+    return
+  if task.doneFut != doneFut:
+    return
+
+  stream.toWrite = Opt.none(WriteTask)
+
+  if stream.closedByEngine or stream.quicStream.isNil:
+    return
+
+  if lsquic_stream_wantwrite(stream.quicStream, 0) == -1:
+    error "could not set stream wantwrite",
+      streamId = lsquic_stream_id(stream.quicStream)
+
 template raiseIfReadReset(stream: Stream) =
   if stream.readResetByPeer():
     raise stream.newStreamResetError("stream read")
@@ -187,18 +221,21 @@ proc readOnce*(
     Future[int].Raising([CancelledError, StreamError]).init("Stream.readOnce")
   stream.toRead = Opt.some(ReadTask(data: dst, dataLen: dstLen, doneFut: doneFut))
 
-  stream.doProcess()
+  try:
+    stream.doProcess()
 
-  let raceFut = await race(stream.closedWaiter, doneFut)
-  if raceFut == stream.closedWaiter:
-    if not doneFut.finished:
-      await doneFut.cancelAndWait()
-    raiseIfReadReset(stream)
-    stream.isEof = true
-    stream.closeWrite = true
-    return 0
+    let raceFut = await race(stream.closedWaiter, doneFut)
+    if raceFut == stream.closedWaiter:
+      if not doneFut.finished:
+        await doneFut.cancelAndWait()
+      raiseIfReadReset(stream)
+      stream.isEof = true
+      stream.closeWrite = true
+      return 0
 
-  return await doneFut
+    return await doneFut
+  finally:
+    stream.clearPendingRead(doneFut)
 
 template readOnce*(stream: Stream, dst: var openArray[byte]): untyped =
   ## Convenience helper that forwards an openArray/seq to the pointer-based API.
@@ -256,13 +293,16 @@ proc write*(
     WriteTask(data: data[0].addr, dataLen: data.len, doneFut: doneFut, offset: n)
   )
 
-  stream.doProcess()
+  try:
+    stream.doProcess()
 
-  let raceFut = await race(stream.closedWaiter, doneFut)
-  if raceFut == stream.closedWaiter:
-    raiseIfWriteReset(stream)
-    if not doneFut.finished:
-      doneFut.fail(newException(StreamError, "stream closed"))
-    stream.closeWrite = true
+    let raceFut = await race(stream.closedWaiter, doneFut)
+    if raceFut == stream.closedWaiter:
+      raiseIfWriteReset(stream)
+      if not doneFut.finished:
+        doneFut.fail(newException(StreamError, "stream closed"))
+      stream.closeWrite = true
 
-  await doneFut
+    await doneFut
+  finally:
+    stream.clearPendingWrite(doneFut)

tests/test_lifecycle.nim

@@ -203,6 +203,29 @@ suite "lifecycle":
     check (await incomingStream.readOnce(buf)) == 0
     await incomingStream.close()
 
+  asyncTest "cancel pending write clears stream write task":
+    let peers = await connectPeers()
+    defer:
+      await stopPeers(peers)
+
+    let outgoingStream = await peers.outgoing.openStream()
+    let writing = outgoingStream.write(newData(16 * 1024 * 1024, 0x7A'u8))
+
+    var observedPending = false
+    for _ in 0 ..< 200:
+      if outgoingStream.toWrite.isSome:
+        observedPending = true
+        break
+      if writing.finished:
+        break
+      await sleepAsync(10.milliseconds)
+
+    check observedPending
+    if not writing.finished:
+      await writing.cancelAndWait()
+
+    check outgoingStream.toWrite.isNone()
+
   asyncTest "read once returns zero repeatedly after eof":
     let peers = await connectPeers()
     defer:
@@ -244,6 +267,43 @@ suite "lifecycle":
     check (await reading) == 0
     await incomingStream.close()
 
+  asyncTest "cancelled blocked read clears pending read":
+    let peers = await connectPeers()
+    defer:
+      await stopPeers(peers)
+
+    let incomingWaiting = peers.incoming.incomingStream()
+    let outgoingStream = await peers.outgoing.openStream()
+    await outgoingStream.write(@[1'u8])
+    check (await incomingWaiting.withTimeout(2.seconds))
+    let incomingStream = await incomingWaiting
+
+    var firstByte = newSeq[byte](1)
+    check (await incomingStream.readOnce(firstByte)) == 1
+    check firstByte[0] == 1
+
+    var cancelledBuf = newSeq[byte](8)
+    let reading = incomingStream.readOnce(cancelledBuf)
+
+    await sleepAsync(100.milliseconds)
+    check not reading.finished
+    check incomingStream.toRead.isSome
+
+    await reading.cancelAndWait()
+
+    check incomingStream.toRead.isNone()
+
+    await outgoingStream.write(@[99'u8])
+    await sleepAsync(100.milliseconds)
+    check cancelledBuf[0] == 0
+
+    var buf = newSeq[byte](1)
+    let nextRead = incomingStream.readOnce(buf)
+    check (await nextRead.withTimeout(2.seconds))
+    check (await nextRead) == 1
+    check buf[0] == 99
+    await incomingStream.close()
+
   asyncTest "peer reset":
     let peers = await connectPeers()
     defer: