status-im
/
nim-eth
mirror of https://github.com/status-im/nim-eth.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fix improper yield usage in rlpx and refine exception handling (#679) 

* Fix improper yield usage in rlpx and refine exception handling

* Handle post hello step error
This commit is contained in:
andri lim 2024-02-19 14:16:33 +07:00 committed by GitHub
parent efe610e27f
commit d8209f623f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 83 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
eth/p2p/discovery.nim
View File

@ -104,7 +104,13 @@ proc send(d: DiscoveryProtocol, n: Node, data: seq[byte]) =
let f = d.transp.sendTo(ta, data)
let cb = proc(data: pointer) {.gcsafe.} =
if f.failed:
debug "Discovery send failed", msg = f.readError.msg
when defined(chronicles_log_level):
try:
# readError will raise FutureError
debug "Discovery send failed", msg = f.readError.msg
except FutureError as exc:
error "Failed to get discovery send future error", msg=exc.msg
f.addCallback cb
proc sendPing*(d: DiscoveryProtocol, n: Node): seq[byte] =

40
eth/p2p/p2p_protocol_dsl.nim
View File

@ -335,8 +335,12 @@ proc augmentUserHandler(p: P2PProtocol, userHandlerProc: NimNode, msgId = -1) =
userHandlerProc.addPragma ident"gcsafe"
# we only take the pragma
let dummy = quote do:
proc dummy(): Future[void] {.async: (raises: [EthP2PError]).}
if p.isRlpx:
userHandlerProc.addPragma ident"async"
userHandlerProc.addPragma dummy.pragma[0]
var
getState = ident"getState"
@ -375,7 +379,17 @@ proc augmentUserHandler(p: P2PProtocol, userHandlerProc: NimNode, msgId = -1) =
template networkState(`peerVar`: `PeerType`): `NetworkStateType` {.used.} =
`NetworkStateType`(`getNetworkState`(`peerVar`.network, `protocolInfo`))
proc addPreludeDefs*(userHandlerProc: NimNode, definitions: NimNode) =
proc addExceptionHandler(userHandlerProc: NimNode) =
let bodyTemp = userHandlerProc.body
userHandlerProc.body = quote do:
try:
`bodyTemp`
except CancelledError as exc:
raise newException(EthP2PError, exc.msg)
except CatchableError as exc:
raise newException(EthP2PError, exc.msg)
proc addPreludeDefs(userHandlerProc: NimNode, definitions: NimNode) =
userHandlerProc.body[0].add definitions
proc eventHandlerToProc(p: P2PProtocol, doBlock: NimNode, handlerName: string): NimNode =
@ -385,6 +399,7 @@ proc eventHandlerToProc(p: P2PProtocol, doBlock: NimNode, handlerName: string):
doBlock.copyChildrenTo(result)
result.name = ident(p.name & handlerName) # genSym(nskProc, p.name & handlerName)
p.augmentUserHandler result
result.addExceptionHandler()
proc addTimeoutParam(procDef: NimNode, defaultValue: int64) =
var
@ -470,7 +485,8 @@ proc newMsg(protocol: P2PProtocol, kind: MessageKind, id: int,
if protocol.useRequestIds:
initResponderCall.add reqIdVar
userHandler.addPreludeDefs newVarStmt(responseVar, initResponderCall)
userHandler.addPreludeDefs quote do:
var `responseVar` {.used.} = `initResponderCall`
result.initResponderCall = initResponderCall
@ -479,6 +495,7 @@ proc newMsg(protocol: P2PProtocol, kind: MessageKind, id: int,
of msgResponse: userHandler.applyDecorator protocol.incomingResponseDecorator
else: discard
userHandler.addExceptionHandler()
result.userHandler = userHandler
protocol.outRecvProcs.add result.userHandler
@ -796,12 +813,17 @@ proc createHandshakeTemplate*(msg: Message,
let peerVar = genSym(nskLet ,"peer")
handshakeExchanger.setBody quote do:
let `peerVar` = `peerValue`
let sendingFuture = `forwardCall`
`handshakeImpl`(`peerVar`,
sendingFuture,
`nextMsg`(`peerVar`, `msgRecName`),
`timeoutVar`)
try:
let `peerVar` = `peerValue`
let sendingFuture = `forwardCall`
`handshakeImpl`(`peerVar`,
sendingFuture,
`nextMsg`(`peerVar`, `msgRecName`),
`timeoutVar`)
except PeerDisconnected as exc:
raise newException(EthP2PError, exc.msg)
except P2PInternalError as exc:
raise newException(EthP2PError, exc.msg)
return handshakeExchanger

20
eth/p2p/private/p2p_types.nim
View File

@ -84,17 +84,21 @@ type
name*: string
version*: int
UnsupportedProtocol* = object of Defect
EthP2PError* = object of CatchableError
UnsupportedProtocol* = object of EthP2PError
# This is raised when you attempt to send a message from a particular
# protocol to a peer that doesn't support the protocol.
MalformedMessageError* = object of CatchableError
UnsupportedMessageError* = object of CatchableError
MalformedMessageError* = object of EthP2PError
UnsupportedMessageError* = object of EthP2PError
PeerDisconnected* = object of CatchableError
PeerDisconnected* = object of EthP2PError
reason*: DisconnectionReason
UselessPeerError* = object of CatchableError
UselessPeerError* = object of EthP2PError
P2PInternalError* = object of EthP2PError
##
## Quasy-private types. Use at your own risk.
@ -155,7 +159,7 @@ type
MessageHandlerDecorator* = proc(msgId: int, n: NimNode): NimNode
ThunkProc* = proc(x: Peer, msgId: int, data: Rlp): Future[void]
{.gcsafe, async: (raises: [RlpError, CatchableError]).}
{.gcsafe, async: (raises: [RlpError, EthP2PError]).}
MessageContentPrinter* = proc(msg: pointer): string
{.gcsafe, raises: [].}
@ -173,10 +177,10 @@ type
{.gcsafe, raises: [].}
HandshakeStep* = proc(peer: Peer): Future[void]
{.gcsafe, raises: [].}
{.gcsafe, async: (raises: [EthP2PError]).}
DisconnectionHandler* = proc(peer: Peer, reason: DisconnectionReason):
Future[void] {.gcsafe, raises: [].}
Future[void] {.gcsafe, async: (raises: [EthP2PError]).}
ConnectionState* = enum
None,

49
eth/p2p/rlpx.nim
View File

@ -176,7 +176,7 @@ proc messagePrinter[MsgType](msg: pointer): string {.gcsafe.} =
# result = $(cast[ptr MsgType](msg)[])
proc disconnect*(peer: Peer, reason: DisconnectionReason,
notifyOtherPeer = false) {.async: (raises:[CatchableError]).}
notifyOtherPeer = false) {.async: (raises:[]).}
template raisePeerDisconnected(msg: string, r: DisconnectionReason) =
var e = newException(PeerDisconnected, msg)
@ -185,7 +185,8 @@ template raisePeerDisconnected(msg: string, r: DisconnectionReason) =
proc disconnectAndRaise(peer: Peer,
reason: DisconnectionReason,
msg: string) {.async.} =
msg: string) {.async:
(raises: [PeerDisconnected]).} =
let r = reason
await peer.disconnect(r)
raisePeerDisconnected(msg, r)
@ -193,24 +194,26 @@ proc disconnectAndRaise(peer: Peer,
proc handshakeImpl[T](peer: Peer,
sendFut: Future[void],
responseFut: Future[T],
timeout: Duration): Future[T] {.async.} =
timeout: Duration): Future[T] {.async:
(raises: [PeerDisconnected, P2PInternalError]).} =
sendFut.addCallback do (arg: pointer) {.gcsafe.}:
if sendFut.failed:
debug "Handshake message not delivered", peer
doAssert timeout.milliseconds > 0
yield responseFut or sleepAsync(timeout)
if not responseFut.finished:
try:
let res = await responseFut.wait(timeout)
return res
except AsyncTimeoutError:
# TODO: Really shouldn't disconnect and raise everywhere. In order to avoid
# understanding what error occured where.
# And also, incoming and outgoing disconnect errors should be seperated,
# probably by seperating the actual disconnect call to begin with.
await disconnectAndRaise(peer, HandshakeTimeout,
"Protocol handshake was not received in time.")
elif responseFut.failed:
raise responseFut.error
else:
return responseFut.read
except CatchableError as exc:
raise newException(P2PInternalError, exc.msg)
# Dispatcher
#
@ -363,7 +366,7 @@ template perPeerMsgId(peer: Peer, MsgType: type): int =
perPeerMsgIdImpl(peer, MsgType.msgProtocol.protocolInfo, MsgType.msgId)
proc invokeThunk*(peer: Peer, msgId: int, msgData: Rlp): Future[void]
{.async: (raises: [CatchableError, rlp.RlpError]).} =
{.async: (raises: [rlp.RlpError, EthP2PError]).} =
template invalidIdError: untyped =
raise newException(UnsupportedMessageError,
"RLPx message with an invalid id " & $msgId &
@ -393,6 +396,7 @@ proc sendMsg*(peer: Peer, data: seq[byte]) {.async.} =
if res != len(cipherText):
# This is ECONNRESET or EPIPE case when remote peer disconnected.
await peer.disconnect(TcpError)
discard
except CatchableError as e:
await peer.disconnect(TcpError)
raise e
@ -948,7 +952,7 @@ proc p2pProtocolBackendImpl*(protocol: P2PProtocol): Backend =
proc `thunkName`(`peerVar`: `Peer`, _: int, data: Rlp)
# Fun error if you just use `RlpError` instead of `rlp.RlpError`:
# "Error: type expected, but got symbol 'RlpError' of kind 'EnumField'"
{.async: (raises: [rlp.RlpError, CatchableError]).} =
{.async: (raises: [rlp.RlpError, EthP2PError]).} =
var `receivedRlp` = data
var `receivedMsg` {.noinit.}: `msgRecName`
`readParamsPrelude`
@ -1067,14 +1071,14 @@ proc removePeer(network: EthereumNode, peer: Peer) =
observer.onPeerDisconnected(peer)
proc callDisconnectHandlers(peer: Peer, reason: DisconnectionReason):
Future[void] {.async.} =
Future[void] {.async: (raises: []).} =
var futures = newSeqOfCap[Future[void]](protocolCount())
for protocol in peer.dispatcher.activeProtocols:
if protocol.disconnectHandler != nil:
futures.add((protocol.disconnectHandler)(peer, reason))
await allFutures(futures)
await noCancel allFutures(futures)
for f in futures:
doAssert(f.finished())
@ -1082,7 +1086,7 @@ proc callDisconnectHandlers(peer: Peer, reason: DisconnectionReason):
trace "Disconnection handler ended with an error", err = f.error.msg
proc disconnect*(peer: Peer, reason: DisconnectionReason,
notifyOtherPeer = false) {.async: (raises: [CatchableError]).} =
notifyOtherPeer = false) {.async: (raises: []).} =
if peer.connectionState notin {Disconnecting, Disconnected}:
peer.connectionState = Disconnecting
# Do this first so sub-protocols have time to clean up and stop sending
@ -1094,15 +1098,16 @@ proc disconnect*(peer: Peer, reason: DisconnectionReason,
await callDisconnectHandlers(peer, reason)
if notifyOtherPeer and not peer.transport.closed:
var fut = peer.sendDisconnectMsg(DisconnectionReasonList(value: reason))
yield fut
if fut.failed:
debug "Failed to deliver disconnect message", peer
proc waitAndClose(peer: Peer, time: Duration) {.async.} =
await sleepAsync(time)
await peer.transport.closeWait()
try:
await peer.sendDisconnectMsg(DisconnectionReasonList(value: reason))
except CatchableError as exc:
debug "Failed to deliver disconnect message", peer, msg=exc.msg
# Give the peer a chance to disconnect
traceAsyncErrors peer.waitAndClose(2.seconds)
elif not peer.transport.closed:
@ -1336,7 +1341,7 @@ proc rlpxConnect*(node: EthereumNode, remote: Node):
10.seconds)
except RlpError:
return err(ProtocolError)
except PeerDisconnected as e:
except PeerDisconnected:
return err(PeerDisconnectedError)
# TODO: Strange compiler error
# case e.reason:
@ -1350,6 +1355,8 @@ proc rlpxConnect*(node: EthereumNode, remote: Node):
# return err(PeerDisconnectedError)
except TransportError:
return err(P2PTransportError)
except P2PInternalError:
return err(P2PHandshakeError)
except CatchableError as e:
raiseAssert($e.name & " " & $e.msg)
@ -1374,6 +1381,8 @@ proc rlpxConnect*(node: EthereumNode, remote: Node):
return err(UselessRlpxPeerError)
except TransportError:
return err(P2PTransportError)
except EthP2PError:
return err(ProtocolError)
except CatchableError as e:
raiseAssert($e.name & " " & $e.msg)
@ -1385,7 +1394,7 @@ proc rlpxConnect*(node: EthereumNode, remote: Node):
# TODO: rework rlpxAccept similar to rlpxConnect.
proc rlpxAccept*(
node: EthereumNode, transport: StreamTransport): Future[Peer] {.async: (raises: [CatchableError]).} =
node: EthereumNode, transport: StreamTransport): Future[Peer] {.async: (raises: []).} =
initTracing(devp2pInfo, node.protocols)
let peer = Peer(transport: transport, network: node)

5
tests/p2p/eth_protocol.nim
View File

@ -33,7 +33,10 @@ p2pProtocol eth(version = 63,
genesisHash: KeccakHash)
requestResponse:
proc getBlockHeaders(peer: Peer, request: openArray[KeccakHash]) {.gcsafe.} = discard
proc getBlockHeaders(peer: Peer, request: openArray[KeccakHash]) {.gcsafe.} =
var headers: seq[BlockHeader]
await response.send(headers)
proc blockHeaders(p: Peer, headers: openArray[BlockHeader])
requestResponse: