mirror of
https://github.com/logos-messaging/logos-messaging-nim.git
synced 2026-01-09 09:23:14 +00:00
e7ae1a0382
* Enhanced peer-ex protocol - added rate limiting, added response status and desc to the rpc * Better error result handling for PeerEx request, adjusted tests * Refactored RateLimit configuration option for better CLI UX - now possible to set separate limits per protocol. Adjusted mountings. Added and adjusted tests * Fix libwaku due to changes of error return type of fetchPeerExchangePeers * Fix rate limit setting tests due to changed defaults * Introduce new gauge to help dasboard effectively show current rate limit applied for protocol * Adjust timeing in filter rate limit test to let macos CI test run ok. * Address review findings, namings, error logs, removed left-overs * Changes to reflect latest spec agreement and changes. PeerExchange RPC is changed the now respond structure will contain status_code and status_desc.
143 lines
4.1 KiB
Nim
143 lines
4.1 KiB
Nim
import std/[json, sequtils]
|
|
import chronos, results, libp2p/multiaddress
|
|
import
|
|
../../../../waku/factory/waku,
|
|
../../../../waku/discovery/waku_dnsdisc,
|
|
../../../../waku/discovery/waku_discv5,
|
|
../../../../waku/waku_peer_exchange,
|
|
../../../../waku/waku_core/peers,
|
|
../../../../waku/node/waku_node,
|
|
../../../alloc
|
|
|
|
type DiscoveryMsgType* = enum
|
|
GET_BOOTSTRAP_NODES
|
|
UPDATE_DISCV5_BOOTSTRAP_NODES
|
|
START_DISCV5
|
|
STOP_DISCV5
|
|
PEER_EXCHANGE
|
|
|
|
type DiscoveryRequest* = object
|
|
operation: DiscoveryMsgType
|
|
|
|
## used in GET_BOOTSTRAP_NODES
|
|
enrTreeUrl: cstring
|
|
nameDnsServer: cstring
|
|
timeoutMs: cint
|
|
|
|
## used in UPDATE_DISCV5_BOOTSTRAP_NODES
|
|
nodes: cstring
|
|
|
|
## used in PEER_EXCHANGE
|
|
numPeers: uint64
|
|
|
|
proc createShared(
|
|
T: type DiscoveryRequest,
|
|
op: DiscoveryMsgType,
|
|
enrTreeUrl: cstring,
|
|
nameDnsServer: cstring,
|
|
timeoutMs: cint,
|
|
nodes: cstring,
|
|
numPeers: uint64,
|
|
): ptr type T =
|
|
var ret = createShared(T)
|
|
ret[].operation = op
|
|
ret[].enrTreeUrl = enrTreeUrl.alloc()
|
|
ret[].nameDnsServer = nameDnsServer.alloc()
|
|
ret[].timeoutMs = timeoutMs
|
|
ret[].nodes = nodes.alloc()
|
|
ret[].numPeers = numPeers
|
|
return ret
|
|
|
|
proc createRetrieveBootstrapNodesRequest*(
|
|
T: type DiscoveryRequest,
|
|
op: DiscoveryMsgType,
|
|
enrTreeUrl: cstring,
|
|
nameDnsServer: cstring,
|
|
timeoutMs: cint,
|
|
): ptr type T =
|
|
return T.createShared(op, enrTreeUrl, nameDnsServer, timeoutMs, "", 0)
|
|
|
|
proc createUpdateBootstrapNodesRequest*(
|
|
T: type DiscoveryRequest, op: DiscoveryMsgType, nodes: cstring
|
|
): ptr type T =
|
|
return T.createShared(op, "", "", 0, nodes, 0)
|
|
|
|
proc createDiscV5StartRequest*(T: type DiscoveryRequest): ptr type T =
|
|
return T.createShared(START_DISCV5, "", "", 0, "", 0)
|
|
|
|
proc createDiscV5StopRequest*(T: type DiscoveryRequest): ptr type T =
|
|
return T.createShared(STOP_DISCV5, "", "", 0, "", 0)
|
|
|
|
proc createPeerExchangeRequest*(
|
|
T: type DiscoveryRequest, numPeers: uint64
|
|
): ptr type T =
|
|
return T.createShared(PEER_EXCHANGE, "", "", 0, "", numPeers)
|
|
|
|
proc destroyShared(self: ptr DiscoveryRequest) =
|
|
deallocShared(self[].enrTreeUrl)
|
|
deallocShared(self[].nameDnsServer)
|
|
deallocShared(self[].nodes)
|
|
deallocShared(self)
|
|
|
|
proc retrieveBootstrapNodes(
|
|
enrTreeUrl: string, ipDnsServer: string
|
|
): Result[seq[string], string] =
|
|
let dnsNameServers = @[parseIpAddress(ipDnsServer)]
|
|
let discoveredPeers: seq[RemotePeerInfo] = retrieveDynamicBootstrapNodes(
|
|
true, enrTreeUrl, dnsNameServers
|
|
).valueOr:
|
|
return err("failed discovering peers from DNS: " & $error)
|
|
|
|
var multiAddresses = newSeq[string]()
|
|
|
|
for discPeer in discoveredPeers:
|
|
for address in discPeer.addrs:
|
|
multiAddresses.add($address & "/" & $discPeer)
|
|
|
|
return ok(multiAddresses)
|
|
|
|
proc updateDiscv5BootstrapNodes(nodes: string, waku: ptr Waku): Result[void, string] =
|
|
waku.wakuDiscv5.updateBootstrapRecords(nodes).isOkOr:
|
|
return err("error in updateDiscv5BootstrapNodes: " & $error)
|
|
return ok()
|
|
|
|
proc performPeerExchangeRequestTo(
|
|
numPeers: uint64, waku: ptr Waku
|
|
): Future[Result[int, string]] {.async.} =
|
|
return (await waku.node.fetchPeerExchangePeers(numPeers)).isOkOr:
|
|
return err($error)
|
|
|
|
proc process*(
|
|
self: ptr DiscoveryRequest, waku: ptr Waku
|
|
): Future[Result[string, string]] {.async.} =
|
|
defer:
|
|
destroyShared(self)
|
|
|
|
case self.operation
|
|
of START_DISCV5:
|
|
let res = await waku.wakuDiscv5.start()
|
|
res.isOkOr:
|
|
return err($error)
|
|
|
|
return ok("discv5 started correctly")
|
|
of STOP_DISCV5:
|
|
await waku.wakuDiscv5.stop()
|
|
|
|
return ok("discv5 stopped correctly")
|
|
of GET_BOOTSTRAP_NODES:
|
|
let nodes = retrieveBootstrapNodes($self[].enrTreeUrl, $self[].nameDnsServer).valueOr:
|
|
return err($error)
|
|
|
|
return ok($(%*nodes))
|
|
of UPDATE_DISCV5_BOOTSTRAP_NODES:
|
|
updateDiscv5BootstrapNodes($self[].nodes, waku).isOkOr:
|
|
return err($error)
|
|
|
|
return ok("discovery request processed correctly")
|
|
of PEER_EXCHANGE:
|
|
let numValidPeers = (await performPeerExchangeRequestTo(self[].numPeers, waku)).valueOr:
|
|
return err("error calling performPeerExchangeRequestTo: " & $error)
|
|
return ok($numValidPeers)
|
|
|
|
return err("discovery request not handled")
|