Event & Request
Broker Design

Callbacks Wired at Init Time

A raw function pointer is captured during construction and stored without lifetime management.

type FilterPushHandler* =
  proc(pubsubTopic: PubsubTopic,
       message: WakuMessage)
    {.async, gcsafe, closure.}
        

type SubscriptionManager* = object
  subscriptions:
    TableRef[(string, ContentTopic),
              FilterPushHandler]   # raw proc pointer

proc registerSubscription*(m, pubsubTopic,
    contentTopic, handler: FilterPushHandler) =
  m.subscriptions[(pubsubTopic, contentTopic)]
    = handler   # stored forever, no cleanup API
        

Raw Push Handlers in the Filter Client

type WakuFilterClient* = ref object of LPProtocol
  rng: ref HmacDrbgContext
  peerManager: PeerManager
  pushHandlers: seq[FilterPushHandler] # ← raw list

# Anyone wiring up the node must call this
proc registerPushHandler*(
    wfc: WakuFilterClient,
    handler: FilterPushHandler) =
  wfc.pushHandlers.add(handler)     # add and forget

# Deep inside recv_service initialization:
node.wakuFilterClient
  .registerPushHandler(
    proc(pubsubTopic: PubsubTopic,
         message: WakuMessage) {.async, closure.} =
      recvService.recentReceivedMsgs.add(...)
  )
    

Raw Push Handlers — What Goes Wrong

The Coupling Chain

Every new subscriber forces a new dependency across the entire init hierarchy.

EventBroker at a Glance

-- Declare once:
EventBroker:
  type MyEvent = object
    field: string

-- Anywhere in your code:
MyEvent.listen(handler)        # register
MyEvent.listen(ctx, handler)   # context-aware
MyEvent.emit(event)            # fire
MyEvent.emit(ctx, field = "x") # shorthand ctor
await MyEvent.dropListener(handle)   # drain in-flight, then cleanup
MyEvent.dropAllListeners()           # drain all in-flight (5s timeout)
        

Step 1 · Declare Your Events

One macro call per event type — can live in a shared events module.

import waku/common/broker/event_broker

EventBroker:
  type DeliveryFeedbackEvent* = ref object
    success*:  DeliverySuccess
    dir*:      DeliveryDirection
    comment*:  string
    msgHash*:  WakuMessageHash
    msg*:      WakuMessage

EventBroker:
  type OnFilterSubscribeEvent* = object
    pubsubTopic*:   string
    contentTopics*: seq[string]

EventBroker:
  type OnFilterUnSubscribeEvent* = object
    pubsubTopic*:   string
    contentTopics*: seq[string]
    

Step 2 · Emit Events

The emitter doesn't import or know about any listener.

proc subscribe*(wfc: WakuFilterClient, ...) {.async.} =
  # ... network call ...
  ?await wfc.sendSubscribeRequest(peer, req)

  # Fire-and-forget: no knowledge of RecvService
  OnFilterSubscribeEvent.emit(
    wfc.brokerCtx, pubsubTopic, contentTopicSeq)

proc unsubscribe*(wfc: WakuFilterClient, ...) {.async.} =
  # ... network call ...
  ?await wfc.sendSubscribeRequest(peer, req)

  OnFilterUnSubscribeEvent.emit(
    wfc.brokerCtx, pubsubTopic, contentTopicSeq)
    

Step 2 · Emit Events (continued)

proc performDeliveryFeedback(
    self: RecvService,
    success: DeliverySuccess,
    dir: DeliveryDirection,
    comment: string,
    msgHash: WakuMessageHash,
    msg: WakuMessage,
) {.gcsafe, raises: [].} =
  info "recv monitor performDeliveryFeedback",
    success, dir, comment,
    msg_hash = shortLog(msgHash)

  DeliveryFeedbackEvent.emit(
    brokerCtx = self.brokerCtx,
    success   = success,
    dir       = dir,
    comment   = comment,
    msgHash   = msgHash,
    msg       = msg,
  )
    

Step 3 · Listen & Clean Up

RecvService manages its own listeners — zero coupling with the filter client.

type RecvService* = ref object of RootObj
  brokerCtx:             BrokerContext
  onSubscribeListener:   OnFilterSubscribeEventListener
  onUnsubscribeListener: OnFilterUnSubscribeEventListener

proc startRecvService*(self: RecvService) =
  self.onSubscribeListener =
    OnFilterSubscribeEvent.listen(
      self.brokerCtx,
      proc(ev: OnFilterSubscribeEvent)
          {.async: (raises: []).} =
        self.onSubscribe(
          ev.pubsubTopic, ev.contentTopics)
    ).valueOr:
      error "Failed to set listener"
      quit(QuitFailure)
    

Step 3 · Listen & Clean Up (continued)

  self.onUnsubscribeListener =
    OnFilterUnSubscribeEvent.listen(
      self.brokerCtx,
      proc(ev: OnFilterUnSubscribeEvent)
          {.async: (raises: []).} =
        self.onUnsubscribe(
          ev.pubsubTopic, ev.contentTopics)
    ).valueOr:
      error "Failed to set listener"
      quit(QuitFailure)

proc stopRecvService*(self: RecvService) {.async.} =
  OnFilterSubscribeEvent.dropListener(
    self.brokerCtx, self.onSubscribeListener)
  OnFilterUnSubscribeEvent.dropListener(
    self.brokerCtx, self.onUnsubscribeListener)
    

EventBroker Guarantees

RequestBroker at a Glance

-- Declare once:
RequestBroker:        # or RequestBroker(sync):
  type MyData = object
    value: string
  proc signature*():
    Future[Result[MyData, string]]

-- Provider side (e.g. in WakuNode):
MyData.setProvider(
  proc(): Future[...] {.async.} =
    ok(MyData(value: "hello")))

-- Requester side (anywhere):
let data = await MyData.request()
        

Step 1 · Declare Your Requests

Sync variants — simple state accessors, no Future.

# Zero-arg sync: simple state accessor
RequestBroker(sync):
  type RequestConnectionStatus* = object
    connectionStatus*: ConnectionStatus

# Arg-based sync: parameterised lookup
RequestBroker(sync):
  type RequestContentTopicsHealth* = object
    contentTopicHealth*:
      seq[tuple[topic: ContentTopic,
                health: TopicHealth]]
  proc signature(topics: seq[ContentTopic]):
    Result[RequestContentTopicsHealth, string]
    

Step 1 · Declare Your Requests (async)

Async variants — for IO-bound operations.

# Zero-arg async: may involve IO
RequestBroker:
  type RequestHealthReport* = object
    healthReport*: HealthReport

# Arg-based async: query a specific protocol
RequestBroker:
  type RequestProtocolHealth* = object
    healthStatus*: ProtocolHealth
  proc signature(protocol: WakuProtocol):
    Future[Result[RequestProtocolHealth, string]]
    

Step 2 · Register a Provider (sync)

Provider registers at startup — requesters don't need to know who provides.

## Sync provider — zero-arg, no IO involved
RequestConnectionStatus.setProvider(
  threadGlobalBrokerContext(),
  proc(): Result[RequestConnectionStatus, string] =
    try:
      let report = waku[].healthMonitor
                         .getSyncNodeHealthReport()
      ok(RequestConnectionStatus(
        connectionStatus: report.connectionStatus))
    except:
      err("Failed: " & getCurrentExceptionMsg()),
).isOkOr:
  error "Failed to set provider"
    

Step 2 · Register a Provider (async)

## Async provider — may perform IO
RequestHealthReport.setProvider(
  threadGlobalBrokerContext(),
  proc(): Future[
      Result[RequestHealthReport, string]]
      {.async.} =
    try:
      let report = await waku[].healthMonitor
                               .getNodeHealthReport()
      ok(RequestHealthReport(healthReport: report))
    except:
      err("Failed: " & getCurrentExceptionMsg()),
).isOkOr:
  error "Failed to set provider"
    

Step 2 · Register a Provider (arg-based)

## Arg-based sync provider
RequestShardTopicsHealth.setProvider(
  node.brokerCtx,
  proc(topics: seq[PubsubTopic]):
      Result[RequestShardTopicsHealth, string] =
    var response: RequestShardTopicsHealth

    for shard in topics:
      var healthStatus = TopicHealth.UNHEALTHY

      if not node.wakuRelay.isNil:
        healthStatus = node.wakuRelay.topicsHealth
          .getOrDefault(shard, TopicHealth.NOT_SUBSCRIBED)

      response.topicHealth.add((shard, healthStatus))

    return ok(response),
).isOkOr:
  error "Can't set provider for RequestShardTopicsHealth",
    error = error
    

Step 3 · Make a Request

Any module can request without knowing who provides — or even if a provider is registered.

## Sync request — no await needed
let statusRes =
  RequestConnectionStatus.request()
statusRes.isOkOr:
  error "No connection status provider"
  return
echo statusRes.get().connectionStatus

## Sync with args
let healthRes =
  RequestShardTopicsHealth.request(topics)
healthRes.isOkOr:
  error "Shard health unavailable"; return

## Async request
let reportRes =
  await RequestHealthReport.request()
reportRes.isOkOr:
  error "Health report unavailable"; return
let report = reportRes.get().healthReport
        

MultiRequestBroker — Code

MultiRequestBroker:
  type CollectedStatus* = object
    label*: string

  proc signature*():
    Future[Result[CollectedStatus, string]]

# Register N providers:
let h1 = CollectedStatus.setProvider(
  proc(): Future[...]
      {.async: (raises: []).} =
    ok(CollectedStatus(label: "relay")))

let h2 = CollectedStatus.setProvider(
  proc(): Future[...]
      {.async: (raises: []).} =
    ok(CollectedStatus(label: "filter")))

# Request fans out — returns seq[T]:
let results = await CollectedStatus.request()
# Result[seq[CollectedStatus], string]
#   @["relay", "filter"]

# Remove individual provider via handle:
CollectedStatus.removeProvider(h1)
# or clear all:
CollectedStatus.clearProviders()
    

MultiRequestBroker — Use Case

• Collecting status/metrics from multiple independent subsystems
• Each subsystem registers itself — no central orchestrator
• All providers called concurrently via allFinished
• Succeeds only if all providers succeed — one failure short-circuits

SignalBroker at a Glance

SignalBroker — Code

SignalBroker:
  type IngestSample* = object
    deviceId*: string
    value*: float64

# One handler — a second onSignal errors:
discard IngestSample.onSignal(
  proc(s: IngestSample)
      {.async: (raises: []).} =
    echo s.deviceId, " = ", s.value)

# signal() is a PLAIN proc — never await it:
let r = IngestSample.signal(
  deviceId = "d1", value = 0.5)
# Result[void, string]

if r.isErr:
  echo "not delivered: ", r.error
    

SignalBroker — Accept, Don't Confirm

• ok() means accepted — a handler exists and the queue had room. It is never a promise the handler ran.
• err is definitive: "no signal handler installed" or "queue full" — bursty senders throttle on the latter.
• Handler exceptions are swallowed (a chronicles warn) — there is no reply path.
• EventBroker's fire-and-forget dispatch fused with RequestBroker's single registry — minus the response.

Why We Needed BrokerContext

How it works

• Every broker stores listeners in buckets, keyed by BrokerContext
• Bucket 0 is always DefaultBrokerContext — used when no context is specified
• Each component instance calls NewBrokerContext() and stores as field
• Context is propagated to submodules at init time

BrokerContext Implementation

type BrokerContext* = distinct uint32

const DefaultBrokerContext* =
  BrokerContext(0xCAFFE14E'u32)

proc NewBrokerContext*(): BrokerContext =
  # Atomically increment global counter
  BrokerContext(
    gContextCounter.fetchAdd(1, moRelaxed))

# In component initialization:
type WakuNode* = ref object
  brokerCtx: BrokerContext  # unique per instance

proc new*(T: type WakuNode, ...): T =
  WakuNode(
    brokerCtx: NewBrokerContext(), ...)

# Thread-global context binding (sync, usable from {.thread.} init):
proc setThreadBrokerContext*(ctx: BrokerContext)
  ## Adopts ctx as this thread's global context
proc initThreadBrokerContext*(): BrokerContext
  ## Creates new ctx + sets as thread-global
proc threadGlobalBrokerContext*(): BrokerContext
  ## Returns this thread's global context (lock-free)

# Async scoped context (backward compat):
template lockGlobalBrokerContext*(
    brokerCtx: BrokerContext,
    body: untyped) =
  await noCancel(
    globalBrokerContextLock.acquire())
  globalBrokerContextValue = brokerCtx
  try: body
  finally: restore + release
    

Context Isolation — Code

Two nodes, same thread

let nodeA = WakuNode.new(...)  # ctx = 0x0001
let nodeB = WakuNode.new(...)  # ctx = 0x0002

let recvA = RecvService.new(nodeA, ...)
let recvB = RecvService.new(nodeB, ...)

# recvA registers under nodeA's context:
OnFilterSubscribeEvent.listen(
  nodeA.brokerCtx,
  proc(ev: ...) {.async.} =
    recvA.onSubscribe(...))

# recvB registers under nodeB's context:
OnFilterSubscribeEvent.listen(
  nodeB.brokerCtx,
  proc(ev: ...) {.async.} =
    recvB.onSubscribe(...))

# Emit from nodeA — only recvA fires:
OnFilterSubscribeEvent.emit(
  nodeA.brokerCtx,
  pubsubTopic, contentTopics)
    

Context Isolation — Bucket Structure

Why Go Multi-Thread?

Use cases that push you across thread boundaries

• CPU-bound work — heavy cryptography, proof generation, or tree hashing that would block the main event loop
• Blocking FFI — C libraries (e.g. database drivers, hardware access) that don’t expose async APIs
• Thread-per-service architecture — each node service owns a thread + Chronos loop; brokers bridge them
• Integration tests — multiple node instances in one process, each on its own thread, sharing typed events

Enabling Multi-Thread Mode

Both EventBroker and RequestBroker accept an (mt) driver annotation. The public API stays almost identical — the macro swaps the internal plumbing.

Cross-Thread Delivery

Costs of Multi-Thread Mode

Benefits of Multi-Thread Mode

Design Considerations

RequestBroker (mt) — Specifics

• Cross-thread request — args go into a request slab cell + ring; reply comes back via a reserved response slot. No allocation on the hot path.
• Provider thread must run a Chronos loop — the shared dispatch loop drains the ring and calls the local provider closure
• Same-thread fast path — provider is invoked directly with zero ring/slab overhead
• Failure modes — err on payload too large, ring full, response-slot pool exhausted, or timeout (default 5 s, configurable via T.setRequestTimeout)
• Capacity knobs — queueDepth, slabCapacity, responseSlots, payload/response size caps; or pick a named preset

EventBroker (mt) — Specifics

• Per-listener-thread ring + slab — one shared ring + slab per (ctx, listener-thread); all listeners on that thread share it
• Single-cell fan-out — one slab cell per emit, refcounted across listeners — no per-listener deep-copy
• Listener concurrency — within a thread, listeners for an event run concurrently (not serialised)
• Async emit — emit is {.async: (raises: []).}; same-thread dispatches via asyncSpawn, cross-thread via the ring + signal
• Ordering — per-target-thread FIFO; cross-thread ordering is per-target, not global
• Capacity knobs — queueDepth, slabCapacity, maxPayloadBytes, plus named presets

Capacity, Footprint & Perf

EventBroker(mt, preset = fastBurst, maxPayloadBytes = 1024):
  type Tick = object
    seq*: uint64
    payload*: seq[byte]

RequestBroker(mt, queueDepth = 1024,
                  responseSlots = 512,
                  maxResponseBytes = 32768):
  type Query = object
    key*: string
        

The Concept

What the Broker FFI-API Adds

Threading Model of a Created Library Context

Author the Library Interface in Nim

when defined(BrokerFfiApi):
  import brokers/api_library

RequestBroker(API):
  type InitializeRequest = object
    configPath*: string
    initialized*: bool
  proc signature*(configPath: string):
    Future[Result[InitializeRequest, string]] {.async.}

RequestBroker(API):
  type ShutdownRequest = object
    status*: int32
  proc signature*():
    Future[Result[ShutdownRequest, string]] {.async.}

EventBroker(API):
  type DeviceDiscovered = object
    deviceId*: int64
    name*: string
    deviceType*: string
    address*: string

when defined(BrokerFfiApi):
  registerBrokerLibrary:
    name: "mylib"
    initializeRequest: InitializeRequest
    shutdownRequest: ShutdownRequest
        

Consume It from C++ Through the Generated Wrapper

Mylib lib;
auto created = lib.createContext();
if (!created.ok()) {
  fprintf(stderr, "FATAL: %s\n", created.error().c_str());
  return 1;
}

auto h_disc = lib.onDeviceDiscovered(
    [](Mylib& owner, int64_t id, const std::string_view name,
     const std::string_view type,
     const std::string_view addr) {
      (void)owner;
    printf("discovered: %.*s\n",
           (int)name.size(), name.data());
  });

auto res = lib.initializeRequest("/opt/devices.yaml");
  std::vector fleet = {
    AddDeviceSpec{"Core-Router", "router", "10.0.0.1"}
  };
  auto add = lib.addDevice(fleet);
auto list = lib.listDevices();

lib.offDeviceDiscovered(h_disc);
lib.shutdown();
        

One Fixed CBOR ABI · One Wrapper Surface

Layered Architecture

One Nim source → one fixed CBOR ABI → wrappers expose one idiomatic surface per language.

Build Flags That Enable the Feature

nim c \
  -d:BrokerFfiApi \        # enables the CBOR FFI codegen
  -d:BrokerFfiApiGenPy \    # opt-in: Python ctypes wrapper
  -d:BrokerFfiApiGenRust \  # opt-in: Rust Cargo crate
  -d:BrokerFfiApiGenGo \    # opt-in: Go cgo module
  --threads:on \
  --app:lib \
  --nimMainPrefix:mylib \
  --mm:orc \
  --outdir:examples/ffiapi/nimlib/build \
  examples/ffiapi/nimlib/mylib.nim
        

Summary

A simple way to navigate the broker toolbox: