chore(connmanager): propagate CancelledError (#1276)

Author: vladopajic

libp2p/connmanager.nim

@@ -42,8 +42,9 @@ type
     else:
       discard
 
-  ConnEventHandler* =
-    proc(peerId: PeerId, event: ConnEvent): Future[void] {.gcsafe, async: (raises: []).}
+  ConnEventHandler* = proc(peerId: PeerId, event: ConnEvent): Future[void] {.
+    gcsafe, async: (raises: [CancelledError])
+  .}
 
   PeerEventKind* {.pure.} = enum
     Left
@@ -57,8 +58,9 @@ type
     else:
       discard
 
-  PeerEventHandler* =
-    proc(peerId: PeerId, event: PeerEvent): Future[void] {.gcsafe, async: (raises: []).}
+  PeerEventHandler* = proc(peerId: PeerId, event: PeerEvent): Future[void] {.
+    gcsafe, async: (raises: [CancelledError])
+  .}
 
   ConnManager* = ref object of RootObj
     maxConnsPerPeer: int

libp2p/protocols/connectivity/autonat/service.nim

@@ -216,7 +216,7 @@ method setup*(
     if self.askNewConnectedPeers:
       self.newConnectedPeerHandler = proc(
           peerId: PeerId, event: PeerEvent
-      ): Future[void] {.async: (raises: []).} =
+      ): Future[void] {.async: (raises: [CancelledError]).} =
         discard askPeer(self, switch, peerId)
 
       switch.connManager.addPeerEventHandler(

libp2p/protocols/connectivity/relay/relay.nim

@@ -343,7 +343,7 @@ proc handleStreamV1(
 proc setup*(r: Relay, switch: Switch) =
   r.switch = switch
   r.switch.addPeerEventHandler(
-    proc(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+    proc(peerId: PeerId, event: PeerEvent) {.async: (raises: [CancelledError]).} =
       r.rsvp.del(peerId),
     Left,
   )

libp2p/protocols/pubsub/pubsub.nim

@@ -680,7 +680,9 @@ proc init*[PubParams: object | bool](
         topicsHigh: int.high,
       )
 
-  proc peerEventHandler(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+  proc peerEventHandler(
+      peerId: PeerId, event: PeerEvent
+  ) {.async: (raises: [CancelledError]).} =
     if event.kind == PeerEventKind.Joined:
       pubsub.subscribePeer(peerId)
     else:

libp2p/protocols/rendezvous.nim

@@ -699,7 +699,9 @@ proc unsubscribe*(
 
 proc setup*(rdv: RendezVous, switch: Switch) =
   rdv.switch = switch
-  proc handlePeer(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+  proc handlePeer(
+      peerId: PeerId, event: PeerEvent
+  ) {.async: (raises: [CancelledError]).} =
     if event.kind == PeerEventKind.Joined:
       rdv.peers.add(peerId)
     elif event.kind == PeerEventKind.Left:

libp2p/services/autorelayservice.nim

@@ -72,12 +72,14 @@ method setup*(
   if hasBeenSetUp:
     proc handlePeerIdentified(
         peerId: PeerId, event: PeerEvent
-    ) {.async: (raises: []).} =
+    ) {.async: (raises: [CancelledError]).} =
       trace "Peer Identified", peerId
       if self.relayPeers.len < self.maxNumRelays:
         self.peerAvailable.fire()
 
-    proc handlePeerLeft(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+    proc handlePeerLeft(
+        peerId: PeerId, event: PeerEvent
+    ) {.async: (raises: [CancelledError]).} =
       trace "Peer Left", peerId
       self.relayPeers.withValue(peerId, future):
         future[].cancel()

libp2p/services/hpservice.nim

@@ -65,7 +65,7 @@ proc closeRelayConn(relayedConn: Connection) {.async: (raises: [CancelledError])
 
 proc newConnectedPeerHandler(
     self: HPService, switch: Switch, peerId: PeerId, event: PeerEvent
-) {.async: (raises: []).} =
+) {.async: (raises: [CancelledError]).} =
   try:
     # Get all connections to the peer. If there is at least one non-relayed connection, return.
     let connections = switch.connManager.getConnections()[peerId].mapIt(it.connection)
@@ -88,6 +88,8 @@ proc newConnectedPeerHandler(
         switch.peerInfo.listenAddrs.mapIt(switch.peerStore.guessDialableAddr(it))
     await dcutrClient.startSync(switch, peerId, natAddrs)
     await closeRelayConn(relayedConn)
+  except CancelledError as err:
+    raise err
   except CatchableError as err:
     debug "Hole punching failed during dcutr", err = err.msg
 
@@ -106,11 +108,8 @@ method setup*(
 
     self.newConnectedPeerHandler = proc(
         peerId: PeerId, event: PeerEvent
-    ) {.async: (raises: []).} =
-      try:
-        await newConnectedPeerHandler(self, switch, peerId, event)
-      except CancelledError:
-        trace "hole punching cancelled"
+    ) {.async: (raises: [CancelledError]).} =
+      await newConnectedPeerHandler(self, switch, peerId, event)
 
     switch.connManager.addPeerEventHandler(
       self.newConnectedPeerHandler, PeerEventKind.Joined

tests/testswitch.nim

@@ -316,7 +316,7 @@ suite "Switch":
 
     var step = 0
     var kinds: set[ConnEventKind]
-    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: []).} =
+    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: [CancelledError]).} =
       kinds = kinds + {event.kind}
       case step
       of 0:
@@ -364,7 +364,7 @@ suite "Switch":
 
     var step = 0
     var kinds: set[ConnEventKind]
-    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: []).} =
+    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: [CancelledError]).} =
       kinds = kinds + {event.kind}
       case step
       of 0:
@@ -412,7 +412,9 @@ suite "Switch":
 
     var step = 0
     var kinds: set[PeerEventKind]
-    proc handler(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+    proc handler(
+        peerId: PeerId, event: PeerEvent
+    ) {.async: (raises: [CancelledError]).} =
       kinds = kinds + {event.kind}
       case step
       of 0:
@@ -459,7 +461,9 @@ suite "Switch":
 
     var step = 0
     var kinds: set[PeerEventKind]
-    proc handler(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+    proc handler(
+        peerId: PeerId, event: PeerEvent
+    ) {.async: (raises: [CancelledError]).} =
       kinds = kinds + {event.kind}
       case step
       of 0:
@@ -512,7 +516,9 @@ suite "Switch":
 
     var step = 0
     var kinds: set[PeerEventKind]
-    proc handler(peerId: PeerId, event: PeerEvent) {.async: (raises: []).} =
+    proc handler(
+        peerId: PeerId, event: PeerEvent
+    ) {.async: (raises: [CancelledError]).} =
       kinds = kinds + {event.kind}
       case step
       of 0:
@@ -570,7 +576,7 @@ suite "Switch":
     var switches: seq[Switch]
     var done = newFuture[void]()
     var onConnect: Future[void]
-    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: []).} =
+    proc hook(peerId: PeerId, event: ConnEvent) {.async: (raises: [CancelledError]).} =
       try:
         case event.kind
         of ConnEventKind.Connected:
@@ -579,6 +585,8 @@ suite "Switch":
         of ConnEventKind.Disconnected:
           check not switches[0].isConnected(peerInfo.peerId)
           done.complete()
+      except CancelledError as e:
+        raise e
       except CatchableError:
         check false # should not get here
 
@@ -608,7 +616,7 @@ suite "Switch":
     var switches: seq[Switch]
     var done = newFuture[void]()
     var onConnect: Future[void]
-    proc hook(peerId2: PeerId, event: ConnEvent) {.async: (raises: []).} =
+    proc hook(peerId2: PeerId, event: ConnEvent) {.async: (raises: [CancelledError]).} =
       try:
         case event.kind
         of ConnEventKind.Connected:
@@ -623,6 +631,8 @@ suite "Switch":
             check not switches[0].isConnected(peerInfo.peerId)
             done.complete()
           conns.dec
+      except CancelledError as e:
+        raise e
       except CatchableError:
         check false # should not get here