logos-delivery/apps/benchmarks/message_path_bench.nim

## Message-path performance benchmark (Phase 1 of the async-copy refactor).
##
## Validates the decode/hash-per-message claims in
## `docs/analysis/async_copy_analysis.md` on the *current* code, before any
## Phase 2/3 change, and produces reproducible before/after numbers.
##
## Two scenarios:
##   * micro  - in-process, no network. Replays the WakuRelay inbound sequence
##              (ordered validator decode + topicHandler decode + full
##              uniqueTopicHandler dispatch chain: trace/filter/archive/sync/
##              internal/legacyApp) by calling the relay's own registered
##              artifacts. NOTE: the onRecv/onValidated/onSend gossipsub
##              observers are NOT exercised here (they fire only in real libp2p
##              dispatch) - the macro scenario covers those.
##   * macro  - two in-process nodes over loopback gossipsub. Node A publishes,
##              node B (archive + sync mounted) receives. Captures the libp2p
##              observer decodes too, so counters here show the full inbound
##              decode/hash budget.
##
## Requires `-d:msgPathCounters` (the nimble task `benchMessagePath` sets it).
## Build+run:  nimble benchMessagePath
##
## Determinism: fixed-seed workload (seed 42), payload mix 10/50/150 kB at
## 25/50/25 %, unique per-message payload prefix + timestamp (no dedup).

{.push raises: [].}

import
  std/[options, algorithm, random, monotimes, times, strformat, strutils],
  results,
  chronos,
  libp2p/protocols/pubsub/rpc/messages,
  libp2p/crypto/crypto as libp2p_keys

import
  logos_delivery/waku/[
    waku_core,
    waku_node,
    node/waku_node/relay,
    node/waku_node/store,
    waku_relay,
    waku_relay/protocol,
  ]
import brokers/broker_context
import tests/testlib/[wakucore, wakunode]
import tests/waku_archive/archive_utils

# ---------------------------------------------------------------------------
# Workload
# ---------------------------------------------------------------------------

const
  WarmupCount = 100
  MeasuredCount = 1000
  Kb = 1000
  PayloadSmall = 10 * Kb
  PayloadMedium = 50 * Kb
  PayloadLarge = 150 * Kb
  PayloadProfile = "10/50/150kB@25/50/25"

type Workload = object
  msgs: seq[WakuMessage]
  encoded: seq[seq[byte]]

proc buildSizeMix(total: int): seq[int] =
  ## Deterministic 25/50/25 size mix, shuffled with a fixed seed so ordering is
  ## stable across runs but not pathologically grouped.
  result = newSeqOfCap[int](total)
  let nSmall = total div 4
  let nLarge = total div 4
  let nMedium = total - nSmall - nLarge
  for _ in 0 ..< nSmall:
    result.add(PayloadSmall)
  for _ in 0 ..< nMedium:
    result.add(PayloadMedium)
  for _ in 0 ..< nLarge:
    result.add(PayloadLarge)
  var r = initRand(42)
  shuffle(r, result)

proc buildWorkload(shard: PubsubTopic, count: int): Workload =
  ## Generate `count` deterministic messages + their pre-encoded proto buffers.
  ## First 8 payload bytes encode the index (uniqueness -> distinct hash -> no
  ## gossipsub dedup). Timestamps are unique and near `now()` so the archive
  ## timestamp validator accepts them.
  let sizes = buildSizeMix(count)
  let baseTs = getNowInNanosecondTime()
  result.msgs = newSeqOfCap[WakuMessage](count)
  result.encoded = newSeqOfCap[seq[byte]](count)
  for i in 0 ..< count:
    let size = sizes[i]
    var payload = newSeq[byte](size)
    # unique 8-byte index prefix
    for b in 0 ..< 8:
      payload[b] = byte((i shr (b * 8)) and 0xFF)
    # deterministic filler (content does not affect decode/hash cost, size does)
    for j in 8 ..< size:
      payload[j] = byte((j * 2654435761 + i) and 0xFF)
    let msg = WakuMessage(
      payload: payload,
      contentTopic: DefaultContentTopic,
      meta: @[],
      version: 2,
      timestamp: Timestamp(int64(baseTs) + int64(i)),
      ephemeral: false,
      proof: @[],
    )
    result.msgs.add(msg)
    result.encoded.add(msg.encode().buffer)

# ---------------------------------------------------------------------------
# Metrics helpers
# ---------------------------------------------------------------------------

type ScenarioResult = object
  scenario: string
  msgs: int
  wallNs: int64
  p50Ns: int64
  p99Ns: int64
  decodeCalls: int64
  hashCalls: int64
  decodedBytes: int64
  hashedBytes: int64
  occupiedDeltaBytes: int64
  gcStats: string

proc percentiles(durs: var seq[int64]): (int64, int64) =
  if durs.len == 0:
    return (0'i64, 0'i64)
  sort(durs)
  let p50 = durs[durs.len div 2]
  let p99 = durs[min(durs.len - 1, (durs.len * 99) div 100)]
  (p50, p99)

proc parseGcCollections(stats: string): int64 =
  ## Best-effort extraction of a collections count from GC_getStatistics().
  for line in stats.splitLines():
    let low = line.toLowerAscii()
    if "collections" in low:
      for tok in low.split({' ', ':', '\t'}):
        try:
          if tok.len > 0 and tok[0] in {'0' .. '9'}:
            return parseBiggestInt(tok)
        except ValueError:
          discard
  return 0

const CsvHeader =
  "scenario,msgs,payload_profile,wall_ms,msg_per_s,ns_per_msg_p50,ns_per_msg_p99," &
  "decodes_per_msg,hashes_per_msg,hashed_MB,decoded_MB,occupied_mem_delta_MB,gc_collections"

proc emit(res: ScenarioResult) =
  let wallMs = res.wallNs.float / 1_000_000.0
  let msgPerS =
    if res.wallNs > 0:
      res.msgs.float * 1_000_000_000.0 / res.wallNs.float
    else:
      0.0
  let msgsDiv = max(res.msgs, 1).float
  let decodesPerMsg = res.decodeCalls.float / msgsDiv
  let hashesPerMsg = res.hashCalls.float / msgsDiv
  let hashedMB = res.hashedBytes.float / 1_000_000.0
  let decodedMB = res.decodedBytes.float / 1_000_000.0
  let memDeltaMB = res.occupiedDeltaBytes.float / 1_000_000.0
  let gcColl = parseGcCollections(res.gcStats)

  echo "----------------------------------------------------------------"
  echo "scenario           : ", res.scenario
  echo "messages           : ", res.msgs
  echo "payload_profile    : ", PayloadProfile
  echo "wall               : ", fmt"{wallMs:.3f} ms"
  echo "throughput         : ", fmt"{msgPerS:.1f} msg/s"
  echo "ns/msg p50 / p99   : ", res.p50Ns, " / ", res.p99Ns
  echo "decodes_per_msg    : ",
    fmt"{decodesPerMsg:.3f}", "  (", res.decodeCalls, " total)"
  echo "hashes_per_msg     : ", fmt"{hashesPerMsg:.3f}", "  (", res.hashCalls, " total)"
  echo "decoded_MB         : ", fmt"{decodedMB:.2f}"
  echo "hashed_MB          : ", fmt"{hashedMB:.2f}"
  echo "occupied_mem_delta : ", fmt"{memDeltaMB:.2f} MB"
  echo "GC_getStatistics   :"
  echo res.gcStats
  echo "CSV,",
    res.scenario,
    ",",
    res.msgs,
    ",",
    PayloadProfile,
    ",",
    fmt"{wallMs:.3f}",
    ",",
    fmt"{msgPerS:.1f}",
    ",",
    res.p50Ns,
    ",",
    res.p99Ns,
    ",",
    fmt"{decodesPerMsg:.3f}",
    ",",
    fmt"{hashesPerMsg:.3f}",
    ",",
    fmt"{hashedMB:.2f}",
    ",",
    fmt"{decodedMB:.2f}",
    ",",
    fmt"{memDeltaMB:.2f}",
    ",",
    gcColl

# ---------------------------------------------------------------------------
# Node setup
# ---------------------------------------------------------------------------

proc dummyHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
  discard

proc buildIngestNode(
    shard: PubsubTopic, appHandler: WakuRelayHandler
): Future[WakuNode] {.async.} =
  ## A single node: relay + archive (in-memory sqlite) + store-sync, subscribed
  ## to `shard`. Built under a fresh broker context (mirrors the test helpers).
  ## `appHandler` is installed as the (only) relay app handler for the shard;
  ## registerRelayHandler ignores later handlers on an already-subscribed shard,
  ## so the receiving node must be built with its counting handler up front.
  var node: WakuNode
  lockNewGlobalBrokerContext:
    node = newTestWakuNode(generateSecp256k1Key())
    node.mountMetadata(DefaultClusterId, @[DefaultShardId]).isOkOr:
      raiseAssert "mountMetadata failed: " & error
    (await node.mountRelay()).isOkOr:
      raiseAssert "mountRelay failed: " & error
    node.mountArchive(newSqliteArchiveDriver()).isOkOr:
      raiseAssert "mountArchive failed: " & error
    (await node.mountStoreSync(DefaultClusterId, @[DefaultShardId], @[], 3600, 300, 20)).isOkOr:
      raiseAssert "mountStoreSync failed: " & error
    await node.start()
  node.subscribe((kind: PubsubSub, topic: shard), appHandler).isOkOr:
    raiseAssert "subscribe failed: " & error
  return node

# ---------------------------------------------------------------------------
# Micro scenario
# ---------------------------------------------------------------------------

proc runMicro(
    node: WakuNode, shard: PubsubTopic, work: Workload, label: string
): Future[ScenarioResult] {.async.} =
  let relay = node.wakuRelay
  let validator = relay.benchTopicValidator(shard)
  let handler = relay.benchTopicHandler(shard)
  doAssert not validator.isNil(), "no validator installed for shard"
  doAssert not handler.isNil(), "no topicHandler installed for shard"

  # Warmup (excluded from timing and counters)
  for i in 0 ..< min(WarmupCount, work.encoded.len):
    discard await validator(shard, Message(topic: shard, data: work.encoded[i]))
    await handler(shard, work.encoded[i])

  GC_fullCollect()
  resetMsgPathCounters()
  let occBefore = getOccupiedMem()
  var durs = newSeqOfCap[int64](work.encoded.len)

  let wallStart = getMonoTime()
  for i in 0 ..< work.encoded.len:
    let t0 = getMonoTime()
    discard await validator(shard, Message(topic: shard, data: work.encoded[i]))
    await handler(shard, work.encoded[i])
    durs.add((getMonoTime() - t0).inNanoseconds)
  let wallNs = (getMonoTime() - wallStart).inNanoseconds

  let occAfter = getOccupiedMem()
  let (p50, p99) = percentiles(durs)

  return ScenarioResult(
    scenario: label,
    msgs: work.encoded.len,
    wallNs: wallNs,
    p50Ns: p50,
    p99Ns: p99,
    decodeCalls: msgPathCounters.decodeCalls,
    hashCalls: msgPathCounters.hashCalls,
    decodedBytes: msgPathCounters.decodedBytes,
    hashedBytes: msgPathCounters.hashedBytes,
    occupiedDeltaBytes: int64(occAfter) - int64(occBefore),
    gcStats: GC_getStatistics(),
  )

# ---------------------------------------------------------------------------
# Macro scenario
# ---------------------------------------------------------------------------

type Arrivals = ref object
  count: int
  target: int
  done: AsyncEvent
  times: seq[MonoTime]

proc runMacro(shard: PubsubTopic, work: Workload): Future[ScenarioResult] {.async.} =
  let arrivals = Arrivals(
    count: 0,
    target: work.msgs.len,
    done: newAsyncEvent(),
    times: newSeqOfCap[MonoTime](work.msgs.len),
  )

  proc countingHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
    arrivals.times.add(getMonoTime())
    arrivals.count += 1
    if arrivals.count >= arrivals.target:
      arrivals.done.fire()

  # Node B: receiver with archive + sync, counting handler installed up front.
  let nodeB = await buildIngestNode(shard, countingHandler)

  # Node A: publisher (relay only).
  var nodeA: WakuNode
  lockNewGlobalBrokerContext:
    nodeA = newTestWakuNode(generateSecp256k1Key())
    nodeA.mountMetadata(DefaultClusterId, @[DefaultShardId]).isOkOr:
      raiseAssert "nodeA mountMetadata failed: " & error
    (await nodeA.mountRelay()).isOkOr:
      raiseAssert "nodeA mountRelay failed: " & error
    await nodeA.start()
  nodeA.subscribe((kind: PubsubSub, topic: shard), dummyHandler).isOkOr:
    raiseAssert "nodeA subscribe failed: " & error

  await nodeA.connectToNodes(@[nodeB.peerInfo.toRemotePeerInfo()])

  # Wait for the relay mesh to form.
  var meshFormed = false
  for _ in 0 ..< 100:
    if nodeA.wakuRelay.getNumPeersInMesh(shard).valueOr(0) > 0:
      meshFormed = true
      break
    await sleepAsync(chronos.milliseconds(100))
  doAssert meshFormed, "nodeA<->nodeB relay mesh did not form"

  # Warmup publishes (separate messages so they don't count against the batch).
  let warmup = buildWorkload(shard, min(WarmupCount, 50))
  for i in 0 ..< warmup.msgs.len:
    discard await nodeA.publish(some(shard), warmup.msgs[i])
  # let warmup drain
  await sleepAsync(chronos.milliseconds(500))

  GC_fullCollect()
  resetMsgPathCounters()
  arrivals.count = 0
  arrivals.times.setLen(0)
  let occBefore = getOccupiedMem()

  var pubOk = 0
  var pubErr = 0
  let wallStart = getMonoTime()
  for i in 0 ..< work.msgs.len:
    let r = await nodeA.publish(some(shard), work.msgs[i])
    if r.isOk():
      pubOk += 1
    else:
      pubErr += 1
    # Flow control: pace the publisher to node B's consumption so the single
    # loopback gossipsub send queue never overflows (bursts drop large msgs).
    # The guard caps the wait so a dropped message can't stall the loop forever.
    var guard = 0
    while (i + 1) - arrivals.count > 32 and guard < 1000:
      await sleepAsync(chronos.milliseconds(1))
      inc guard

  # Wait for all arrivals on node B (bounded).
  let allReceived = await arrivals.done.wait().withTimeout(chronos.seconds(30))
  let wallNs = (getMonoTime() - wallStart).inNanoseconds
  echo "macro publish: ok=", pubOk, " err=", pubErr, " received=", arrivals.count
  let occAfter = getOccupiedMem()

  if not allReceived:
    echo "WARNING macro: only ",
      arrivals.count, "/", arrivals.target, " messages received before timeout"

  # inter-arrival deltas as ns/msg proxy
  var durs = newSeqOfCap[int64](arrivals.times.len)
  for i in 1 ..< arrivals.times.len:
    durs.add((arrivals.times[i] - arrivals.times[i - 1]).inNanoseconds)
  let (p50, p99) = percentiles(durs)

  let received = arrivals.count
  let res = ScenarioResult(
    scenario: "macro",
    msgs: received,
    wallNs: wallNs,
    p50Ns: p50,
    p99Ns: p99,
    decodeCalls: msgPathCounters.decodeCalls,
    hashCalls: msgPathCounters.hashCalls,
    decodedBytes: msgPathCounters.decodedBytes,
    hashedBytes: msgPathCounters.hashedBytes,
    occupiedDeltaBytes: int64(occAfter) - int64(occBefore),
    gcStats: GC_getStatistics(),
  )

  await nodeA.stop()
  await nodeB.stop()
  return res

# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

proc main() {.async.} =
  when not defined(msgPathCounters):
    echo "ERROR: build with -d:msgPathCounters (use `nimble benchMessagePath`)"
    return

  let shard = DefaultPubsubTopic
  echo "=== message_path_bench ==="
  echo CsvHeader

  # Micro: run twice for the determinism check.
  let microNode = await buildIngestNode(shard, dummyHandler)
  let work = buildWorkload(shard, MeasuredCount)

  let micro1 = await runMicro(microNode, shard, work, "micro_run1")
  emit(micro1)
  let micro2 = await runMicro(microNode, shard, work, "micro_run2")
  emit(micro2)

  let s1 = micro1.msgs.float * 1e9 / micro1.wallNs.float
  let s2 = micro2.msgs.float * 1e9 / micro2.wallNs.float
  let variancePct = abs(s1 - s2) / max(s1, s2) * 100.0
  echo "----------------------------------------------------------------"
  echo "micro determinism  : run1=",
    fmt"{s1:.1f}",
    " msg/s, run2=",
    fmt"{s2:.1f}",
    " msg/s, variance=",
    fmt"{variancePct:.2f}",
    "% (target <5%)"

  await microNode.stop()

  # Macro.
  let macroRes = await runMacro(shard, work)
  emit(macroRes)

  echo "=== done ==="

when isMainModule:
  waitFor main()