chore(rln-relay): add isReady check

tests/waku_rln_relay/test_rln_group_manager_onchain.nim

@@ -555,6 +555,36 @@ suite "Onchain group manager":
       manager.validRootBuffer.len() == 0
       manager.validRoots[credentialCount - 2] == expectedLastRoot
 
+  asyncTest "isReady should return false if ethRpc is none":
+    var manager = await setup()
+    await manager.init()
+
+    manager.ethRpc = none(Web3)
+
+    check:
+      (await manager.isReady()) == false
+
+  asyncTest "isReady should return false if lastSeenBlockHead > lastProcessed":
+    var manager = await setup()
+    await manager.init()
+
+    manager.lastSeenBlockHead = 10
+    manager.latestProcessedBlock = 5
+
+    check:
+      (await manager.isReady()) == false
+
+  asyncTest "isReady should return true if ethRpc is not syncing":
+    # not syncing implies the node is ready
+    var manager = await setup()
+    await manager.init()
+
+    manager.latestProcessedBlock = 10
+    manager.lastSeenBlockHead = 10
+
+    check:
+      (await manager.isReady()) == true
+
 
   ################################
   ## Terminating/removing Ganache

waku/node/waku_node.nim

@@ -732,8 +732,8 @@ proc lightpushPublish*(node: WakuNode, pubsubTopic: Option[PubsubTopic], message
 when defined(rln):
   proc mountRlnRelay*(node: WakuNode,
                       rlnConf: WakuRlnConfig,
-                      spamHandler: Option[SpamHandler] = none(SpamHandler),
-                      registrationHandler: Option[RegistrationHandler] = none(RegistrationHandler)) {.async.} =
+                      spamHandler = none(SpamHandler),
+                      registrationHandler = none(RegistrationHandler)) {.async.} =
     info "mounting rln relay"
 
     if node.wakuRelay.isNil():
@@ -903,3 +903,11 @@ proc stop*(node: WakuNode) {.async.} =
       await node.wakuRlnRelay.stop()
 
   node.started = false
+
+proc isReady*(node: WakuNode): Future[bool] {.async.} =
+  when defined(rln):
+    if node.wakuRlnRelay == nil:
+      return false
+    return await node.wakuRlnRelay.isReady()
+  ## TODO: add other protocol `isReady` checks
+  return true

waku/waku_rln_relay/group_manager/group_manager_base.nim

@@ -162,3 +162,6 @@ method generateProof*(g: GroupManager,
   if proofGenRes.isErr():
     return err("proof generation failed: " & $proofGenRes.error())
   return ok(proofGenRes.value())
+
+method isReady*(g: GroupManager): Future[bool] {.base,gcsafe.} =
+  raise newException(CatchableError, "isReady proc for " & $g.type & " is not implemented yet")

waku/waku_rln_relay/group_manager/on_chain/group_manager.nim

@@ -73,6 +73,8 @@ type
     # in event of a reorg. we store 5 in the buffer. Maybe need to revisit this,
     # because the average reorg depth is 1 to 2 blocks.
     validRootBuffer*: Deque[MerkleNode]
+    # this variable tracks the last seen head
+    lastSeenBlockHead*: BlockNumber
 
 const DefaultKeyStorePath* = "rlnKeystore.json"
 const DefaultKeyStorePassword* = "password"
@@ -313,13 +315,18 @@ proc getAndHandleEvents(g: OnchainGroupManager,
                         fromBlock: BlockNumber,
                         toBlock: Option[BlockNumber] = none(BlockNumber)): Future[void] {.async.} =
   initializedGuard(g)
+  proc getLatestBlockNumber(): BlockNumber =
+    if toBlock.isSome(): 
+      return toBlock.get()
+    return fromBlock
 
   let blockTable = await g.getBlockTable(fromBlock, toBlock)
+  if blockTable.len() > 0:
+    g.lastSeenBlockHead = getLatestBlockNumber()
   await g.handleEvents(blockTable)
   await g.handleRemovedEvents(blockTable)
 
-  g.latestProcessedBlock = if toBlock.isSome(): toBlock.get()
-                             else: fromBlock
+  g.latestProcessedBlock = getLatestBlockNumber()
   let metadataSetRes = g.setMetadata()
   if metadataSetRes.isErr():
     # this is not a fatal error, hence we don't raise an exception
@@ -528,3 +535,27 @@ method stop*(g: OnchainGroupManager): Future[void] {.async.} =
     error "failed to flush to the tree db"
 
   g.initialized = false
+
+proc isSyncing*(g: OnchainGroupManager): Future[bool] {.async,gcsafe.} =
+  let ethRpc = g.ethRpc.get()
+
+  try:
+    let syncing = await ethRpc.provider.eth_syncing()
+    return syncing.getBool()
+  except CatchableError:
+    error "failed to get the syncing status", error = getCurrentExceptionMsg()
+    return false
+
+method isReady*(g: OnchainGroupManager): Future[bool] {.async,gcsafe.} =
+  initializedGuard(g)
+
+  if g.ethRpc.isNone():
+    return false
+
+  if g.lastSeenBlockHead == 0:
+    return false
+
+  if g.latestProcessedBlock < g.lastSeenBlockHead:
+    return false
+
+  return not (await g.isSyncing())

waku/waku_rln_relay/group_manager/static/group_manager.nim

@@ -109,3 +109,9 @@ method stop*(g: StaticGroupManager): Future[void] =
   var retFut = newFuture[void]("StaticGroupManager.stop")
   retFut.complete()
   return retFut
+
+method isReady*(g: StaticGroupManager): Future[bool] {.gcsafe.} =
+  initializedGuard(g)
+  var retFut = newFuture[bool]("StaticGroupManager.isReady")
+  retFut.complete(true)
+  return retFut

waku/waku_rln_relay/rln_relay.nim

@@ -400,6 +400,18 @@ proc mount(conf: WakuRlnConfig,
   return WakuRLNRelay(groupManager: groupManager,
                       messageBucket: messageBucket)
 
+proc isReady*(rlnPeer: WakuRLNRelay): Future[bool] {.async.} =
+  ## returns true if the rln-relay protocol is ready to relay messages
+  ## returns false otherwise
+
+  # could be nil during startup
+  if rlnPeer.groupManager == nil:
+    return false
+  try:
+    return await rlnPeer.groupManager.isReady()
+  except CatchableError:
+    error "could not check if the rln-relay protocol is ready", err = getCurrentExceptionMsg()
+    return false
 
 proc new*(T: type WakuRlnRelay,
           conf: WakuRlnConfig,