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

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 — caller sends request via channel, blocks (or awaits) until the provider thread responds via a separate response channel
• Round-trip latency — two channel hops (request + response), both using AsyncChannel; expect ~2× the cost of a single emit
• Provider thread must run a Chronos loop — processLoop receives requests and dispatches to the local provider closure
• Same-thread fast path — when requester and provider share a thread, the provider is called directly with zero channel overhead
• Timeout handling — cross-thread requests use withTimeout (default 5 s); if the provider is unresponsive, request returns err. Configure via T.setRequestTimeout(duration) / T.requestTimeout(). Same-thread requests are unaffected.

EventBroker (mt) — Specifics

• Per-listener-thread channel — one AsyncChannel per (context, listener-thread) pair; multiple listeners on the same thread share one channel
• In-flight tracking — processLoop tracks running listener futures and drains them on shutdown (5 s timeout)
• Listener concurrency — within a thread, all listeners for an event run concurrently (not serialised)
• Async emit — emit is {.async: (raises: []).}; cross-thread sends use sendSync (brief: buffer + signal), same-thread dispatches via asyncSpawn
• Ordering — per-thread FIFO guaranteed by the channel; cross-thread ordering is per-target-thread, not global

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();
        

Build Flags That Enable the Feature

nim c \
  -d:BrokerFfiApi \
  --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:

Opportunities!

A natural extension point beyond the core broker runtime: