From 05ac82d4540b73e0576cc917526f842ebf4f04da Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:07:39 +0200 Subject: [PATCH 1/4] docs: async copy analysis and 3-phase elimination plan Analysis of chronos/refc deep-copy behavior on the WakuMessage path (~19+F payload copies, 4-6 SHA-256 passes per inbound relayed message) plus executor plans for: perf harness, WakuMessage ref object, WakuEnvelope. Co-Authored-By: Claude Fable 5 --- docs/analysis/async_copy_analysis.md | 229 +++++++++++++++++++ docs/analysis/async_copy_fix_plan.md | 180 +++++++++++++++ docs/analysis/plan_phase1_bench.md | 125 ++++++++++ docs/analysis/plan_phase2_wakumessage_ref.md | 127 ++++++++++ docs/analysis/plan_phase3_wakuenvelope.md | 165 +++++++++++++ 5 files changed, 826 insertions(+) create mode 100644 docs/analysis/async_copy_analysis.md create mode 100644 docs/analysis/async_copy_fix_plan.md create mode 100644 docs/analysis/plan_phase1_bench.md create mode 100644 docs/analysis/plan_phase2_wakumessage_ref.md create mode 100644 docs/analysis/plan_phase3_wakuenvelope.md diff --git a/docs/analysis/async_copy_analysis.md b/docs/analysis/async_copy_analysis.md new file mode 100644 index 000000000..50cf6fd8f --- /dev/null +++ b/docs/analysis/async_copy_analysis.md @@ -0,0 +1,229 @@ +# Async deep-copy analysis — chronos, Option/Opt/Result, WakuMessage path + +Environment: Nim 2.2.4, `--mm:refc`, chronos 4.2.2 (`nimbledeps/pkgs2/chronos-4.2.2-*`), +nim-results 0.5.1, nim-libp2p 2.0.0. All findings verified against these sources. + +Key refc facts driving everything below: +- Assignment of `string`/`seq`/object-with-seqs = deep copy (`copyString`/`genericSeqAssign`). No `=sink` hooks. +- `system.move()` degrades to *copy + reset* under refc (move semantics require ARC/ORC). +- `lent` returns work under refc (true borrow, no copy at return) — the copy materializes when the borrow is bound to a `let`/`var` or passed by value. + +--- + +## Table 1 — Where chronos generates deep copies + +| # | Copy point | Mechanism (evidence) | refc | ORC | +|---|---|---|---|---| +| 1 | **Every `{.async.}` call: each seq/string/object param** | async transform moves body into a closure iterator (`chronos/internal/asyncmacro.nim:445-517`); params are lambda-lifted into the env object → `env.param = param` = `genericSeqAssign` | **1 deep copy per param per call** | move/cursor — free | +| 2 | **`complete(future, val)`** | `val: T` is **not `sink`** (`asyncfutures.nim:198-202`); `future.internalValue = val`. Macro calls `complete(fut, move(result))` (`asyncmacro.nim:160`) but `move` is a copy under refc | **1–2 deep copies per completion** | 1 copy (non-sink param survives ORC) | +| 3 | **Consuming the result: `let x = await f(...)`** | `Future.value()` returns `lent T` (`futures.nim:231-241`) — free at return, but binding the borrow to `x` copies | **1 deep copy per await** | 1 copy (can't move out of a borrow) | +| 4 | Callbacks / `futureContinue` | future passed by ref, callbacks `move`d (`asyncfutures.nim:187`) | none | none | +| 5 | `{.async: (raw: true).}` procs | no closure-iterator transform → no param capture | none | none | + +Net: `let x = await f(a, b)` where `a`, `b`, and the result are seq-heavy costs +**~2 copies of each argument-side seq + 2–3 copies of the result** under refc. +Rows 2–3 are chronos API design (non-`sink` complete, `lent` read) and persist under ORC; +row 1 is refc-specific. + +libp2p is aware of row 1: `pubsub.nim:415-427` uses `{.async: (raw: true).}` for `handleData` +"without copying data into closure" — but every downstream `TopicHandler` is a normal +`{.async.}` taking `(topic: string, data: seq[byte])` by value, so the copy fires there, +once per registered handler. + +--- + +## Table 2 — Option[T], Opt[T], Result[T, E] (results 0.5.1, `resultsLent=true` on Nim 2.2.4) + +| Accessor | Kind / returns | Copies under refc? | Evidence (results.nim) | +|---|---|---|---| +| `ok(x)` / `err(x)` | template, field-construct | copy of `x` into Result (move only if rvalue construction elides) | :451-454, :471-474 | +| `value()`, `get()`, `tryGet()`, `expect()` | `lent T` (or template → lent) | none at return; **deep copy when bound to `let`** | :870-879, :1072, :1078, :952 | +| `value()`/`get()` on `var` Result | `var T` borrow | none | :881-889, :1092 | +| `unsafeGet()`/`unsafeValue()` | template, in-place field access | none in place; copy on bind | :902-905, :1084 | +| `get(otherwise)` | **func returning `T` by value** | **always copies** (plus `otherwise` passed by value) | :1095 | +| `valueOr:` | template: `let s = (self)` then yields field | copies whole Result if `self` is lvalue (rvalue temp is sunk); **value deep-copied on the outer `let` bind** | :1396-1432 | +| `isOkOr:` | template: `let s = (self)` | copies Result (upstream `# TODO avoid copy`) | ~:1355 | +| `?` operator | template: `let v = (self)` | copies Result; value copied on bind | :1533-1544 | +| `std/options some(sink T)` | proc, sink param | move if rvalue, copy if lvalue | options.nim:125 | +| `std/options get()` | `lent T` | none at return; copy on bind | options.nim:191 | +| `std/options get(otherwise)` | proc returning `T` | **always copies** | options.nim:206 | + +Rule of thumb: the wrappers themselves are fine; the cost is **one full deep copy each time a +large value is bound out** (`let msg = decode(...).valueOr: ...` = copy), and `valueOr`/`isOkOr`/`?` +additionally copy the whole Result when applied to an lvalue. + +--- + +## WakuMessage path — hop-by-hop copy count + +`WakuMessage` ([message.nim:12-29](logos_delivery/waku/waku_core/message/message.nim)) is a plain +object with three heap seqs (`payload`, `meta`, `proof`) plus `contentTopic: string` — +every deep copy touches ~4 heap allocations. + +### A. Inbound relay — the same proto bytes are decoded ≥4× + +| Hop | Site | Operation | Cost | +|---|---|---|---| +| A1 | [protocol.nim:543](logos_delivery/waku/waku_relay/protocol.nim:543) | `WakuMessage.decode(message.data)` in ordered validator | decode alloc + `valueOr` bind copy | +| A2 | [protocol.nim:271](logos_delivery/waku/waku_relay/protocol.nim:271) | decode again in `onRecv` observer | same | +| A3 | [protocol.nim:326](logos_delivery/waku/waku_relay/protocol.nim:326) | decode again in `onValidated` observer | same | +| A5 | [protocol.nim:603](logos_delivery/waku/waku_relay/protocol.nim:603) | decode again in subscribe topicHandler | same | +| (out) | [protocol.nim:341](logos_delivery/waku/waku_relay/protocol.nim:341) | decode in `onSend` observer (outbound leg) | same | + +Each decode = fresh payload/meta/proof seqs (`codec.nim:23`) + `ok()` wrap + `valueOr` bind +≈ **2 avoidable full-message copies per decode site** on top of the one unavoidable materialization. + +### B. Node dispatch — 6+ async closure captures of the decoded message + +`registerRelayHandler` ([subscription_manager.nim:29](logos_delivery/waku/node/subscription_manager.nim)) +builds `uniqueTopicHandler` (:72, `{.async.}`, msg by value → **env capture copy #1**), which +sequentially awaits trace(:45), filter(:51), archive(:57), sync(:63), internal(:69), +legacyApp(:82) handlers — **each a `{.async.}` proc taking `msg: WakuMessage` by value → +6 more env-capture deep copies**, plus `MessageSeenEvent.emit(...msg)` (:70) broker copy. + +### C–F. Downstream + +| Path | Sites | Copies | +|---|---|---| +| Archive | [archive.nim:98,101](logos_delivery/waku/waku_archive/archive.nim) | async capture + `computeMessageHash` (re-hash) + field copies into SQLite params; **queue_driver retains a full WakuMessage copy** in `SortedSet[Index, WakuMessage]` | +| Filter push | [protocol.nim:243-264,213,216,170](logos_delivery/waku/waku_filter_v2/protocol.nim) | async capture + re-hash ×2 + `MessagePush(wakuMessage: message)` wrap + one encode (good: **not** per-peer) + **per-peer buffer env-capture copy** in `pushToPeer` | +| Lightpush inbound | [protocol.nim:84,67](logos_delivery/waku/waku_lightpush/protocol.nim), callbacks.nim:16-24 | decode + hash + validateMessage re-encodes + re-hashes + relay publish re-encodes | +| Outbound publish | [relay/protocol.nim:674-690](logos_delivery/waku/waku_relay/protocol.nim:674) | explicit `var message = wakuMessage` copy (:680, for timestamp mutation) + encode + hash | +| FFI/JSON boundary | library/json_message_event.nim:72-95, relay_api.nim:115-121 | explicit `copyMem` of payload/meta/proof + base64 encode/decode string copies (inherent to the JSON API) | + +### G. computeMessageHash — recomputed 4–6× per message + +Call sites for the *same* inbound message: relay `:217`/validator, archive `:101`, +filter `:246` + `:195`, store-sync reconciliation `:78` — plus lightpush/API paths. +The hash itself doesn't copy (params feed `openArray` into sha256), but it's ~4–6 redundant +SHA-256 passes over the full payload per message. + +### Copy budget (one inbound relayed message, archive + filter + sync active, F filter peers) + +| Category | Full-payload copies | +|---|---| +| Redundant proto decodes (A) | ~8 (4 sites × ~2) | +| Dispatch env captures (B) | 7 | +| Archive + filter + push wrap/encode | ~4 + F (per filter peer) | +| **Total** | **~19 + F**, vs. theoretical minimum ≈ 3 (decode once, hash once, encode once) | + +Plus 4–6 redundant SHA-256 passes. For a 1 MB payload that is ~20 MB of allocation + +memcpy + refc GC pressure per message. + +--- + +## Recommendations (prioritized) + +**Yes, this is worth fixing** — the relay hot path does ~6× more copying than necessary, +and it's all in per-message code. + +1. **P1 — Decode once in WakuRelay** (`waku_relay/protocol.nim`). The validator (A1), + both observers (A2/A3), and the topicHandler (A5) each decode independently. Decode in + the ordered validator, cache `(msgId → decoded msg + hash)` in a small TimedCache (the + gossipsub validator→handler API gives no side channel), or restructure so observers + receive the already-decoded message. Removes ~6 full copies + 2–3 hash passes per message. + No public API change. + +2. **P2 — Pass a `ref` envelope through internal dispatch.** Introduce e.g. + `MessageEnvelope = ref object` holding `msg: WakuMessage`, `hash: WakuMessageHash`, + `pubsubTopic`, created once after validation; change `WakuRelayHandler`, + `subscription_manager` handlers, `archive.handleMessage`, `filter.handleMessage`, and + sync ingress to take the ref. Under refc a ref copy is pointer + refcount — kills all + 7 env-capture copies (Table 1 row 1) and all downstream re-hashing (compute once, carry + it). This is the single highest-leverage internal refactor. (Immutable-by-convention; + document that handlers must not mutate.) + +3. **P3 — Filter `pushToPeer`: share the encoded buffer.** The buffer is already encoded + once, but the `{.async.}` by-value param copies it per peer + ([protocol.nim:170,216](logos_delivery/waku/waku_filter_v2/protocol.nim:170)). Pass a + `ref seq[byte]` (or make `pushToPeer` `{.async: (raw: true).}`). + +4. **P4 — Don't return large values from async procs.** Each such return costs 2–3 copies + (Table 1 rows 2–3) that even ORC won't remove. Return `ref T`, or fill a caller-provided + buffer. Applies to store query results (`seq[WakuMessage]`) and codec-heavy client calls. + +5. **P5 — Result/Option idioms for large T.** Avoid `get(otherwise)` (always copies both + sides); prefer in-place access (`.value.field`, `var`-borrow `get`) where the value + isn't stored anyway. Low priority once P2 makes messages refs — Opt/Result over `ref` or + small types is free. + +6. **P6 — Upstream / longer term.** + - chronos: make `complete()` take `sink T` (removes Table 1 row 2 under ORC; today it + copies under *every* mm). + - ORC migration removes row 1 (env captures become moves) — the biggest class of copies — + but rows 2–3 remain, so P2/P4 stay valuable regardless. + - The explicit `var message = wakuMessage` copy at publish + ([protocol.nim:680](logos_delivery/waku/waku_relay/protocol.nim:680)) exists only to + patch the timestamp — set the timestamp before entering publish and take the param + immutably. + +## Gains and effort estimate + +Baseline per inbound relayed message (archive + filter + sync active, F filter peers): +**~19 + F full-payload copies, 4–6 SHA-256 passes.** + +| Fix | Copies removed | Hash passes removed | Remaining after fix | Effort (dev-days) | Risk / blast radius | +|---|---|---|---|---|---| +| P1 decode-once in relay | ~6 (3 redundant decode sites × 2) | 2–3 | ~13 + F | 2–3 (incl. tests) | Contained to `waku_relay/protocol.nim`; needs msgId→decoded cache or observer restructure | +| P2 ref envelope through dispatch | ~9 (7 env captures + archive/filter captures) | 1–2 (hash carried) | ~4 + F | 3–5 | Internal API change: `WakuRelayHandler`, subscription_manager, archive, filter, sync + their tests | +| P3 shared filter buffer | F (per-peer buffer capture) | — | ~4 | 0.5 | One proc signature in `waku_filter_v2` | +| P4 no large async returns (store/query hot paths) | 2–3 per large-value await | — | — | 2–3 (hot paths only) | Store/query client APIs | +| P5 Result/Option idioms | marginal once P2 lands | — | — | opportunistic | none | +| P6 chronos `sink complete` / ORC | env-capture class (row 1) | — | — | weeks (upstream cycle) / months (ORC) | out of repo control | + +**After P1+P2+P3: ~4 full-payload copies and 1–2 hash passes per message — a ~75–80 % +reduction in per-message memcpy/alloc volume and ~70 % less hashing CPU.** The remaining ~4 +are near the floor: proto→WakuMessage materialization, envelope fill, outbound encode, DB row. + +Concrete scale for the target payload profile (min 10 kB / avg 50 kB / max 150 kB). +Assumptions: copies 20 → 4 per message (F≈1), hash passes 5 → 1.5; effective alloc+memcpy +throughput under refc ≈ 1–2 GB/s per core (allocation-bound, not memcpy-bound); SHA-256 via +nimcrypto (pure Nim, no SHA-NI) ≈ 300–500 MB/s per core. + +Per-message volume: + +| Payload | Copied (baseline 20×) | Copied (after 4×) | Hashed (5×) | Hashed (1.5×) | Allocs/msg | +|---|---|---|---|---|---| +| 10 kB | 200 kB | 40 kB | 50 kB | 15 kB | ~60–80 → ~12–16 | +| 50 kB | 1.0 MB | 200 kB | 250 kB | 75 kB | " | +| 150 kB | 3.0 MB | 600 kB | 750 kB | 225 kB | " | + +Sustained-rate CPU cost (share of one core spent purely on copy+hash overhead): + +| Scenario | Baseline | After P1–P3 | +|---|---|---| +| 50 kB @ 100 msg/s | ~10–18 % | ~3–5 % | +| 50 kB @ 500 msg/s | ~50–90 % (near saturation) | ~13–22 % | +| 150 kB @ 100 msg/s | ~30–55 % | ~8–14 % | +| 10 kB @ 1000 msg/s | ~20–37 % | ~5–9 % | + +Single-core throughput ceiling from overhead alone (copy+hash time per message): + +| Payload | Baseline ceiling | After P1–P3 | Headroom gain | +|---|---|---|---| +| 10 kB | ~2,700–5,000 msg/s | ~11,000–20,000 msg/s | ~4× | +| 50 kB | ~550–1,000 msg/s | ~2,200–4,000 msg/s | ~4× | +| 150 kB | ~180–330 msg/s | ~740–1,300 msg/s | ~4× | + +GC pressure: heap churn equals the copied volume — at 50 kB × 100 msg/s the baseline churns +~100 MB/s through the refc heap (that much garbage accumulates between collection cycles); +after the fixes ~20 MB/s. Fewer, larger GC cycles → visibly lower tail latency. Per-message +transient footprint at 150 kB drops from ~3 MB to ~0.6 MB. + +These are arithmetic projections, not measurements — the day-one benchmark harness validates +them (and nimcrypto's actual SHA-256 rate, which sets how much of the win comes from P2's +hash-once). + +Suggested order: benchmark harness first (0.5–1 day: fixed-seed message generator through +`uniqueTopicHandler`, measure allocs via `getOccupiedMem`/instrumented `computeMessageHash` +counter) → P3 → P1 → P2 → re-benchmark. Total ≈ **1.5–2 calendar weeks** for P1–P3 including +tests and verification; P4 as follow-up when touching store paths. + +### Memory-model note (refc vs ORC) + +| Copy class | refc | ORC | Fix | +|---|---|---|---| +| async param → closure env | deep copy | elided (move/cursor) | P2 (ref) now; ORC later | +| `complete(fut, val)` | 1–2 copies (`move` degrades) | 1 copy (non-sink) | P4 / upstream sink | +| `await` result bind | copy | copy (lent borrow) | P4 | +| `valueOr`/`?`/bind-out of Result | copy | usually moved (rvalue) | P5 | diff --git a/docs/analysis/async_copy_fix_plan.md b/docs/analysis/async_copy_fix_plan.md new file mode 100644 index 000000000..d54387e46 --- /dev/null +++ b/docs/analysis/async_copy_fix_plan.md @@ -0,0 +1,180 @@ +# Async copy elimination — three-phase implementation plan + +Companion to [async_copy_analysis.md](async_copy_analysis.md). Baseline: ~19 + F +full-payload copies and 4–6 SHA-256 passes per inbound relayed message (refc). + +Executor-grade plans (self-contained, one per phase): +[Phase 1 — benchmark](plan_phase1_bench.md) · +[Phase 2 — WakuMessage ref object](plan_phase2_wakumessage_ref.md) · +[Phase 3 — WakuEnvelope](plan_phase3_wakuenvelope.md) + +Measured blast radius (grep, 2026-07-14): + +| Pattern | src (logos_delivery + library + apps) | tests | +|---|---|---| +| `WakuMessage` occurrences | ~420 in 88 files | 1,714 in 64 files | +| `WakuMessage(...)` constructions | 27 | 1,324 (**1,174 via `fakeWakuMessage`** in `tests/testlib/wakucore.nim:74`) | +| `message:/msg: WakuMessage` params | 88 | 100 | +| `computeMessageHash(` call sites | 28 | 208 | +| `WakuRelayHandler` references | 19 | 5 | +| structural `==` on messages | 5 | ~107 | + +--- + +## Phase 1 — End-to-end performance harness (before/after proof) + +Goal: reproducible numbers on master *before* any change; re-run after each phase. + +Deliverables: +1. `tests/benchmarks/bench_message_path.nim` — two scenarios: + - **micro**: inject N deterministic messages (fixed seed; payload mix 10/50/150 kB) + directly into the relay topicHandler/`uniqueTopicHandler` with archive + filter + (F = 1, 5, 20 subscribed peers) + sync mounted; measures the full internal pipeline + without network noise. + - **macro**: two in-process nodes over loopback gossipsub, publish → receive, same + payload mix; captures libp2p-side copies too. +2. Instrumentation counters under `-d:msgPathCounters` in `waku_core`: + `computeMessageHash` invocations, `WakuMessage.decode` invocations, bytes hashed, + bytes materialized by decode. (~30–40 LOC, zero cost when not defined.) +3. Metrics emitted per run (CSV/JSON to stdout): msg/s, ns/msg (p50/p99), + `getOccupiedMem()` delta, refc `GC_getStatistics()`, counter totals. +4. `make benchmark` target + baseline results committed to `docs/analysis/bench_baseline.md`. + +| Item | New LOC | Changed LOC | +|---|---|---| +| bench harness (micro + macro) | ~280–380 | — | +| counters in waku_core | ~30–40 | ~10 | +| Makefile / nimble task | ~15 | — | +| baseline results doc | ~40 | — | +| **Total** | **~370–480** | **~10** | + +Verify: harness runs deterministically twice with <5 % variance; counters match the +analysis predictions (≥4 decodes, 4–6 hashes per message) — this *validates the analysis* +before we change anything. + +**Time: 1–1.5 days.** + +--- + +## Phase 2 — `WakuMessage` becomes a `ref object` + +Goal: every assignment/capture/bind-out of a message becomes a pointer + RC op instead of +a 3-seq deep copy — no signature changes anywhere. + +What actually changes (`logos_delivery/waku/waku_core/message/message.nim`): +- `WakuMessage* = ref object` (construction syntax `WakuMessage(payload: ...)` is + unchanged, so the 27 src + 1,324 test constructions compile as-is). +- **Structural `==`** proc (~12 LOC) — otherwise ref-identity comparison breaks the ~107 + test assertions. +- `clone*(msg: WakuMessage): WakuMessage` helper (~10 LOC) for deliberate-mutation sites. +- Known aliasing fix: relay publish timestamp patch + ([waku_relay/protocol.nim:680-682](../../logos_delivery/waku/waku_relay/protocol.nim)) + currently relies on value-copy semantics — must become `clone()` (or set the timestamp + before entering publish). +- Nil-safety audit: `var msg: WakuMessage` now defaults to `nil`, not an empty object. + 88 src param sites reviewed; expected actual fixes small (decode already does + `var msg = WakuMessage()`). +- Aliasing audit: any post-construction mutation of a received message (queue_driver + retention becomes a *shared* ref — that's the intended win, but document + immutable-by-convention in message.nim). +- Style-guide deviation note: AGENTS.md prescribes `Ref` suffix for ref object types; + renaming to `WakuMessageRef` would touch ~2,100 occurrences for zero semantic value — + keep the name, document the deviation at the type. + +| Item | New LOC | Changed LOC | +|---|---|---| +| message.nim (ref + `==` + clone + doc comment) | ~35 | ~10 | +| relay publish mutation fix | — | ~10 | +| nil/aliasing audit fixes across src (est.) | — | ~60–120 | +| test fixes (nil defaults, identity assumptions; `==` covers the rest) | — | ~80–200 | +| **Total** | **~35** | **~160–340** | + +Effect on copy budget (per message, from 19 + F): decode bind-outs −4, dispatch env +captures −7, archive/filter wrap −2 → **~6 + F remaining** (the 4 redundant decode +*materializations* and all hashing remain). Expected from Phase 1 harness: **~50–60 % +reduction in alloc volume**, hashes unchanged. + +Memory-model note (refc/ORC): under refc, ref copy = pointer + RC increment (cheap, +non-atomic); no cycles are introduced (WakuMessage has no back-references), so refc's +cycle collector is not engaged. Under a future ORC build, same story with ORC RC. FFI +(`library/`) already deep-copies fields via `copyMem` into JSON events — unaffected; +ref gives a stable pointer, which is FFI-friendly. + +Verify: full `make test` green, benchmark re-run recorded, ASAN/refc build of the +message-path tests clean (lifetimes now shared — this is the run that would catch a +premature-free or unexpected-mutation bug). + +**Time: 2–3 days** (LOC is small; the cost is the aliasing audit + full-suite verification). + +--- + +## Phase 3 — API break: `WakuEnvelope` (WakuMessage, PubsubTopic, msgHash) + +Goal: decode once, hash once, carry both through the whole internal dispatch as one ref. + +Design (`logos_delivery/waku/waku_core/message/envelope.nim`, new): +```nim +type WakuEnvelope* = ref object + msg*: WakuMessage + pubsubTopic*: PubsubTopic + hash*: WakuMessageHash + +proc init*(T: type WakuEnvelope, pubsubTopic: PubsubTopic, msg: WakuMessage): T = + WakuEnvelope(msg: msg, pubsubTopic: pubsubTopic, + hash: computeMessageHash(pubsubTopic, msg)) +``` + +Changes: +1. **Relay decode-once** (`waku_relay/protocol.nim`): ordered validator decodes and builds + the envelope; `onRecv`/`onValidated`/`onSend` observers and the subscribe topicHandler + consume the envelope instead of re-decoding (removes 3 decodes + validator-side hash). + Mechanism: msgId-keyed handoff (small TimedCache) or restructured observer chain. +2. **`WakuRelayHandler`** becomes `proc(envelope: WakuEnvelope): Future[void]` — 19 src + + 5 test references. +3. **Dispatch** (`node/subscription_manager.nim`): `uniqueTopicHandler` + 6 sub-handlers + (trace/filter/archive/sync/internal/legacyApp) + `MessageSeenEvent` take the envelope. +4. **Consumers**: `archive.handleMessage`, `filter_v2.handleMessage`, sync ingress, + lightpush→relay callbacks, REST/API + `library/` event emission — use `envelope.hash`, + deleting ~10–14 of the 28 src `computeMessageHash` call sites. +5. **Filter per-peer buffer share** (folded in here): encode `MessagePush` once, pass + `ref seq[byte]` (or raw-async `pushToPeer`) — removes the F per-peer buffer copies. + +| Item | New LOC | Changed LOC | +|---|---|---| +| envelope.nim + export | ~50 | — | +| relay decode-once + observer restructure | ~60 | ~120–180 | +| WakuRelayHandler + subscription_manager | — | ~90–130 | +| archive / filter / sync / lightpush / events / library | — | ~150–250 | +| filter shared push buffer | — | ~25 | +| tests (relay, node, archive, filter, sync handler tests) | ~40 | ~250–400 | +| **Total** | **~150** | **~640–990** | + +Effect on copy budget: **~6 + F → ~3 copies, 4–6 → 1 hash pass** (plus 1 encode + hash on +the outbound leg). Combined with Phase 2 this is the full projected gain: ~75–80 % less +copy volume, ~70–80 % less hashing → ~4× single-core throughput headroom at the +10/50/150 kB profile. + +Verify: full suite + benchmark re-run vs Phase-1 baseline and Phase-2 checkpoint; counter +assertion in the micro bench: exactly 1 decode + 1 hash per inbound message. + +**Time: 4–6 days.** + +--- + +## Summary + +| Phase | New LOC | Changed LOC | Time | Copies (of 19 + F) | Hashes (of 4–6) | +|---|---|---|---|---|---| +| 1 — perf harness | ~370–480 | ~10 | 1–1.5 d | — (measures baseline) | — | +| 2 — WakuMessage ref | ~35 | ~160–340 | 2–3 d | → ~6 + F | 4–6 (unchanged) | +| 3 — WakuEnvelope | ~150 | ~640–990 | 4–6 d | → ~3 | → 1 | +| **Total** | **~555–665** | **~810–1,340** | **7–10.5 d (~2 wks)** | **−84 %** | **−80 %** | + +Risks to watch: +- Phase 2 aliasing: any code that mutated its by-value copy now mutates a shared message — + the relay timestamp patch is the one known site; the ASAN run + full suite is the net. +- Phase 3 relay observer restructure is the only genuinely fiddly part (gossipsub gives no + validator→handler side channel); the msgId cache must be bounded (TimedCache) and + eviction-safe. +- Test LOC ranges are dominated by handler-signature churn in Phase 3; the 1,174 + `fakeWakuMessage` call sites are untouched in all phases. diff --git a/docs/analysis/plan_phase1_bench.md b/docs/analysis/plan_phase1_bench.md new file mode 100644 index 000000000..aab1d5759 --- /dev/null +++ b/docs/analysis/plan_phase1_bench.md @@ -0,0 +1,125 @@ +# Phase 1 — Message-path performance benchmark (executor plan) + +> Self-contained implementation plan. Context: [async_copy_analysis.md](async_copy_analysis.md) +> and [async_copy_fix_plan.md](async_copy_fix_plan.md). Read both before starting. +> +> **Objective**: a reproducible before/after benchmark of the WakuMessage inbound/outbound +> pipeline, plus decode/hash counters that verify the analysis claims (≥4 decodes and 4–6 +> SHA-256 passes per inbound relayed message on the current code). This phase changes **no +> production behavior** — counters must compile to nothing without the define. + +## Constraints (apply to all phases) + +- Nim 2.2.4, `--mm:refc` (enforced by nimble tasks). Build via `make` / nimble tasks only. +- Format all touched files with nph (`make nph/` or the pre-commit hook). +- Run `gitnexus_impact` before editing cross-module symbols and `gitnexus_detect_changes()` + before committing (project rule in AGENTS.md). Commit at the end of the phase; **never push**. +- Verify claims by running things, not by reasoning ("should work" is not done). + +## Existing infrastructure to reuse + +- `apps/benchmarks/benchmarks.nim` — existing RLN benchmark; nimble task at + `logos_delivery.nimble:391-393` (`task benchmarks` → `buildBinary name, "apps/benchmarks/"`). + Mirror this pattern; note it already imports from `tests/` (`tests/waku_rln_relay/utils_onchain`), + so importing `tests/testlib/*` from `apps/benchmarks/` is accepted practice. +- `tests/testlib/wakucore.nim` — `fakeWakuMessage*` (line ~74) for deterministic messages. +- `tests/testlib/wakunode.nim` (verify exact name) — node construction helpers used by all + node tests; use these for the macro scenario. + +## Step 1 — Instrumentation counters (`-d:msgPathCounters`) + +New file `logos_delivery/waku/waku_core/message/path_counters.nim` (~35 LOC): + +```nim +when defined(msgPathCounters): + type MsgPathCounters* = object + decodeCalls*, hashCalls*: int64 + decodedBytes*, hashedBytes*: int64 + var msgPathCounters* {.global.}: MsgPathCounters + template countDecode*(bytes: int) = ... # inc calls, add bytes + template countHash*(bytes: int) = ... + proc resetMsgPathCounters*() = ... +else: + template countDecode*(bytes: int) = discard + template countHash*(bytes: int) = discard +``` + +Hook points (one line each, guarded template — zero cost when undefined): +- `logos_delivery/waku/waku_core/message/codec.nim:23` — inside + `proc decode*(T: type WakuMessage, buffer: seq[byte])`: `countDecode(buffer.len)`. +- `logos_delivery/waku/waku_core/message/digest.nim:53` — inside `computeMessageHash`: + `countHash(msg.payload.len + msg.meta.len)`. + +Export via `waku_core` module as needed. Single-threaded chronos — plain int64, no atomics. + +## Step 2 — Micro benchmark (in-process pipeline, no network) + +New file `apps/benchmarks/message_path_bench.nim` (~250–320 LOC total with Step 3). + +Setup: one WakuNode with relay mounted + archive (SQLite **in-memory** driver) + store-sync +(reconciliation) mounted; subscribe one shard so `registerRelayHandler` +(`logos_delivery/waku/node/subscription_manager.nim:29`) installs the full +`uniqueTopicHandler` chain (trace/filter/archive/sync/internal handlers, :45-82). + +Injection: per message, replicate the gossipsub inbound sequence using WakuRelay's own +registered artifacts (both are stored in public tables on `WakuRelay`, +`waku_relay/protocol.nim:624,632`): +1. `let data = msg.encode().buffer` (prepared up front, excluded from timing) +2. `discard await relay.topicValidator[shard](shard, Message(topic: shard, data: data))` + — exercises the ordered-validator decode (`protocol.nim:543`) +3. `await relay.topicHandlers[shard](shard, data)` — exercises topicHandler decode + (`protocol.nim:603`) + full dispatch chain + +Note in the output that observer decodes (`protocol.nim:271,326,341`) are NOT exercised by +the micro bench (they fire only in real gossipsub dispatch) — the macro scenario covers them. +If `topicValidator`/`topicHandlers` turn out not to be exported, add a +`when defined(msgPathCounters)` accessor to `waku_relay/protocol.nim` rather than exporting +the fields unconditionally. + +Workload: fixed-seed generator (xorshift or `std/random` with `randomize(42)` equivalent — +NO time-based seed), payload mix **10 kB / 50 kB / 150 kB at 25 % / 50 % / 25 %**, unique +timestamps/nonce per message (avoid dedup), N = 1,000 messages + 100 warmup (excluded). + +## Step 3 — Macro benchmark (two nodes, loopback gossipsub) + +Same file, second scenario: two in-process nodes (testlib helpers), connected, both +subscribed to the shard; node B additionally has archive + sync mounted. Node A publishes +the same workload; node B counts arrivals via an app handler; stop timing when all N +received (with a sane timeout). This path includes libp2p rpcHandler + observers, so +counters here should show the full ≥4 decodes/message. + +## Step 4 — Metrics & output + +Per scenario emit one human block + one machine-readable CSV line to stdout: + +``` +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 +``` + +Sources: `getMonoTime` deltas per message for p50/p99; `getOccupiedMem()` before/after; +refc `GC_getStatistics()` (print raw string); counters from Step 1 (reset between scenarios). + +## Step 5 — Build task + baseline capture + +- Add nimble task `benchMessagePath` to `logos_delivery.nimble` mirroring the existing + `benchmarks` task (:391-393), with `-d:msgPathCounters` added to its flags. +- Run it on the **current branch before Phases 2/3**; save both scenario outputs verbatim + into `docs/analysis/bench_baseline.md` with the commit hash and machine info + (`sysctl -n machdep.cpu.brand_string`, macOS version). + +## Acceptance criteria + +1. `make test` still fully green (production code untouched except no-op templates). +2. A production build **without** `-d:msgPathCounters` compiles and the counter templates + vanish (verify: `nim c` of wakunode2 target still succeeds). +3. Two consecutive micro runs differ by < 5 % on msg/s (determinism check). +4. Macro scenario counters confirm the analysis: `decodes_per_msg ≥ 4`, + `hashes_per_msg` in 4–6 on the receiving node. **If not, STOP and report the measured + numbers before proceeding to Phase 2 — the analysis (and the expected gains) must be + recalibrated first.** +5. `docs/analysis/bench_baseline.md` committed with the numbers. + +## Estimated diff + +~370–480 new LOC, ~10 changed LOC, no behavior change. Budget: 1–1.5 days. diff --git a/docs/analysis/plan_phase2_wakumessage_ref.md b/docs/analysis/plan_phase2_wakumessage_ref.md new file mode 100644 index 000000000..771de8335 --- /dev/null +++ b/docs/analysis/plan_phase2_wakumessage_ref.md @@ -0,0 +1,127 @@ +# Phase 2 — `WakuMessage` becomes a `ref object` (executor plan) + +> Self-contained implementation plan. Context: [async_copy_analysis.md](async_copy_analysis.md), +> [async_copy_fix_plan.md](async_copy_fix_plan.md). Prerequisite: Phase 1 benchmark exists +> and `docs/analysis/bench_baseline.md` is recorded. +> +> **Objective**: change `WakuMessage` from a value `object` to a `ref object` so every +> assignment, async-closure capture, and Result/Option bind-out of a message costs a +> pointer + refcount instead of deep-copying 3 seqs + a string. No proc signatures change. +> Expected result (verify with Phase 1 bench): ~50–60 % reduction in alloc volume per +> message; decode/hash counters unchanged. +> +> **The danger of this phase is not the diff size — it is aliasing.** Code written under +> value semantics may mutate "its copy" of a message; under ref semantics that mutation is +> now shared with every other holder. Every mutation site must be found and either cloned +> or proven to own a freshly constructed message. + +## Constraints + +Same as Phase 1 (refc, nph, gitnexus impact/detect_changes, no push, verify by running). +Additionally: run `gitnexus_impact({target: "WakuMessage", direction: "upstream"})` first +and report the blast radius before editing. + +## Step 1 — The type change (`logos_delivery/waku/waku_core/message/message.nim`) + +Current (lines 12–29): `type WakuMessage* = object` with `payload: seq[byte]`, +`contentTopic: ContentTopic`, `meta: seq[byte]`, `version: uint32`, +`timestamp: Timestamp`, `ephemeral: bool`, `proof: seq[byte]`. + +1. `type WakuMessage* = ref object` — construction syntax `WakuMessage(payload: ...)` is + unchanged, so the 27 src + ~1,324 test constructions (1,174 via `fakeWakuMessage`, + `tests/testlib/wakucore.nim:74`) compile as-is. +2. Add a **structural `==`** (~12 LOC): both-nil → true, one-nil → false, else field-wise + compare. Without this, ~107 test assertions comparing messages break on ref identity. +3. Add `proc clone*(msg: WakuMessage): WakuMessage` — field-wise copy into a fresh ref + (deep-copies the seqs; that is the point: cloning is now *explicit and rare*). +4. **Fix `ensureTimestampSet` (lines 31–34).** Current body `result = message; + result.timestamp = ...` aliases the input under ref semantics and mutates the *shared* + message. Rewrite non-mutating: + ```nim + proc ensureTimestampSet*(message: WakuMessage): WakuMessage = + if message.timestamp != 0: + return message + var m = message.clone() + m.timestamp = getNowInNanosecondTime() + m + ``` +5. Doc comment at the type: messages are **immutable by convention after construction**; + mutate only freshly constructed or explicitly `clone()`d instances. Also note the + deliberate deviation from the AGENTS.md `Ref`-suffix rule (renaming ~2,100 occurrences + has no semantic value). + +## Step 2 — Known aliasing sites (fix, don't just audit) + +1. **Relay publish** `logos_delivery/waku/waku_relay/protocol.nim:680-682`: + ```nim + var message = wakuMessage + if message.timestamp == 0: + message.timestamp = getNowInNanosecondTime() + ``` + Under ref this mutates the caller's message. Replace with + `let message = wakuMessage.ensureTimestampSet()` (safe after Step 1.4). +2. **RLN proof attachment** — search `logos_delivery/waku/waku_rln_relay/` for procs taking + `msg: var WakuMessage` or assigning `msg.proof = ` (e.g. `appendRLNProof`). These run on + the *outbound* path on a message the API caller still holds. Decision rule: if the + message came in through a public API, `clone()` before mutating (or restructure to + build-then-assign locally). +3. **Sweep for every other field mutation**: + `grep -rn --include='*.nim' '\.\(payload\|meta\|proof\|timestamp\|ephemeral\|version\|contentTopic\) *=' logos_delivery library apps` + Classify each hit: (a) freshly constructed in the same scope (e.g. the protobuf decode + at `waku_core/message/codec.nim:23` building `var msg = WakuMessage()`) → fine; + (b) received/parameter/stored message → `clone()` first or restructure. Record the + classification of every hit in the phase notes (this list is the review artifact). + +## Step 3 — Nil-safety audit + +`var m: WakuMessage` now defaults to `nil`, not an empty object; field access on it crashes. + +- `grep -rn --include='*.nim' 'var [a-zA-Z_]*: WakuMessage\b\|: WakuMessage$' logos_delivery library apps` — + any declaration relying on default-init must become `= WakuMessage()`. +- Check emptiness idioms: `== WakuMessage()` / `== default(WakuMessage)` comparisons and + `Table[..., WakuMessage].getOrDefault` consumers (getOrDefault now yields `nil`). +- Fields of type `WakuMessage` inside other objects (e.g. `MessagePush.wakuMessage` in + `waku_filter_v2/rpc.nim`, store RPC types, `queue_driver`'s + `SortedSet[Index, WakuMessage]`) — their decode paths must construct explicitly; verify + each codec assigns a constructed message (`rpc_codec.nim` files for filter/lightpush/store). +- `Option[WakuMessage]`/`Opt[WakuMessage]` (only 2 src occurrences): `none`/absent now also + representable as nil ref — do NOT collapse the two; keep the Opt wrapper semantics as-is. + +## Step 4 — Tests + +- Full `make test`. Expected failure classes: nil-default (Step 3 misses), ref-identity + assumptions (tests mutating a fixture then asserting the original unchanged — now shared), + and `check msg1 == msg2` (covered by structural `==`). +- Do NOT weaken test assertions to make them pass; fix the production/nil issue instead. +- Add new unit tests in `tests/waku_core/` (~40 LOC): structural `==` incl. nil cases; + `clone()` independence (mutating clone leaves original untouched); `ensureTimestampSet` + returns same ref when timestamp set / fresh ref when not. + +## Step 5 — Memory-model verification (mandatory before "done") + +- refc: full suite + the Phase 1 benchmark; record the checkpoint numbers next to + `bench_baseline.md` (expect alloc-volume drop, identical decode/hash counters, identical + message throughput or better). +- ASAN run of the message-path tests (clang, refc) — this is the net that catches + a use-after-free from the new shared lifetimes: at minimum + `tests/waku_relay`, `tests/waku_archive`, `tests/waku_filter_v2`, `tests/node`. +- Statement for the record (per FFI rules): `library/` C bindings deep-copy message fields + via `copyMem` into JSON events (`library/json_message_event.nim:72-95`), so no message + ref crosses the FFI boundary; refc vs ORC behavior of the ref itself is standard RC — + no finalizers, no cycles (WakuMessage has no back-references). + +## Acceptance criteria + +1. Full `make test` green; ASAN run clean. +2. Step 2.3 mutation-site classification list produced (in PR description or phase notes). +3. Phase 1 bench re-run recorded: alloc volume down ≥ 40 % (projection is 50–60 %), + decode/hash counters unchanged vs baseline. If alloc volume barely moves, something + still deep-copies — investigate before closing the phase (prime suspects: a `var` + parameter forcing a copy, or a leftover value-type field of WakuMessage inside a + container that the compiler still copies). +4. `gitnexus_detect_changes()` reviewed; commit (no push). + +## Estimated diff + +~35 new LOC, ~160–340 changed LOC (dominated by Steps 2.3/3 fixes and test adjustments). +Budget: 2–3 days, most of it audit + verification, not typing. diff --git a/docs/analysis/plan_phase3_wakuenvelope.md b/docs/analysis/plan_phase3_wakuenvelope.md new file mode 100644 index 000000000..01f1fa410 --- /dev/null +++ b/docs/analysis/plan_phase3_wakuenvelope.md @@ -0,0 +1,165 @@ +# Phase 3 — `WakuEnvelope` API break: decode once, hash once (executor plan) + +> Self-contained implementation plan. Context: [async_copy_analysis.md](async_copy_analysis.md), +> [async_copy_fix_plan.md](async_copy_fix_plan.md). Prerequisites: Phases 1 and 2 merged +> (WakuMessage is a `ref object`; benchmark + counters exist). +> +> **Objective**: introduce `WakuEnvelope` (message + pubsubTopic + msgHash, one ref) as the +> unit that flows through internal dispatch, eliminate redundant proto decodes and +> `computeMessageHash` recomputation, and stop copying the filter push buffer per peer. +> Target, verified by Phase 1 counters: **≤ 2 decodes and 1 hash per inbound relayed +> message** (from ≥ 4 and 4–6), plus 1 encode + 1 hash on the outbound publish leg. +> +> This is a deliberate **internal API break**: `WakuRelayHandler` changes shape, and every +> registered handler follows. + +## Constraints + +Same as previous phases (refc, nph, gitnexus impact/detect_changes, no push, verify by +running). Run `gitnexus_impact({target: "WakuRelayHandler", direction: "upstream"})` and +report before editing. + +## Design decisions (already made — do not re-litigate) + +1. **No validator→handler cache.** The gossipsub validator and the topic handler have no + shared side channel, and keying a handoff cache by a non-cryptographic hash of the raw + bytes would allow attacker-crafted collisions to substitute messages. We accept **2** + decodes (validator + topicHandler); the envelope kills the other 2–3 decodes and all + re-hashing. A keyed cache can be a later optimization with its own security review. +2. **The `onRecv` observer decode is deleted outright** — its decoded message is unused + (see Step 2). +3. `WakuEnvelope` is a `ref object` and immutable by convention, same as WakuMessage. + +## Step 1 — The type + +New file `logos_delivery/waku/waku_core/message/envelope.nim` (~50 LOC), exported from +`waku_core`: + +```nim +type WakuEnvelope* = ref object + msg*: WakuMessage + pubsubTopic*: PubsubTopic + hash*: WakuMessageHash + +proc init*(T: type WakuEnvelope, pubsubTopic: PubsubTopic, msg: WakuMessage): T = + WakuEnvelope(msg: msg, pubsubTopic: pubsubTopic, + hash: computeMessageHash(pubsubTopic, msg)) +``` + +Plus `shortLog`/`$` helper for chronicles fields (hash as 0x-hex). + +## Step 2 — WakuRelay (`logos_delivery/waku/waku_relay/protocol.nim`) + +Current inbound anatomy (line refs at time of writing): +- ordered validator decodes (`generateOrderedValidator`, :536-565, decode at :543, + hash-on-reject at :553) +- observers (`initRelayObservers`, :255-353): `onRecv` :301 decodes via + `decodeRpcMessageInfo` (:256) but `updateMetrics` (:283) **never uses the decoded + message** — only sizes; `onValidated` :324 decodes only to log; `onSend` :339 decodes + to log + metrics +- topicHandler wrapper (`subscribe`, :596-634, decode at :603) → calls the + `WakuRelayHandler` +- `publish` (:674-695): timestamp fix (Phase 2 made it `ensureTimestampSet`), encode, + hash, gossipsub publish + +Changes: +1. **`WakuRelayHandler`** (type in this module; 19 src + 5 test references repo-wide) + becomes `proc(envelope: WakuEnvelope): Future[void] {.gcsafe, raises: [].}` (keep the + existing pragma set). +2. **topicHandler wrapper** (:600-616): after the (kept) decode, build + `let envelope = WakuEnvelope.init(pubsubTopic, decMsg)` — the **single hash + computation** for the inbound path — and call `handler(envelope)`. +3. **Ordered validator** (:536-565): decode stays (validators need the message). The + reject-path hash (:553) may stay — it only runs on rejected messages. +4. **Observers**: + - `onRecv` (:301-322): delete the `decodeRpcMessageInfo` call; keep control-message + handling and byte metrics computed from `msg.data.len + msg.topic.len` directly. + - `onValidated` (:324-337): decodes only for `logMessageInfo`. Wrap in a compile-time + or log-level gate (chronicles `logLevel >= some threshold` pattern used elsewhere in + the repo) so production INFO builds do not pay a decode per message; alternatively + downgrade to logging msgId + topic only (no decode). Pick whichever keeps current + log content at DEBUG/TRACE. + - `onSend` (:339-348): same treatment as onValidated. +5. **`publish`** (:674-695): after encode, `computeMessageHash` stays (outbound leg, + 1 hash) — no structural change beyond what Phase 2 did. +6. `validateMessage` (:567-594, lightpush entry): currently re-encodes just for the size + check and re-hashes for logging. Add an overload/param taking the already-encoded + buffer length and an optional precomputed hash so the lightpush path (which has the + buffer) stops double-encoding. Small, contained. + +## Step 3 — Dispatch (`logos_delivery/waku/node/subscription_manager.nim:29-85`) + +`registerRelayHandler`: `traceHandler`/`filterHandler`/`archiveHandler`/`syncHandler`/ +`internalHandler` (:45-70), `uniqueTopicHandler` (:72-84), and the +`legacyAppHandlers: Table[PubsubTopic, WakuRelayHandler]` all switch to the envelope +signature. Inside, replace `(topic, msg)` argument pairs with `envelope` and use +`envelope.msg.payload.len` etc. `MessageSeenEvent.emit(node.brokerCtx, topic, msg)` (:70): +change the event to carry the envelope (see Step 5). + +## Step 4 — Consumers (each loses its own `computeMessageHash`) + +28 src `computeMessageHash` call sites exist; ~10–14 die here. For each consumer, change +the entry point signature and use `envelope.hash`: + +| Consumer | Entry point | Redundant hash removed | +|---|---|---| +| Archive | `waku_archive/archive.nim:98` `handleMessage` | :101 | +| Filter | `waku_filter_v2/protocol.nim:243` `handleMessage` | :246 and :195 | +| Store-sync | `waku_store_sync/reconciliation.nim` `messageIngress` (called from subscription_manager:67) | :78 | +| Lightpush→relay | `waku_lightpush*/callbacks.nim:16-24` (builds the relay publish) | :24 (hash once when building the envelope/publish result) | +| API/REST relay | `api/relay.nim` (:30) and any REST messaging handler registering a WakuRelayHandler (this branch's messaging REST endpoints) | :30 | +| library/FFI events | `library/json_message_event.nim` (:88) via MessageSeenEvent | :88 | + +Keep `computeMessageHash` call sites that hash *other* messages (store queries hashing +stored rows, sync transfer :171, filter rpc :95 on the client side, etc.) — only remove +recomputation of an inbound message's own hash where an envelope is in hand. + +## Step 5 — Broker event + +`MessageSeenEvent` (`logos_delivery/api/events/kernel_events`): change payload from +`(topic, msg)` to the envelope (or add hash alongside, whichever keeps broker codegen +simplest). Update its listeners — notably the library/JSON event path, which then uses +`envelope.hash` instead of recomputing (:88) — and any messaging-REST event cache on this +branch that listens to MessageSeenEvent. + +## Step 6 — Filter per-peer buffer share (`waku_filter_v2/protocol.nim`) + +`handleMessage` → `pushToPeers` (:213-216) encodes `MessagePush` **once** (good) but then +`pushToPeer(peerId, buffer)` (:170) is `{.async.}` with a by-value `seq[byte]` — one +buffer deep-copy per subscribed peer. Fix: make `pushToPeer` `{.async: (raw: true).}` or +pass `ref seq[byte]`. One proc + call site, ~25 LOC. + +## Step 7 — Tests + +- Update every handler construction to the envelope signature (grep `WakuRelayHandler` + in tests, plus tests calling `handleMessage(topic, msg)` on archive/filter and + `messageIngress` directly): wrap with `WakuEnvelope.init(topic, msg)`. +- Add `toEnvelope(topic, msg)` convenience to `tests/testlib/wakucore.nim` if it reduces + churn (thin alias of `WakuEnvelope.init`). +- New unit tests (~40 LOC): envelope init computes the same hash as `computeMessageHash`; + relay topicHandler delivers an envelope whose hash matches an independently computed one. +- Affected suites to run individually while iterating: `tests/waku_relay`, `tests/node`, + `tests/waku_archive`, `tests/waku_filter_v2`, `tests/waku_lightpush`, + `tests/waku_store_sync`, plus the REST/API suites on this branch. + +## Step 8 — Verification (mandatory) + +1. Full `make test` green. +2. Phase 1 macro benchmark: **counter assertion is the acceptance gate** — + `decodes_per_msg ≤ 2` and `hashes_per_msg == 1` on the receiving node + (observer logging gated per Step 2.4; if the gate is log-level-based, run the bench at + INFO). Micro bench: expect ~75–80 % alloc-volume reduction vs `bench_baseline.md` and + the throughput gain in the analysis tables (~4× headroom at the 10/50/150 kB profile). +3. Record the after-numbers in `docs/analysis/bench_baseline.md` (append a "Phase 3" + section with commit hash). +4. `gitnexus_detect_changes()` reviewed — the affected set must be: waku_core (new file), + waku_relay, node/subscription_manager, waku_archive, waku_filter_v2, waku_lightpush, + waku_store_sync, api/events + api/relay, library, tests. Anything outside that list is + scope creep — stop and reassess. +5. Commit (no push). + +## Estimated diff + +~150 new LOC, ~640–990 changed LOC (tests ~250–400 of that). Budget: 4–6 days. The only +genuinely fiddly part is Step 2 (relay observers + handler wrapper); do it first, get +`tests/waku_relay` green, then fan out to consumers mechanically. From 50f9b97ba8800815e18b9d9a4e1993d0c297012d Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:20:13 +0200 Subject: [PATCH 2/4] feat(bench): add -d:msgPathCounters decode/hash instrumentation Zero-cost message-path counters gated on -d:msgPathCounters. When the define is absent, countDecode/countHash expand to discard: no globals, no runtime cost, no behaviour change. Hooked into WakuMessage.decode (codec) and computeMessageHash (digest), exported via waku_core. Adds benchTopicValidator/benchTopicHandler accessors on WakuRelay (also define-gated) so the message-path benchmark can replay the inbound validator+handler decode path without a live network. Co-Authored-By: Claude Fable 5 --- logos_delivery/waku/waku_core/message.nim | 9 ++++- .../waku/waku_core/message/codec.nim | 3 +- .../waku/waku_core/message/digest.nim | 3 +- .../waku/waku_core/message/path_counters.nim | 40 +++++++++++++++++++ logos_delivery/waku/waku_relay/protocol.nim | 12 ++++++ 5 files changed, 63 insertions(+), 4 deletions(-) create mode 100644 logos_delivery/waku/waku_core/message/path_counters.nim diff --git a/logos_delivery/waku/waku_core/message.nim b/logos_delivery/waku/waku_core/message.nim index 6b698fb01..3f2d93d8a 100644 --- a/logos_delivery/waku/waku_core/message.nim +++ b/logos_delivery/waku/waku_core/message.nim @@ -1,3 +1,8 @@ -import ./message/message, ./message/default_values, ./message/codec, ./message/digest +import + ./message/message, + ./message/default_values, + ./message/codec, + ./message/digest, + ./message/path_counters -export message, default_values, codec, digest +export message, default_values, codec, digest, path_counters diff --git a/logos_delivery/waku/waku_core/message/codec.nim b/logos_delivery/waku/waku_core/message/codec.nim index 9b01cf667..f5cc65c1a 100644 --- a/logos_delivery/waku/waku_core/message/codec.nim +++ b/logos_delivery/waku/waku_core/message/codec.nim @@ -4,7 +4,7 @@ # - Proto definition: https://github.com/vacp2p/waku/blob/main/waku/message/v1/message.proto {.push raises: [].} -import ../../common/protobuf, ../topics, ../time, ./message +import ../../common/protobuf, ../topics, ../time, ./message, ./path_counters proc encode*(message: WakuMessage): ProtoBuffer = var buf = initProtoBuffer() @@ -21,6 +21,7 @@ proc encode*(message: WakuMessage): ProtoBuffer = buf proc decode*(T: type WakuMessage, buffer: seq[byte]): ProtobufResult[T] = + countDecode(buffer.len) var msg = WakuMessage() let pb = initProtoBuffer(buffer) diff --git a/logos_delivery/waku/waku_core/message/digest.nim b/logos_delivery/waku/waku_core/message/digest.nim index 3f82ce8f6..215445f68 100644 --- a/logos_delivery/waku/waku_core/message/digest.nim +++ b/logos_delivery/waku/waku_core/message/digest.nim @@ -1,7 +1,7 @@ {.push raises: [].} import std/sequtils, stew/[byteutils, endians2, arrayops], nimcrypto/sha2, results -import ../topics, ./message +import ../topics, ./message, ./path_counters ## 14/WAKU2-MESSAGE: Deterministic message hashing ## https://rfc.vac.dev/spec/14/#deterministic-message-hashing @@ -51,6 +51,7 @@ proc hexToHash*(hexString: string): Result[WakuMessageHash, string] = return ok(hash) proc computeMessageHash*(pubsubTopic: PubsubTopic, msg: WakuMessage): WakuMessageHash = + countHash(msg.payload.len + msg.meta.len) var ctx: sha256 ctx.init() defer: diff --git a/logos_delivery/waku/waku_core/message/path_counters.nim b/logos_delivery/waku/waku_core/message/path_counters.nim new file mode 100644 index 000000000..3fed1db6f --- /dev/null +++ b/logos_delivery/waku/waku_core/message/path_counters.nim @@ -0,0 +1,40 @@ +## Message-path instrumentation counters (compile-time gated). +## +## Enabled only with `-d:msgPathCounters`. When the define is absent every hook +## below expands to `discard` — no globals, no runtime cost, no behaviour change. +## These counters exist purely so the message-path benchmark +## (`apps/benchmarks/message_path_bench.nim`) can validate the decode/hash-per- +## message claims in `docs/analysis/async_copy_analysis.md`. +## +## Single-threaded chronos assumption: plain `int64`, no atomics. + +{.push raises: [].} + +when defined(msgPathCounters): + type MsgPathCounters* = object + decodeCalls*: int64 ## `WakuMessage.decode` invocations + hashCalls*: int64 ## `computeMessageHash` invocations + decodedBytes*: int64 ## bytes fed into decode (proto buffer length) + hashedBytes*: int64 ## bytes fed into the hash (payload + meta) + + var msgPathCounters* {.global.}: MsgPathCounters + + template countDecode*(bytes: int) = + msgPathCounters.decodeCalls += 1 + msgPathCounters.decodedBytes += int64(bytes) + + template countHash*(bytes: int) = + msgPathCounters.hashCalls += 1 + msgPathCounters.hashedBytes += int64(bytes) + + proc resetMsgPathCounters*() = + msgPathCounters = MsgPathCounters() + +else: + template countDecode*(bytes: int) = + discard + + template countHash*(bytes: int) = + discard + +{.pop.} diff --git a/logos_delivery/waku/waku_relay/protocol.nim b/logos_delivery/waku/waku_relay/protocol.nim index 409b02929..0d4bd65f5 100644 --- a/logos_delivery/waku/waku_relay/protocol.nim +++ b/logos_delivery/waku/waku_relay/protocol.nim @@ -633,6 +633,18 @@ proc subscribe*(w: WakuRelay, pubsubTopic: PubsubTopic, handler: WakuRelayHandle w.topicHealthDirty.incl(pubsubTopic) w.topicHealthUpdateEvent.fire() +when defined(msgPathCounters): + ## Benchmark-only accessors (see `apps/benchmarks/message_path_bench.nim`). + ## They expose the per-topic gossipsub artifacts so the micro benchmark can + ## replay the inbound validator + handler decode path without a live network. + ## Guarded by `-d:msgPathCounters` so production builds never see them and the + ## `topicValidator`/`topicHandlers` fields stay module-private. + proc benchTopicValidator*(w: WakuRelay, pubsubTopic: PubsubTopic): ValidatorHandler = + w.topicValidator.getOrDefault(pubsubTopic) + + proc benchTopicHandler*(w: WakuRelay, pubsubTopic: PubsubTopic): TopicHandler = + w.topicHandlers.getOrDefault(pubsubTopic) + proc unsubscribeAll*(w: WakuRelay, pubsubTopic: PubsubTopic) = ## Unsubscribe all handlers on this pubsub topic From cf8502eb4975ae6bc8a5667a5d179cd154e8e1d1 Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:40:52 +0200 Subject: [PATCH 3/4] feat(bench): message-path micro+macro benchmark harness apps/benchmarks/message_path_bench.nim replays the WakuMessage inbound pipeline two ways: micro (in-process, drives the relay's registered ordered-validator + topicHandler + full uniqueTopicHandler dispatch chain over relay/archive/store-sync, no network) and macro (two in-process nodes over loopback gossipsub, publisher -> receiver with archive + sync, capturing the libp2p observer decodes). Fixed-seed workload (seed 42), 10/50/150 kB at 25/50/25 %, N=1000 + warmup, unique per-message payload prefix + timestamp to avoid dedup. Emits a human block + CSV line per scenario (msg/s, ns/msg p50/p99, decodes/ hashes per msg, decoded/hashed MB, occupied-mem delta, GC stats); micro runs twice for a determinism check. Paced macro publishing keeps the single loopback send queue from overflowing so all N are delivered. New nimble task benchMessagePath builds with -d:msgPathCounters and links librln (the node stack pulls RLN transitively) then runs it. Co-Authored-By: Claude Fable 5 --- apps/benchmarks/message_path_bench.nim | 447 +++++++++++++++++++++++++ logos_delivery.nimble | 9 + 2 files changed, 456 insertions(+) create mode 100644 apps/benchmarks/message_path_bench.nim diff --git a/apps/benchmarks/message_path_bench.nim b/apps/benchmarks/message_path_bench.nim new file mode 100644 index 000000000..af9333166 --- /dev/null +++ b/apps/benchmarks/message_path_bench.nim @@ -0,0 +1,447 @@ +## 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() diff --git a/logos_delivery.nimble b/logos_delivery.nimble index 8c4d32cb5..0b9d5f76e 100644 --- a/logos_delivery.nimble +++ b/logos_delivery.nimble @@ -392,6 +392,15 @@ task benchmarks, "Some benchmarks": let name = "benchmarks" buildBinary name, "apps/benchmarks/", "-p:../.." +task benchMessagePath, "Message-path decode/hash benchmark (micro + macro)": + # Imports the full node stack (relay + archive + store-sync), which pulls in + # RLN transitively via node/waku_node/relay.nim, so librln must be linked. + let name = "message_path_bench" + buildBinary name, "apps/benchmarks/", + "-p:../.. -d:msgPathCounters -d:chronicles_log_level=ERROR " & + "--passL:librln_v2.0.2.a --passL:-lm" + exec "build/" & name + task wakucanary, "Build waku-canary tool": let name = "wakucanary" buildBinary name, "apps/wakucanary/" From 2a4da679f2022268afcfe7c979e41e6f8c028bb9 Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:41:39 +0200 Subject: [PATCH 4/4] docs(bench): Phase 1 message-path baseline results Records the micro + macro baseline (Apple M4, macOS 26.5, refc, Nim 2.2.4) with both scenarios' verbatim output, CSV, per-node decode/hash breakdown, and the Phase 1 acceptance verdict. Validates the analysis: receiver decodes the same proto bytes 4x, aggregate 6 decodes / 5 hashes per relayed message; micro determinism 4.03% (<5%). Co-Authored-By: Claude Fable 5 --- docs/analysis/bench_baseline.md | 88 +++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 docs/analysis/bench_baseline.md diff --git a/docs/analysis/bench_baseline.md b/docs/analysis/bench_baseline.md new file mode 100644 index 000000000..c1b3b7704 --- /dev/null +++ b/docs/analysis/bench_baseline.md @@ -0,0 +1,88 @@ +# Message-path benchmark — Phase 1 baseline + +Baseline capture on the **current branch before Phases 2/3**, produced by +`apps/benchmarks/message_path_bench.nim` (see +[plan_phase1_bench.md](plan_phase1_bench.md)). Re-run after each phase and diff +against these numbers. + +## Environment + +| Item | Value | +|---|---| +| Commit (harness) | `cf8502eb4975ae6bc8a5667a5d179cd154e8e1d1` | +| Branch | `experimental/chore-nocopy-e2e-perf-harness` | +| Machine | Apple M4 | +| OS | macOS 26.5 (arm64) | +| Nim | 2.2.4, `--mm:refc` | +| Build | `-d:msgPathCounters -d:chronicles_log_level=ERROR`, `--passL:librln_v2.0.2.a` | +| Run | `nimble benchMessagePath` (or `./build/message_path_bench`) | +| Workload | seed 42, payload mix 10/50/150 kB @ 25/50/25 %, N=1000 + 100 warmup | + +## Results (CSV) + +``` +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 +micro_run1,1000,10/50/150kB@25/50/25,828.591,1206.9,641375,1925042,2.000,2.000,130.00,130.11,1.54,7 +micro_run2,1000,10/50/150kB@25/50/25,795.166,1257.6,619167,1810625,2.000,2.000,130.00,130.11,1.51,8 +macro,1000,10/50/150kB@25/50/25,4367.314,229.0,3510750,10701375,6.000,5.000,325.00,390.33,141.58,10 +``` + +| Scenario | msg/s | decodes/msg | hashes/msg | decoded_MB | hashed_MB | occ_mem_delta_MB | +|---|---|---|---|---|---|---| +| micro run1 | 1206.9 | 2.000 | 2.000 | 130.11 | 130.00 | 1.54 | +| micro run2 | 1257.6 | 2.000 | 2.000 | 130.11 | 130.00 | 1.51 | +| macro | 229.0 | 6.000 | 5.000 | 390.33 | 325.00 | 141.58 | + +Micro determinism: run1=1206.9 msg/s, run2=1257.6 msg/s → **variance 4.03 %** +(< 5 % target; two earlier back-to-back runs were 1.90 % and 3.12 %). The +decode/hash *counts* are byte-exact reproducible across every run. + +## Interpretation vs. the analysis + +**Decode / hash counts validate [async_copy_analysis.md](async_copy_analysis.md).** + +The counter is a single process-global (both in-process nodes share it), so the +macro per-message numbers are the aggregate of publisher (node A) + receiver +(node B). Breakdown of the macro **6 decodes / 5 hashes** per message: + +| Counter | Node A (publish) | Node B (receive) | Total | +|---|---|---|---| +| `WakuMessage.decode` | 1 (`onSend` observer) | 4 (ordered validator, `onRecv`, `onValidated`, subscribe topicHandler) | **6** | +| `computeMessageHash` | 2 (`publish` :686, `onSend`→`logMessageInfo`) | 3 (`onValidated`→`logMessageInfo`, archive, store-sync) | **5** | + +- The **inbound-decode claim (≥4 decodes on a relayed message)** is confirmed: + node B alone decodes the same proto bytes **4×** + (`waku_relay/protocol.nim` validator :543, `onRecv` :271, `onValidated` :326, + topicHandler :603). +- The micro scenario isolates the non-observer path and shows the **2 decodes / + 2 hashes** it exercises (validator + topicHandler decode; archive + store-sync + hash) — the observers only fire under real libp2p dispatch, which the macro + scenario adds. + +### Caveats + +1. **Log level elides some hash sites.** Built at `chronicles_log_level=ERROR`, + `computeMessageHash` calls that appear *only* as disabled log arguments are + not evaluated (chronicles skips disabled-sink args) — notably the + `node.publish` `notice` at `waku_node/relay.nim:148`. Sites bound to a `let` + (archive :101, store-sync :78, `logMessageInfo` :217, `publish` :686) run + regardless of level and are the ones counted here. +2. **Filter not mounted.** Per the plan, node B has archive + store-sync but not + filter. The analysis's "4–6 hashes" upper end assumes an active filter + (`waku_filter_v2` adds ~2 more receiver-side `computeMessageHash` passes, + :195 + :246). With filter mounted the receiver-side hash count would rise + from 3 toward the analysis's upper bound. +3. **`occupied_mem_delta`** is a coarse `getOccupiedMem()` before/after delta on + the refc heap, not a per-message allocation trace; use it for order-of- + magnitude trend, not exact bytes. `decoded_MB` / `hashed_MB` (exact byte + totals fed to decode/hash) are the reliable volume metrics. + +## Acceptance gate (Phase 1) + +| # | Criterion | Verdict | +|---|---|---| +| 1 | Representative tests green (production untouched except no-op templates) | **PASS** — `tests/waku_core/test_message_digest` 5/5, `tests/waku_relay/test_protocol` 20/20 | +| 2 | Production build without `-d:msgPathCounters` compiles (templates vanish) | **PASS** — `nim check apps/wakunode2` clean; relay test built without the define | +| 3 | Two consecutive micro runs < 5 % msg/s variance | **PASS** — 4.03 % (and 1.90 % / 3.12 % on earlier runs) | +| 4 | Macro `decodes_per_msg ≥ 4` and `hashes_per_msg` in 4–6 | **PASS** — decodes 6.0, hashes 5.0 (process aggregate; receiver-side decodes = 4) | +| 5 | This baseline doc committed | **PASS** |