logos-messaging/logos-messaging-nim
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-nim.git synced 2026-01-07 16:33:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(refactoring): results lib refactors (mostly replace isOk) (#3610)

Browse Source 
* Changes isOk usage into better patterns with e.g. valueOr / isOkOr
* Some other refactoring included
* This PR partially addresses #1969
This commit is contained in:
Fabiana Cecin 2025-10-08 19:14:54 -03:00 committed by GitHub
parent 4b0bb29aa9
commit 4d68e2abd5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 182 additions and 284 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
apps/benchmarks/benchmarks.nim
View File

@ -28,10 +28,9 @@ proc benchmark(
iter = i, elapsed_ms = (getTime() - start_time).inMilliseconds iter = i, elapsed_ms = (getTime() - start_time).inMilliseconds
discard await manager.updateRoots() discard await manager.updateRoots()
let proofResult = await manager.fetchMerkleProofElements() manager.merkleProofCache = (await manager.fetchMerkleProofElements()).valueOr:
if proofResult.isErr(): error "Failed to fetch Merkle proof", error = error
error "Failed to fetch Merkle proof", error = proofResult.error quit(QuitFailure)
manager.merkleProofCache = proofResult.get()
let epoch = default(Epoch) let epoch = default(Epoch)
debug "epoch in bytes", epochHex = epoch.inHex() debug "epoch in bytes", epochHex = epoch.inHex()

81
apps/chat2/chat2.nim
View File

@ -132,25 +132,14 @@ proc showChatPrompt(c: Chat) =
except IOError: except IOError:
discard discard
proc getChatLine(c: Chat, msg: WakuMessage): Result[string, string] = proc getChatLine(payload: seq[byte]): string =
# No payload encoding/encryption from Waku # No payload encoding/encryption from Waku
let let pb = Chat2Message.init(payload).valueOr:
pb = Chat2Message.init(msg.payload) return string.fromBytes(payload)
chatLine = return $pb
if pb.isOk:
pb[].toString()
else:
string.fromBytes(msg.payload)
return ok(chatline)
proc printReceivedMessage(c: Chat, msg: WakuMessage) = proc printReceivedMessage(c: Chat, msg: WakuMessage) =
let let chatLine = getChatLine(msg.payload)
pb = Chat2Message.init(msg.payload)
chatLine =
if pb.isOk:
pb[].toString()
else:
string.fromBytes(msg.payload)
try: try:
echo &"{chatLine}" echo &"{chatLine}"
except ValueError: except ValueError:
@ -173,18 +162,16 @@ proc startMetricsServer(
): Result[MetricsHttpServerRef, string] = ): Result[MetricsHttpServerRef, string] =
info "Starting metrics HTTP server", serverIp = $serverIp, serverPort = $serverPort info "Starting metrics HTTP server", serverIp = $serverIp, serverPort = $serverPort
let metricsServerRes = MetricsHttpServerRef.new($serverIp, serverPort) let server = MetricsHttpServerRef.new($serverIp, serverPort).valueOr:
if metricsServerRes.isErr(): return err("metrics HTTP server start failed: " & $error)
return err("metrics HTTP server start failed: " & $metricsServerRes.error)
let server = metricsServerRes.value
try: try:
waitFor server.start() waitFor server.start()
except CatchableError: except CatchableError:
return err("metrics HTTP server start failed: " & getCurrentExceptionMsg()) return err("metrics HTTP server start failed: " & getCurrentExceptionMsg())
info "Metrics HTTP server started", serverIp = $serverIp, serverPort = $serverPort info "Metrics HTTP server started", serverIp = $serverIp, serverPort = $serverPort
ok(metricsServerRes.value) ok(server)
proc publish(c: Chat, line: string) = proc publish(c: Chat, line: string) =
# First create a Chat2Message protobuf with this line of text # First create a Chat2Message protobuf with this line of text
@ -202,19 +189,17 @@ proc publish(c: Chat, line: string) =
version: 0, version: 0,
timestamp: getNanosecondTime(time), timestamp: getNanosecondTime(time),
) )
if not isNil(c.node.wakuRlnRelay): if not isNil(c.node.wakuRlnRelay):
# for future version when we support more than one rln protected content topic, # for future version when we support more than one rln protected content topic,
# we should check the message content topic as well # we should check the message content topic as well
let appendRes = c.node.wakuRlnRelay.appendRLNProof(message, float64(time)) if c.node.wakuRlnRelay.appendRLNProof(message, float64(time)).isErr():
if appendRes.isErr():
debug "could not append rate limit proof to the message" debug "could not append rate limit proof to the message"
else: else:
debug "rate limit proof is appended to the message" debug "rate limit proof is appended to the message"
let decodeRes = RateLimitProof.init(message.proof) let proof = RateLimitProof.init(message.proof).valueOr:
if decodeRes.isErr():
error "could not decode the RLN proof" error "could not decode the RLN proof"
return
let proof = decodeRes.get()
# TODO move it to log after dogfooding # TODO move it to log after dogfooding
let msgEpoch = fromEpoch(proof.epoch) let msgEpoch = fromEpoch(proof.epoch)
if fromEpoch(c.node.wakuRlnRelay.lastEpoch) == msgEpoch: if fromEpoch(c.node.wakuRlnRelay.lastEpoch) == msgEpoch:
@ -438,7 +423,7 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
let resolved = await dnsResolver.resolveTxt(domain) let resolved = await dnsResolver.resolveTxt(domain)
return resolved[0] # Use only first answer return resolved[0] # Use only first answer
var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl.get(), resolver) let wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl.get(), resolver)
if wakuDnsDiscovery.isOk: if wakuDnsDiscovery.isOk:
let discoveredPeers = await wakuDnsDiscovery.get().findPeers() let discoveredPeers = await wakuDnsDiscovery.get().findPeers()
if discoveredPeers.isOk: if discoveredPeers.isOk:
@ -446,8 +431,10 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
discoveredNodes = discoveredPeers.get() discoveredNodes = discoveredPeers.get()
echo "Discovered and connecting to " & $discoveredNodes echo "Discovered and connecting to " & $discoveredNodes
waitFor chat.node.connectToNodes(discoveredNodes) waitFor chat.node.connectToNodes(discoveredNodes)
else:
warn "Failed to find peers via DNS discovery", error = discoveredPeers.error
else: else:
warn "Failed to init Waku DNS discovery" warn "Failed to init Waku DNS discovery", error = wakuDnsDiscovery.error
let peerInfo = node.switch.peerInfo let peerInfo = node.switch.peerInfo
let listenStr = $peerInfo.addrs[0] & "/p2p/" & $peerInfo.peerId let listenStr = $peerInfo.addrs[0] & "/p2p/" & $peerInfo.peerId
@ -483,21 +470,19 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
else: else:
newSeq[byte](0) newSeq[byte](0)
let let chatLine = getChatLine(payload)
pb = Chat2Message.init(payload)
chatLine =
if pb.isOk:
pb[].toString()
else:
string.fromBytes(payload)
echo &"{chatLine}" echo &"{chatLine}"
info "Hit store handler" info "Hit store handler"
let queryRes = await node.query( block storeQueryBlock:
StoreQueryRequest(contentTopics: @[chat.contentTopic]), storenode.get() let queryRes = (
) await node.query(
if queryRes.isOk(): StoreQueryRequest(contentTopics: @[chat.contentTopic]), storenode.get()
storeHandler(queryRes.value) )
).valueOr:
error "Store query failed", error = error
break storeQueryBlock
storeHandler(queryRes)
# NOTE Must be mounted after relay # NOTE Must be mounted after relay
if conf.lightpushnode != "": if conf.lightpushnode != "":
@ -511,8 +496,9 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
error = peerInfo.error error = peerInfo.error
if conf.filternode != "": if conf.filternode != "":
let peerInfo = parsePeerInfo(conf.filternode) if (let peerInfo = parsePeerInfo(conf.filternode); peerInfo.isErr()):
if peerInfo.isOk(): error "Filter not mounted. Couldn't parse conf.filternode", error = peerInfo.error
else:
await node.mountFilter() await node.mountFilter()
await node.mountFilterClient() await node.mountFilterClient()
@ -523,8 +509,6 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
chat.printReceivedMessage(msg) chat.printReceivedMessage(msg)
# TODO: Here to support FilterV2 relevant subscription. # TODO: Here to support FilterV2 relevant subscription.
else:
error "Filter not mounted. Couldn't parse conf.filternode", error = peerInfo.error
# Subscribe to a topic, if relay is mounted # Subscribe to a topic, if relay is mounted
if conf.relay: if conf.relay:
@ -545,11 +529,8 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
proc spamHandler(wakuMessage: WakuMessage) {.gcsafe, closure.} = proc spamHandler(wakuMessage: WakuMessage) {.gcsafe, closure.} =
debug "spam handler is called" debug "spam handler is called"
let chatLineResult = chat.getChatLine(wakuMessage) let chatLineResult = getChatLine(wakuMessage.payload)
if chatLineResult.isOk(): echo "spam message is found and discarded : " & chatLineResult
echo "A spam message is found and discarded : ", chatLineResult.value
else:
echo "A spam message is found and discarded"
chat.prompt = false chat.prompt = false
showChatPrompt(chat) showChatPrompt(chat)

8
apps/chat2bridge/chat2bridge.nim
View File

@ -136,9 +136,7 @@ proc toMatterbridge(
proc pollMatterbridge(cmb: Chat2MatterBridge, handler: MbMessageHandler) {.async.} = proc pollMatterbridge(cmb: Chat2MatterBridge, handler: MbMessageHandler) {.async.} =
while cmb.running: while cmb.running:
let getRes = cmb.mbClient.getMessages() if (let getRes = cmb.mbClient.getMessages(); getRes.isOk()):
if getRes.isOk():
for jsonNode in getRes[]: for jsonNode in getRes[]:
await handler(jsonNode) await handler(jsonNode)
else: else:
@ -169,9 +167,7 @@ proc new*(
let mbClient = MatterbridgeClient.new(mbHostUri, mbGateway) let mbClient = MatterbridgeClient.new(mbHostUri, mbGateway)
# Let's verify the Matterbridge configuration before continuing # Let's verify the Matterbridge configuration before continuing
let clientHealth = mbClient.isHealthy() if mbClient.isHealthy().valueOr(false):
if clientHealth.isOk() and clientHealth[]:
info "Reached Matterbridge host", host = mbClient.host info "Reached Matterbridge host", host = mbClient.host
else: else:
raise newException(ValueError, "Matterbridge client not reachable/healthy") raise newException(ValueError, "Matterbridge client not reachable/healthy")

6
apps/chat2bridge/config_chat2bridge.nim
View File

@ -126,11 +126,9 @@ proc completeCmdArg*(T: type keys.KeyPair, val: string): seq[string] =
return @[] return @[]
proc parseCmdArg*(T: type crypto.PrivateKey, p: string): T = proc parseCmdArg*(T: type crypto.PrivateKey, p: string): T =
let key = SkPrivateKey.init(p) let key = SkPrivateKey.init(p).valueOr:
if key.isOk():
crypto.PrivateKey(scheme: Secp256k1, skkey: key.get())
else:
raise newException(ValueError, "Invalid private key") raise newException(ValueError, "Invalid private key")
return crypto.PrivateKey(scheme: Secp256k1, skkey: key)
proc completeCmdArg*(T: type crypto.PrivateKey, val: string): seq[string] = proc completeCmdArg*(T: type crypto.PrivateKey, val: string): seq[string] =
return @[] return @[]

52
apps/chat2mix/chat2mix.nim
View File

@ -146,25 +146,14 @@ proc showChatPrompt(c: Chat) =
except IOError: except IOError:
discard discard
proc getChatLine(c: Chat, msg: WakuMessage): Result[string, string] = proc getChatLine(payload: seq[byte]): string =
# No payload encoding/encryption from Waku # No payload encoding/encryption from Waku
let let pb = Chat2Message.init(payload).valueOr:
pb = Chat2Message.init(msg.payload) return string.fromBytes(payload)
chatLine = return $pb
if pb.isOk:
pb[].toString()
else:
string.fromBytes(msg.payload)
return ok(chatline)
proc printReceivedMessage(c: Chat, msg: WakuMessage) = proc printReceivedMessage(c: Chat, msg: WakuMessage) =
let let chatLine = getChatLine(msg.payload)
pb = Chat2Message.init(msg.payload)
chatLine =
if pb.isOk:
pb[].toString()
else:
string.fromBytes(msg.payload)
try: try:
echo &"{chatLine}" echo &"{chatLine}"
except ValueError: except ValueError:
@ -375,18 +364,17 @@ proc maintainSubscription(
let peerOpt = selectRandomServicePeer( let peerOpt = selectRandomServicePeer(
wakuNode.peerManager, some(actualFilterPeer), WakuFilterSubscribeCodec wakuNode.peerManager, some(actualFilterPeer), WakuFilterSubscribeCodec
) )
if peerOpt.isOk(): peerOpt.isOkOr:
actualFilterPeer = peerOpt.get()
info "Found new peer for codec",
codec = filterPubsubTopic, peer = constructMultiaddrStr(actualFilterPeer)
noFailedSubscribes = 0
continue # try again with new peer without delay
else:
error "Failed to find new service peer. Exiting." error "Failed to find new service peer. Exiting."
noFailedServiceNodeSwitches += 1 noFailedServiceNodeSwitches += 1
break break
actualFilterPeer = peerOpt.get()
info "Found new peer for codec",
codec = filterPubsubTopic, peer = constructMultiaddrStr(actualFilterPeer)
noFailedSubscribes = 0
continue # try again with new peer without delay
else: else:
if noFailedSubscribes > 0: if noFailedSubscribes > 0:
noFailedSubscribes -= 1 noFailedSubscribes -= 1
@ -530,7 +518,7 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
let resolved = await dnsResolver.resolveTxt(domain) let resolved = await dnsResolver.resolveTxt(domain)
return resolved[0] # Use only first answer return resolved[0] # Use only first answer
var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl.get(), resolver) let wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl.get(), resolver)
if wakuDnsDiscovery.isOk: if wakuDnsDiscovery.isOk:
let discoveredPeers = await wakuDnsDiscovery.get().findPeers() let discoveredPeers = await wakuDnsDiscovery.get().findPeers()
if discoveredPeers.isOk: if discoveredPeers.isOk:
@ -538,8 +526,10 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
discoveredNodes = discoveredPeers.get() discoveredNodes = discoveredPeers.get()
echo "Discovered and connecting to " & $discoveredNodes echo "Discovered and connecting to " & $discoveredNodes
waitFor chat.node.connectToNodes(discoveredNodes) waitFor chat.node.connectToNodes(discoveredNodes)
else:
warn "Failed to find peers via DNS discovery", error = discoveredPeers.error
else: else:
warn "Failed to init Waku DNS discovery" warn "Failed to init Waku DNS discovery", error = wakuDnsDiscovery.error
let peerInfo = node.switch.peerInfo let peerInfo = node.switch.peerInfo
let listenStr = $peerInfo.addrs[0] & "/p2p/" & $peerInfo.peerId let listenStr = $peerInfo.addrs[0] & "/p2p/" & $peerInfo.peerId
@ -575,13 +565,7 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
else: else:
newSeq[byte](0) newSeq[byte](0)
let let chatLine = getChatLine(payload)
pb = Chat2Message.init(payload)
chatLine =
if pb.isOk:
pb[].toString()
else:
string.fromBytes(payload)
echo &"{chatLine}" echo &"{chatLine}"
info "Hit store handler" info "Hit store handler"

25
apps/liteprotocoltester/publisher.nim
View File

@ -190,25 +190,22 @@ proc publishMessages(
) )
if not preventPeerSwitch and noFailedPush > maxFailedPush: if not preventPeerSwitch and noFailedPush > maxFailedPush:
info "Max push failure limit reached, Try switching peer." info "Max push failure limit reached, Try switching peer."
let peerOpt = selectRandomServicePeer( actualServicePeer = selectRandomServicePeer(
wakuNode.peerManager, some(actualServicePeer), WakuLightPushCodec wakuNode.peerManager, some(actualServicePeer), WakuLightPushCodec
) ).valueOr:
if peerOpt.isOk():
actualServicePeer = peerOpt.get()
info "New service peer in use",
codec = lightpushPubsubTopic,
peer = constructMultiaddrStr(actualServicePeer)
noFailedPush = 0
noOfServicePeerSwitches += 1
lpt_change_service_peer_count.inc(labelValues = ["publisher"])
continue # try again with new peer without delay
else:
error "Failed to find new service peer. Exiting." error "Failed to find new service peer. Exiting."
noFailedServiceNodeSwitches += 1 noFailedServiceNodeSwitches += 1
break break
info "New service peer in use",
codec = lightpushPubsubTopic,
peer = constructMultiaddrStr(actualServicePeer)
noFailedPush = 0
noOfServicePeerSwitches += 1
lpt_change_service_peer_count.inc(labelValues = ["publisher"])
continue # try again with new peer without delay
await sleepAsync(messageInterval) await sleepAsync(messageInterval)
proc setupAndPublish*( proc setupAndPublish*(

23
apps/liteprotocoltester/receiver.nim
View File

@ -89,23 +89,20 @@ proc maintainSubscription(
await sleepAsync(2.seconds) # Wait a bit before retrying await sleepAsync(2.seconds) # Wait a bit before retrying
continue continue
elif not preventPeerSwitch: elif not preventPeerSwitch:
let peerOpt = selectRandomServicePeer( actualFilterPeer = selectRandomServicePeer(
wakuNode.peerManager, some(actualFilterPeer), WakuFilterSubscribeCodec wakuNode.peerManager, some(actualFilterPeer), WakuFilterSubscribeCodec
) ).valueOr:
if peerOpt.isOk():
actualFilterPeer = peerOpt.get()
info "Found new peer for codec",
codec = filterPubsubTopic, peer = constructMultiaddrStr(actualFilterPeer)
noFailedSubscribes = 0
lpt_change_service_peer_count.inc(labelValues = ["receiver"])
isFirstPingOnNewPeer = true
continue # try again with new peer without delay
else:
error "Failed to find new service peer. Exiting." error "Failed to find new service peer. Exiting."
noFailedServiceNodeSwitches += 1 noFailedServiceNodeSwitches += 1
break break
info "Found new peer for codec",
codec = filterPubsubTopic, peer = constructMultiaddrStr(actualFilterPeer)
noFailedSubscribes = 0
lpt_change_service_peer_count.inc(labelValues = ["receiver"])
isFirstPingOnNewPeer = true
continue # try again with new peer without delay
else: else:
if noFailedSubscribes > 0: if noFailedSubscribes > 0:
noFailedSubscribes -= 1 noFailedSubscribes -= 1

32
apps/networkmonitor/networkmonitor.nim
View File

@ -1,7 +1,7 @@
{.push raises: [].} {.push raises: [].}
import import
std/[net, tables, strutils, times, sequtils, random], std/[net, tables, strutils, times, sequtils, random, sugar],
results, results,
chronicles, chronicles,
chronicles/topics_registry, chronicles/topics_registry,
@ -183,16 +183,14 @@ proc setConnectedPeersMetrics(
for maddr in peerInfo.addrs: for maddr in peerInfo.addrs:
if $maddr notin customPeerInfo.maddrs: if $maddr notin customPeerInfo.maddrs:
customPeerInfo.maddrs.add $maddr customPeerInfo.maddrs.add $maddr
let typedRecord = discNode.toTypedRecord() let typedRecord = discNode.toTypedRecord().valueOr:
if not typedRecord.isOk():
warn "could not convert record to typed record", record = discNode warn "could not convert record to typed record", record = discNode
continue continue
if not typedRecord.get().ip.isSome(): let ipAddr = typedRecord.ip.valueOr:
warn "ip field is not set", record = typedRecord.get() warn "ip field is not set", record = typedRecord
continue continue
let ip = $typedRecord.get().ip.get().join(".") customPeerInfo.ip = $ipAddr.join(".")
customPeerInfo.ip = ip
# try to ping the peer # try to ping the peer
if shouldReconnect(customPeerInfo): if shouldReconnect(customPeerInfo):
@ -374,14 +372,9 @@ proc retrieveDynamicBootstrapNodes(
if resolved.len > 0: if resolved.len > 0:
return resolved[0] # Use only first answer return resolved[0] # Use only first answer
var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl, resolver) var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl, resolver).errorOr:
if wakuDnsDiscovery.isOk(): return (await value.findPeers()).mapErr(e => $e)
return (await wakuDnsDiscovery.get().findPeers()).mapErr( warn "Failed to init Waku DNS discovery"
proc(e: cstring): string =
$e
)
else:
warn "Failed to init Waku DNS discovery"
debug "No method for retrieving dynamic bootstrap nodes specified." debug "No method for retrieving dynamic bootstrap nodes specified."
ok(newSeq[RemotePeerInfo]()) # Return an empty seq by default ok(newSeq[RemotePeerInfo]()) # Return an empty seq by default
@ -391,11 +384,10 @@ proc getBootstrapFromDiscDns(
): Future[Result[seq[enr.Record], string]] {.async.} = ): Future[Result[seq[enr.Record], string]] {.async.} =
try: try:
let dnsNameServers = @[parseIpAddress("1.1.1.1"), parseIpAddress("1.0.0.1")] let dnsNameServers = @[parseIpAddress("1.1.1.1"), parseIpAddress("1.0.0.1")]
let dynamicBootstrapNodesRes = let dynamicBootstrapNodes = (
await retrieveDynamicBootstrapNodes(conf.dnsDiscoveryUrl, dnsNameServers) await retrieveDynamicBootstrapNodes(conf.dnsDiscoveryUrl, dnsNameServers)
if not dynamicBootstrapNodesRes.isOk(): ).valueOr:
error("failed discovering peers from DNS") return err("Failed retrieving dynamic bootstrap nodes: " & $error)
let dynamicBootstrapNodes = dynamicBootstrapNodesRes.get()
# select dynamic bootstrap nodes that have an ENR containing a udp port. # select dynamic bootstrap nodes that have an ENR containing a udp port.
# Discv5 only supports UDP https://github.com/ethereum/devp2p/blob/master/discv5/discv5-theory.md) # Discv5 only supports UDP https://github.com/ethereum/devp2p/blob/master/discv5/discv5-theory.md)
@ -411,7 +403,7 @@ proc getBootstrapFromDiscDns(
discv5BootstrapEnrs.add(enr) discv5BootstrapEnrs.add(enr)
return ok(discv5BootstrapEnrs) return ok(discv5BootstrapEnrs)
except CatchableError: except CatchableError:
error("failed discovering peers from DNS") error("failed discovering peers from DNS: " & getCurrentExceptionMsg())
proc initAndStartApp( proc initAndStartApp(
conf: NetworkMonitorConf conf: NetworkMonitorConf

4
examples/lightpush_publisher.nim
View File

@ -86,7 +86,9 @@ proc setupAndPublish(rng: ref HmacDrbgContext) {.async.} =
timestamp: now(), timestamp: now(),
) # current timestamp ) # current timestamp
let lightpushPeer = parsePeerInfo(LightpushPeer).get() let lightpushPeer = parsePeerInfo(LightpushPeer).valueOr:
error "failed to parse LightpushPeer", error = error
quit(QuitFailure)
let res = await node.legacyLightpushPublish( let res = await node.legacyLightpushPublish(
some(LightpushPubsubTopic), message, lightpushPeer some(LightpushPubsubTopic), message, lightpushPeer

10
waku/discovery/waku_discv5.nim
View File

@ -421,13 +421,11 @@ proc setupDiscoveryV5*(
addBootstrapNode(enrUri, discv5BootstrapEnrs) addBootstrapNode(enrUri, discv5BootstrapEnrs)
for enr in discv5BootstrapEnrs: for enr in discv5BootstrapEnrs:
let peerInfoRes = enr.toRemotePeerInfo() let peerInfo = enr.toRemotePeerInfo().valueOr:
if peerInfoRes.isOk():
nodePeerManager.addPeer(peerInfoRes.get(), PeerOrigin.Discv5)
else:
debug "could not convert discv5 bootstrap node to peerInfo, not adding peer to Peer Store", debug "could not convert discv5 bootstrap node to peerInfo, not adding peer to Peer Store",
enr = enr.toUri(), error = peerInfoRes.error enr = enr.toUri(), error = error
continue
nodePeerManager.addPeer(peerInfo, PeerOrigin.Discv5)
discv5BootstrapEnrs.add(dynamicBootstrapEnrs) discv5BootstrapEnrs.add(dynamicBootstrapEnrs)
let discv5Config = let discv5Config =

24
waku/discovery/waku_dnsdisc.nim
View File

@ -6,7 +6,7 @@
## EIP-1459 is defined in https://eips.ethereum.org/EIPS/eip-1459 ## EIP-1459 is defined in https://eips.ethereum.org/EIPS/eip-1459
import import
std/[options, net, sequtils], std/[options, net, sequtils, sugar],
chronicles, chronicles,
chronos, chronos,
metrics, metrics,
@ -67,14 +67,11 @@ proc findPeers*(
for enr in discoveredEnr: for enr in discoveredEnr:
# Convert discovered ENR to RemotePeerInfo and add to discovered nodes # Convert discovered ENR to RemotePeerInfo and add to discovered nodes
let res = enr.toRemotePeerInfo() let peerInfo = enr.toRemotePeerInfo().valueOr:
error "Failed to convert ENR to peer info", enr = $enr, error = error
if res.isOk():
discoveredNodes.add(res.get())
else:
error "Failed to convert ENR to peer info", enr = $enr, err = res.error()
waku_dnsdisc_errors.inc(labelValues = ["peer_info_failure"]) waku_dnsdisc_errors.inc(labelValues = ["peer_info_failure"])
continue
discoveredNodes.add(peerInfo)
if discoveredNodes.len > 0: if discoveredNodes.len > 0:
info "Successfully discovered nodes", count = discoveredNodes.len info "Successfully discovered nodes", count = discoveredNodes.len
waku_dnsdisc_discovered.inc(discoveredNodes.len.int64) waku_dnsdisc_discovered.inc(discoveredNodes.len.int64)
@ -117,14 +114,9 @@ proc retrieveDynamicBootstrapNodes*(
if resolved.len > 0: if resolved.len > 0:
return resolved[0] # Use only first answer return resolved[0] # Use only first answer
var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl, resolver) var wakuDnsDiscovery = WakuDnsDiscovery.init(dnsDiscoveryUrl, resolver).errorOr:
if wakuDnsDiscovery.isOk(): return (await value.findPeers()).mapErr(e => $e)
return (await wakuDnsDiscovery.get().findPeers()).mapErr( warn "Failed to init Waku DNS discovery"
proc(e: cstring): string =
$e
)
else:
warn "Failed to init Waku DNS discovery"
debug "No method for retrieving dynamic bootstrap nodes specified." debug "No method for retrieving dynamic bootstrap nodes specified."
ok(newSeq[RemotePeerInfo]()) # Return an empty seq by default ok(newSeq[RemotePeerInfo]()) # Return an empty seq by default

54
waku/factory/node_factory.nim
View File

@ -251,21 +251,15 @@ proc setupProtocols(
mountStoreClient(node) mountStoreClient(node)
if conf.remoteStoreNode.isSome(): if conf.remoteStoreNode.isSome():
let storeNode = parsePeerInfo(conf.remoteStoreNode.get()) let storeNode = parsePeerInfo(conf.remoteStoreNode.get()).valueOr:
if storeNode.isOk(): return err("failed to set node waku store peer: " & error)
node.peerManager.addServicePeer(storeNode.value, store_common.WakuStoreCodec) node.peerManager.addServicePeer(storeNode, WakuStoreCodec)
else:
return err("failed to set node waku store peer: " & storeNode.error)
mountLegacyStoreClient(node) mountLegacyStoreClient(node)
if conf.remoteStoreNode.isSome(): if conf.remoteStoreNode.isSome():
let storeNode = parsePeerInfo(conf.remoteStoreNode.get()) let storeNode = parsePeerInfo(conf.remoteStoreNode.get()).valueOr:
if storeNode.isOk(): return err("failed to set node waku legacy store peer: " & error)
node.peerManager.addServicePeer( node.peerManager.addServicePeer(storeNode, WakuLegacyStoreCodec)
storeNode.value, legacy_common.WakuLegacyStoreCodec
)
else:
return err("failed to set node waku legacy store peer: " & storeNode.error)
if conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume: if conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume:
node.setupStoreResume() node.setupStoreResume()
@ -377,12 +371,10 @@ proc setupProtocols(
mountLightPushClient(node) mountLightPushClient(node)
mountLegacyLightPushClient(node) mountLegacyLightPushClient(node)
if conf.remoteLightPushNode.isSome(): if conf.remoteLightPushNode.isSome():
let lightPushNode = parsePeerInfo(conf.remoteLightPushNode.get()) let lightPushNode = parsePeerInfo(conf.remoteLightPushNode.get()).valueOr:
if lightPushNode.isOk(): return err("failed to set node waku lightpush peer: " & error)
node.peerManager.addServicePeer(lightPushNode.value, WakuLightPushCodec) node.peerManager.addServicePeer(lightPushNode, WakuLightPushCodec)
node.peerManager.addServicePeer(lightPushNode.value, WakuLegacyLightPushCodec) node.peerManager.addServicePeer(lightPushNode, WakuLegacyLightPushCodec)
else:
return err("failed to set node waku lightpush peer: " & lightPushNode.error)
# Filter setup. NOTE Must be mounted after relay # Filter setup. NOTE Must be mounted after relay
if conf.filterServiceConf.isSome(): if conf.filterServiceConf.isSome():
@ -400,16 +392,13 @@ proc setupProtocols(
await node.mountFilterClient() await node.mountFilterClient()
if conf.remoteFilterNode.isSome(): if conf.remoteFilterNode.isSome():
let filterNode = parsePeerInfo(conf.remoteFilterNode.get()) let filterNode = parsePeerInfo(conf.remoteFilterNode.get()).valueOr:
if filterNode.isOk(): return err("failed to set node waku filter peer: " & error)
try: try:
node.peerManager.addServicePeer(filterNode.value, WakuFilterSubscribeCodec) node.peerManager.addServicePeer(filterNode, WakuFilterSubscribeCodec)
except CatchableError: except CatchableError:
return err( return
"failed to mount waku filter client protocol: " & getCurrentExceptionMsg() err("failed to mount waku filter client protocol: " & getCurrentExceptionMsg())
)
else:
return err("failed to set node waku filter peer: " & filterNode.error)
# waku peer exchange setup # waku peer exchange setup
if conf.peerExchangeService: if conf.peerExchangeService:
@ -422,12 +411,9 @@ proc setupProtocols(
err("failed to mount waku peer-exchange protocol: " & getCurrentExceptionMsg()) err("failed to mount waku peer-exchange protocol: " & getCurrentExceptionMsg())
if conf.remotePeerExchangeNode.isSome(): if conf.remotePeerExchangeNode.isSome():
let peerExchangeNode = parsePeerInfo(conf.remotePeerExchangeNode.get()) let peerExchangeNode = parsePeerInfo(conf.remotePeerExchangeNode.get()).valueOr:
if peerExchangeNode.isOk(): return err("failed to set node waku peer-exchange peer: " & error)
node.peerManager.addServicePeer(peerExchangeNode.value, WakuPeerExchangeCodec) node.peerManager.addServicePeer(peerExchangeNode, WakuPeerExchangeCodec)
else:
return
err("failed to set node waku peer-exchange peer: " & peerExchangeNode.error)
if conf.peerExchangeDiscovery: if conf.peerExchangeDiscovery:
await node.mountPeerExchangeClient() await node.mountPeerExchangeClient()

5
waku/factory/validator_signed.nim
View File

@ -67,9 +67,8 @@ proc addSignedShardsValidator*(
if msg.timestamp != 0: if msg.timestamp != 0:
if msg.withinTimeWindow(): if msg.withinTimeWindow():
let msgHash = SkMessage(topic.msgHash(msg)) let msgHash = SkMessage(topic.msgHash(msg))
let recoveredSignature = SkSignature.fromRaw(msg.meta) SkSignature.fromRaw(msg.meta).isErrOr:
if recoveredSignature.isOk(): if value.verify(msgHash, protectedShard.key):
if recoveredSignature.get.verify(msgHash, protectedShard.key):
outcome = errors.ValidationResult.Accept outcome = errors.ValidationResult.Accept
if outcome != errors.ValidationResult.Accept: if outcome != errors.ValidationResult.Accept:

14
waku/utils/noise.nim
View File

@ -12,11 +12,9 @@ proc decodePayloadV2*(
case message.version case message.version
of 2: of 2:
# We attempt to decode the WakuMessage payload # We attempt to decode the WakuMessage payload
let deserializedPayload2 = deserializePayloadV2(message.payload) let deserializedPayload2 = deserializePayloadV2(message.payload).valueOr:
if deserializedPayload2.isOk():
return ok(deserializedPayload2.get())
else:
return err("Failed to decode WakuMessage") return err("Failed to decode WakuMessage")
return ok(deserializedPayload2)
else: else:
return err("Wrong message version while decoding payload") return err("Wrong message version while decoding payload")
@ -28,13 +26,11 @@ proc encodePayloadV2*(
raises: [NoiseMalformedHandshake, NoisePublicKeyError] raises: [NoiseMalformedHandshake, NoisePublicKeyError]
.} = .} =
# We attempt to encode the PayloadV2 # We attempt to encode the PayloadV2
let serializedPayload2 = serializePayloadV2(payload2) let serializedPayload2 = serializePayloadV2(payload2).valueOr:
if not serializedPayload2.isOk():
return err("Failed to encode PayloadV2") return err("Failed to encode PayloadV2")
# If successful, we create and return a WakuMessage # If successful, we create and return a WakuMessage
let msg = WakuMessage( let msg =
payload: serializedPayload2.get(), version: 2, contentTopic: contentTopic WakuMessage(payload: serializedPayload2, version: 2, contentTopic: contentTopic)
)
return ok(msg) return ok(msg)

35
waku/waku_api/rest/legacy_store/handlers.nim
View File

@ -70,30 +70,27 @@ proc parseCursor(
digest: Option[string], digest: Option[string],
): Result[Option[HistoryCursor], string] = ): Result[Option[HistoryCursor], string] =
# Parse sender time # Parse sender time
let parsedSenderTime = parseTime(senderTime) let parsedSenderTime = parseTime(senderTime).valueOr:
if not parsedSenderTime.isOk(): return err(error)
return err(parsedSenderTime.error)
# Parse store time # Parse store time
let parsedStoreTime = parseTime(storeTime) let parsedStoreTime = parseTime(storeTime).valueOr:
if not parsedStoreTime.isOk(): return err(error)
return err(parsedStoreTime.error)
# Parse message digest # Parse message digest
let parsedMsgDigest = parseMsgDigest(digest) let parsedMsgDigest = parseMsgDigest(digest).valueOr:
if not parsedMsgDigest.isOk(): return err(error)
return err(parsedMsgDigest.error)
# Parse cursor information # Parse cursor information
if parsedPubsubTopic.isSome() and parsedSenderTime.value.isSome() and if parsedPubsubTopic.isSome() and parsedSenderTime.isSome() and
parsedStoreTime.value.isSome() and parsedMsgDigest.value.isSome(): parsedStoreTime.isSome() and parsedMsgDigest.isSome():
return ok( return ok(
some( some(
HistoryCursor( HistoryCursor(
pubsubTopic: parsedPubsubTopic.get(), pubsubTopic: parsedPubsubTopic.get(),
senderTime: parsedSenderTime.value.get(), senderTime: parsedSenderTime.get(),
storeTime: parsedStoreTime.value.get(), storeTime: parsedStoreTime.get(),
digest: parsedMsgDigest.value.get(), digest: parsedMsgDigest.get(),
) )
) )
) )
@ -225,16 +222,14 @@ proc installStoreApiHandlers*(
endTime.toOpt(), endTime.toOpt(),
pageSize.toOpt(), pageSize.toOpt(),
ascending.toOpt(), ascending.toOpt(),
) ).valueOr:
return RestApiResponse.badRequest(error)
if not histQuery.isOk():
return RestApiResponse.badRequest(histQuery.error)
if peerAddr.isNone() and not node.wakuLegacyStore.isNil(): if peerAddr.isNone() and not node.wakuLegacyStore.isNil():
## The user didn't specify a peer address and self-node is configured as a store node. ## The user didn't specify a peer address and self-node is configured as a store node.
## In this case we assume that the user is willing to retrieve the messages stored by ## In this case we assume that the user is willing to retrieve the messages stored by
## the local/self store node. ## the local/self store node.
return await node.retrieveMsgsFromSelfNode(histQuery.get()) return await node.retrieveMsgsFromSelfNode(histQuery)
# Parse the peer address parameter # Parse the peer address parameter
let parsedPeerAddr = parseUrlPeerAddr(peerAddr.toOpt()).valueOr: let parsedPeerAddr = parseUrlPeerAddr(peerAddr.toOpt()).valueOr:
@ -253,4 +248,4 @@ proc installStoreApiHandlers*(
"No suitable service peer & none discovered" "No suitable service peer & none discovered"
) )
return await node.performHistoryQuery(histQuery.value, peerAddr) return await node.performHistoryQuery(histQuery, peerAddr)

13
waku/waku_api/rest/legacy_store/types.nim
View File

@ -60,13 +60,11 @@ proc parseMsgDigest*(
return ok(none(waku_store_common.MessageDigest)) return ok(none(waku_store_common.MessageDigest))
let decodedUrl = decodeUrl(input.get()) let decodedUrl = decodeUrl(input.get())
let base64Decoded = base64.decode(Base64String(decodedUrl)) let base64DecodedArr = base64.decode(Base64String(decodedUrl)).valueOr:
return err(error)
var messageDigest = waku_store_common.MessageDigest() var messageDigest = waku_store_common.MessageDigest()
if not base64Decoded.isOk():
return err(base64Decoded.error)
let base64DecodedArr = base64Decoded.get()
# Next snippet inspired by "nwaku/waku/waku_archive/archive.nim" # Next snippet inspired by "nwaku/waku/waku_archive/archive.nim"
# TODO: Improve coherence of MessageDigest type # TODO: Improve coherence of MessageDigest type
messageDigest = block: messageDigest = block:
@ -220,10 +218,9 @@ proc readValue*(
of "data": of "data":
if data.isSome(): if data.isSome():
reader.raiseUnexpectedField("Multiple `data` fields found", "MessageDigest") reader.raiseUnexpectedField("Multiple `data` fields found", "MessageDigest")
let decoded = base64.decode(reader.readValue(Base64String)) let decoded = base64.decode(reader.readValue(Base64String)).valueOr:
if not decoded.isOk():
reader.raiseUnexpectedField("Failed decoding data", "MessageDigest") reader.raiseUnexpectedField("Failed decoding data", "MessageDigest")
data = some(decoded.get()) data = some(decoded)
else: else:
reader.raiseUnexpectedField("Unrecognided field", cstring(fieldName)) reader.raiseUnexpectedField("Unrecognided field", cstring(fieldName))

11
waku/waku_api/rest/server.nim
View File

@ -91,7 +91,7 @@ proc new*(
): Future[HttpResponseRef] {.async: (raises: [CancelledError]).} = ): Future[HttpResponseRef] {.async: (raises: [CancelledError]).} =
discard discard
let sres = HttpServerRef.new( server.httpServer = HttpServerRef.new(
address, address,
defaultProcessCallback, defaultProcessCallback,
serverFlags, serverFlags,
@ -106,12 +106,9 @@ proc new*(
maxRequestBodySize, maxRequestBodySize,
dualstack = dualstack, dualstack = dualstack,
middlewares = middlewares, middlewares = middlewares,
) ).valueOr:
if sres.isOk(): return err(error)
server.httpServer = sres.get() return ok(server)
ok(server)
else:
err(sres.error)
proc getRouter(): RestRouter = proc getRouter(): RestRouter =
# TODO: Review this `validate` method. Check in nim-presto what is this used for. # TODO: Review this `validate` method. Check in nim-presto what is this used for.

48
waku/waku_archive/driver/postgres_driver/postgres_driver.nim
View File

@ -1233,33 +1233,30 @@ proc refreshPartitionsInfo(
debug "refreshPartitionsInfo" debug "refreshPartitionsInfo"
self.partitionMngr.clearPartitionInfo() self.partitionMngr.clearPartitionInfo()
let partitionNamesRes = await self.getPartitionsList() let partitionNames = (await self.getPartitionsList()).valueOr:
if not partitionNamesRes.isOk(): return err("could not get partitions list: " & $error)
return err("Could not retrieve partitions list: " & $partitionNamesRes.error) for partitionName in partitionNames:
else: ## partitionName contains something like 'messages_1708449815_1708449875'
let partitionNames = partitionNamesRes.get() let bothTimes = partitionName.replace("messages_", "")
for partitionName in partitionNames: let times = bothTimes.split("_")
## partitionName contains something like 'messages_1708449815_1708449875' if times.len != 2:
let bothTimes = partitionName.replace("messages_", "") return err(fmt"loopPartitionFactory wrong partition name {partitionName}")
let times = bothTimes.split("_")
if times.len != 2:
return err(fmt"loopPartitionFactory wrong partition name {partitionName}")
var beginning: int64 var beginning: int64
try: try:
beginning = parseInt(times[0]) beginning = parseInt(times[0])
except ValueError: except ValueError:
return err("Could not parse beginning time: " & getCurrentExceptionMsg()) return err("Could not parse beginning time: " & getCurrentExceptionMsg())
var `end`: int64 var `end`: int64
try: try:
`end` = parseInt(times[1]) `end` = parseInt(times[1])
except ValueError: except ValueError:
return err("Could not parse end time: " & getCurrentExceptionMsg()) return err("Could not parse end time: " & getCurrentExceptionMsg())
self.partitionMngr.addPartitionInfo(partitionName, beginning, `end`) self.partitionMngr.addPartitionInfo(partitionName, beginning, `end`)
return ok() return ok()
const DefaultDatabasePartitionCheckTimeInterval = timer.minutes(10) const DefaultDatabasePartitionCheckTimeInterval = timer.minutes(10)
@ -1338,10 +1335,7 @@ proc removePartition(
let partitionName = partition.getName() let partitionName = partition.getName()
debug "beginning of removePartition", partitionName debug "beginning of removePartition", partitionName
var partSize = "" let partSize = (await self.getTableSize(partitionName)).valueOr("")
let partSizeRes = await self.getTableSize(partitionName)
if partSizeRes.isOk():
partSize = partSizeRes.get()
## Detach and remove the partition concurrently to not block the parent table (messages) ## Detach and remove the partition concurrently to not block the parent table (messages)
let detachPartitionQuery = let detachPartitionQuery =

3
waku/waku_archive/driver/postgres_driver/postgres_healthcheck.nim
View File

@ -24,8 +24,7 @@ proc checkConnectivity*(
var numTrial = 0 var numTrial = 0
while numTrial < MaxNumTrials: while numTrial < MaxNumTrials:
let res = await connPool.pgQuery(HealthCheckQuery) (await connPool.pgQuery(HealthCheckQuery)).isErrOr:
if res.isOk():
## Connection resumed. Let's go back to the normal healthcheck. ## Connection resumed. Let's go back to the normal healthcheck.
break errorBlock break errorBlock

3
waku/waku_archive_legacy/driver/postgres_driver/postgres_healthcheck.nim
View File

@ -24,8 +24,7 @@ proc checkConnectivity*(
var numTrial = 0 var numTrial = 0
while numTrial < MaxNumTrials: while numTrial < MaxNumTrials:
let res = await connPool.pgQuery(HealthCheckQuery) (await connPool.pgQuery(HealthCheckQuery)).isErrOr:
if res.isOk():
## Connection resumed. Let's go back to the normal healthcheck. ## Connection resumed. Let's go back to the normal healthcheck.
break errorBlock break errorBlock