Merge 68869a9 into ff3f8bb

apps/wakunode2/spammer.nim

@@ -0,0 +1,118 @@
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import
+  chronos,
+  chronicles,
+  stew/[byteutils, results],
+  std/times,
+  libp2p/protocols/pubsub/gossipsub,
+  strutils
+
+import
+  ../../waku/factory/waku,
+  ../../waku/factory/external_config,
+  ../../waku/waku_core,
+  ../../waku/waku_relay,
+  ../../waku/node/waku_node,
+  ../../waku/node/peer_manager/peer_manager,
+  ../../waku/waku_rln_relay/rln_relay,
+  ../../tests/waku_rln_relay/rln/waku_rln_relay_utils
+
+proc send(
+    waku: Waku, contentTopic: ContentTopic
+): Future[Result[void, string]] {.async.} =
+  var ephemeral = true
+
+  var message = WakuMessage(
+    payload: toBytes("Hello World!" & intToStr(int(getTime().toUnix()))),
+    contentTopic: contentTopic,
+    #      meta: metaBytes,
+    version: 2,
+    timestamp: getNanosecondTime(getTime().toUnixFloat()),
+    ephemeral: ephemeral,
+  )
+
+  let rlnRes =
+    waku.node.wakuRlnRelay.unsafeAppendRLNProof(message, float64(getTime().toUnix()))
+  if rlnRes.isOk:
+    let pubRes = await waku.node.publish(none(PubsubTopic), message)
+    if pubRes.isErr():
+      error "failed to publish", msg = pubRes.error
+      return err(pubRes.error)
+  else:
+    error "failed to append RLNProof", err = rlnRes.error
+    return err(rlnRes.error)
+
+  return ok()
+
+proc burstPublish(
+    waku: Waku, conf: WakuNodeConf, contentTopic: ContentTopic
+) {.async.} =
+  var futures: seq[Future[Result[void, string]]]
+  var i: uint64 = 0
+  var start = getTime().toUnixFloat()
+
+  while i < conf.rlnRelayUserMessageLimit:
+    futures.add(send(waku, contentTopic))
+    inc i
+
+  let results = await allFinished(futures)
+
+  var current = getTime().toUnixFloat()
+  var tillNextBurst =
+    int(int64(conf.rlnEpochSizeSec * 1000) - int64((current - start) * 1000))
+  info "Published messages",
+    sleep = tillNextBurst, msgCount = conf.rlnRelayUserMessageLimit
+
+  await sleepAsync(tillNextBurst)
+
+proc iterativePublish(
+    waku: Waku, conf: WakuNodeConf, contentTopic: ContentTopic
+) {.async.} =
+  var start = getTime().toUnixFloat()
+
+  (await send(waku, contentTopic)).isOkOr:
+    error "Failed to publish", err = error
+
+  #echo await (waku.node.isReady())
+  var current = getTime().toUnixFloat()
+  var tillNextMsg = int(int64(conf.spammerDelay) - int64((current - start) * 1000))
+  info "Published message", sleep = tillNextMsg
+
+  await sleepAsync(tillNextMsg)
+
+proc runSpammer*(
+    waku: Waku, conf: WakuNodeConf, contentTopic: ContentTopic = "/spammer/0/test/plain"
+) {.async.} =
+  if not conf.spammerEnable:
+    return
+
+  if not conf.rlnRelay:
+    error "RLN not configured!"
+    quit(QuitFailure)
+
+  while true:
+    var (inRelayPeers, outRelayPeers) =
+      waku.node.peerManager.connectedPeers(WakuRelayCodec)
+
+    var numPeers = len(inRelayPeers) + len(outRelayPeers)
+    if numPeers > 0:
+      break
+    info "Waiting for peers", numPeers = numPeers
+    await sleepAsync(1000)
+
+  #var rate = int(float(1000) / float(conf.msgRate))
+  #var delayBetweenMsg =
+  #  float(conf.rlnEpochSizeSec * 1000) /
+  #  (float(conf.rlnRelayUserMessageLimit) * conf.msgRateMultiplier)
+
+  info "Sending message with delay", delay = conf.spammerDelay
+
+  while true:
+    if conf.spammerBurst:
+      await burstPublish(waku, conf, contentTopic)
+    else:
+      await iterativePublish(waku, conf, contentTopic)

apps/wakunode2/wakunode2.nim

@@ -12,6 +12,7 @@ import
   system/ansi_c,
   libp2p/crypto/crypto
 import
+  ./spammer,
   ../../tools/rln_keystore_generator/rln_keystore_generator,
   ../../tools/rln_db_inspector/rln_db_inspector,
   ../../waku/common/logging,
@@ -52,6 +53,8 @@ when isMainModule:
   of inspectRlnDb:
     doInspectRlnDb(conf)
   of noCommand:
+    if conf.spammerEnable:
+      doRlnKeystoreGenerator(conf, false)
     # NOTE: {.threadvar.} is used to make the global variable GC safe for the closure uses it
     # It will always be called from main thread anyway.
     # Ref: https://nim-lang.org/docs/manual.html#threads-gc-safety
@@ -132,4 +135,6 @@ when isMainModule:
 
     info "Node setup complete"
 
+    asyncSpawn runSpammer(waku, conf)
+
     runForever()

tools/rln_keystore_generator/rln_keystore_generator.nim

@@ -15,7 +15,7 @@ import
 logScope:
   topics = "rln_keystore_generator"
 
-proc doRlnKeystoreGenerator*(conf: WakuNodeConf) =
+proc doRlnKeystoreGenerator*(conf: WakuNodeConf, quitOnSucces: bool = true) =
   # 1. load configuration
   trace "configuration", conf = $conf
 
@@ -42,7 +42,7 @@ proc doRlnKeystoreGenerator*(conf: WakuNodeConf) =
     idSecretHash = credential.idSecretHash.inHex(),
     idCommitment = credential.idCommitment.inHex()
 
-  if not conf.execute:
+  if quitOnSucces and not conf.execute:
     info "not executing, exiting"
     quit(0)
 
@@ -91,7 +91,6 @@ proc doRlnKeystoreGenerator*(conf: WakuNodeConf) =
     userMessageLimit: conf.rlnRelayUserMessageLimit,
   )
 
-
   let persistRes = addMembershipCredentials(
     conf.rlnRelayCredPath, keystoreCred, conf.rlnRelayCredPassword, RLNAppInfo
   )
@@ -106,4 +105,6 @@ proc doRlnKeystoreGenerator*(conf: WakuNodeConf) =
   except CatchableError:
     error "failure while stopping OnchainGroupManager", error = getCurrentExceptionMsg()
     quit(0) # 0 because we already registered on-chain
-  quit(0)
+
+  if quitOnSucces:
+    quit(0)

waku/factory/external_config.nim

@@ -595,6 +595,31 @@ type WakuNodeConf* = object
       name: "request-rate-period"
     .}: int64
 
+    spammerEnable* {.desc: "Enable spammer", defaultValue: false, name: "spammer".}:
+      bool
+    # msgRate* {.
+    #   desc: "Number of messages published per epoch",
+    #   defaultValue: 10,
+    #   name: "spammer-msg-rate"
+    # .}: int
+
+    # msgRateMultiplier* {.
+    #   desc: "Multiply max message per epoch to publish spam",
+    #   defaultValue: 1,
+    #   name: "spammer-msg-multiplier"
+    # .}: float
+    spammerDelay* {.
+      desc: "Delay between spawning a publish method (in miliseconds)",
+      defaultValue: 0,
+      name: "spammer-delay-between-msg"
+    .}: int
+
+    spammerBurst* {.
+      desc: "Send messages in burst instead of one by one",
+      defaultValue: false,
+      name: "spammer-burst"
+    .}: bool
+
 ## Parsing
 
 # NOTE: Keys are different in nim-libp2p