fix(quic): muxer handling (#1885)

Author: vladopajic

libp2p/transports/quictransport.nim

@@ -121,20 +121,15 @@ method close*(session: QuicSession) {.async: (raises: []).} =
   safeClose(session.connection)
   await procCall P2PConnection(session).close()
 
-proc getStream*(
+proc getStream(
     session: QuicSession, direction = Direction.In
-): Future[QuicStream] {.async: (raises: [QuicTransportError]).} =
+): Future[QuicStream] {.async: (raises: [CancelledError, QuicError]).} =
   var stream: Stream
-  try:
-    case direction
-    of Direction.In:
-      stream = await session.connection.incomingStream()
-    of Direction.Out:
-      stream = await session.connection.openStream()
-  except CancelledError as exc:
-    raise (ref QuicTransportError)(msg: "cancelled getStream: " & exc.msg, parent: exc)
-  except QuicError as exc:
-    raise (ref QuicTransportError)(msg: "error in getStream: " & exc.msg, parent: exc)
+  case direction
+  of Direction.In:
+    stream = await session.connection.incomingStream()
+  of Direction.Out:
+    stream = await session.connection.openStream()
 
   let qs =
     QuicStream.new(stream, session.observedAddr, session.localAddr, session.peerId)
@@ -181,7 +176,7 @@ method newStream*(
 .} =
   try:
     return await m.session.getStream(Direction.Out)
-  except QuicTransportError as exc:
+  except QuicError as exc:
     raise newException(MuxerError, "error in newStream: " & exc.msg, exc)
 
 method handle*(m: QuicMuxer): Future[void] {.async: (raises: []).} =
@@ -192,19 +187,30 @@ method handle*(m: QuicMuxer): Future[void] {.async: (raises: []).} =
     trace "finished handling stream"
     doAssert(chann.closed, "connection not closed by handler!")
 
-  try:
-    while not m.session.atEof:
-      let stream = await m.session.getStream(Direction.In)
-      asyncSpawn handleStream(stream)
-  except QuicTransportError as exc:
-    trace "Exception in quic handler", msg = exc.msg
+  while not m.session.atEof:
+    let stream =
+      try:
+        await m.session.getStream(Direction.In)
+      except CancelledError:
+        # keep handling, until connection is closed
+        continue
+      except QuicError as exc:
+        if exc.msg == "connection closed":
+          # stop handling, connection was closed
+          break
+        else:
+          # keep handling, until connection is closed. 
+          # this stream failed but we need to keep handling for other streams.
+          trace "QuicMuxer.handler got error while opening stream", msg = exc.msg
+          continue
+    asyncSpawn handleStream(stream)
 
 method close*(m: QuicMuxer) {.async: (raises: []).} =
   try:
     await m.session.close()
     if not isNil(m.handleFut):
       m.handleFut.cancelSoon()
-  except CatchableError as exc:
+  except CatchableError:
     discard
 
 # Transport

tests/transports/testquic.nim

@@ -94,7 +94,8 @@ suite "Quic transport":
       # client should be able to write even when server has not accepted
       let client = await createTransport()
       let conn = await client.dial("", server.addrs[0])
-      let stream = await getStream(QuicSession(conn), Direction.Out)
+      let muxer = QuicMuxer.new(conn)
+      let stream = await muxer.newStream()
       await stream.write("client")
       await client.stop()
 

tests/transports/utils.nim

@@ -47,18 +47,21 @@ proc isQuicTransport*(ma: MultiAddress): bool =
 proc createServerAcceptConn*(
     server: QuicTransport
 ): proc(): Future[void] {.
-  async: (raises: [transport.TransportError, LPStreamError, CancelledError])
+  async: (raises: [transport.TransportError, LPError, LPStreamError, CancelledError])
 .} =
   proc handler() {.
-      async: (raises: [transport.TransportError, LPStreamError, CancelledError])
+      async:
+        (raises: [transport.TransportError, LPError, LPStreamError, CancelledError])
   .} =
     let conn = await server.accept()
     if conn == nil:
       return
 
-    let stream = await getStream(QuicSession(conn), Direction.In)
+    let muxer = QuicMuxer.new(conn)
+    let stream = await muxer.newStream()
     defer:
       await stream.close()
+      await muxer.close()
 
     var resp: array[6, byte]
     await stream.readExactly(addr resp, 6)