better `upgraded` lifetime handling (avoid NPE) (#506)

* avoid npe on connection upgrade

* add `onUpgraded` event

libp2p/connmanager.nim

@@ -240,9 +240,7 @@ proc cleanupConn(c: ConnManager, conn: Connection) {.async.} =
 proc onConnUpgraded(c: ConnManager, conn: Connection) {.async.} =
   try:
     trace "Triggering connect events", conn
-    doAssert(not isNil(conn.upgraded),
+    conn.upgrade()
-      "The `upgraded` event hasn't been properly initialized!")
-    conn.upgraded.complete()
 
     let peerId = conn.peerInfo.peerId
     await c.triggerPeerEvents(

libp2p/stream/connection.nim

@@ -36,6 +36,21 @@ type
 
 proc timeoutMonitor(s: Connection) {.async, gcsafe.}
 
+proc isUpgraded*(s: Connection): bool =
+  if not isNil(s.upgraded):
+    return s.upgraded.finished
+
+proc upgrade*(s: Connection, failed: ref Exception = nil) =
+  if not isNil(failed):
+    s.upgraded.fail(failed)
+    return
+
+  s.upgraded.complete()
+
+proc onUpgrade*(s: Connection) {.async.} =
+  if not isNil(s.upgraded):
+    await s.upgraded
+
 func shortLog*(conn: Connection): string =
   if conn.isNil: "Connection(nil)"
   elif conn.peerInfo.isNil: $conn.oid

libp2p/switch.nim

@@ -244,10 +244,8 @@ proc upgradeIncoming(s: Switch, incomingConn: Connection) {.async, gcsafe.} = #
       await ms.handle(cconn)
     except CatchableError as exc:
       debug "Exception in secure handler during incoming upgrade", msg = exc.msg, conn
-      if  not isNil(cconn) and
+      if not cconn.isUpgraded:
-          not isNil(cconn.upgraded) and
+        cconn.upgrade(exc)
-          not(cconn.upgraded.finished):
-        cconn.upgraded.fail(exc)
     finally:
       if not isNil(cconn):
         await cconn.close()
@@ -265,10 +263,8 @@ proc upgradeIncoming(s: Switch, incomingConn: Connection) {.async, gcsafe.} = #
     await ms.handle(incomingConn, active = true)
   except CatchableError as exc:
     debug "Exception upgrading incoming", exc = exc.msg
-    if  not isNil(incomingConn) and
+    if not incomingConn.isUpgraded:
-        not isNil(incomingConn.upgraded) and
+      incomingConn.upgrade(exc)
-        not(incomingConn.upgraded.finished):
-      incomingConn.upgraded.fail(exc)
   finally:
     if not isNil(incomingConn):
       await incomingConn.close()
@@ -441,7 +437,7 @@ proc upgradeMonitor(conn: Connection, upgrades: AsyncSemaphore) {.async.} =
     # upgrade, this timeout guarantees that a
     # "hanged" remote doesn't hold the upgrade
     # forever
-    await conn.upgraded.wait(30.seconds) # wait for connection to be upgraded
+    await conn.onUpgrade.wait(30.seconds) # wait for connection to be upgraded
     trace "Connection upgrade succeeded"
   except CatchableError as exc:
     libp2p_failed_upgrades_incoming.inc()