sinal/forwarder

A bounded forwarder: a process that runs the handlers of the events handed to it, so a slow or blocked handler stalls the forwarder instead of the emitter.

Isolating a library’s events

A library emits with sinal.emit. The application decides, per event-name prefix, whether those events run synchronously (the default) or in a forwarder. It supervises one forwarder and routes the library’s prefix to it, once, at start:

import gleam/erlang/process
import gleam/otp/static_supervisor as supervisor
import sinal/forwarder

pub fn start(name: process.Name(forwarder.Message)) {
  let observations = forwarder.new(name)
  let assert Ok(_) =
    supervisor.new(supervisor.OneForOne)
    |> supervisor.add(forwarder.supervised(observations))
    |> supervisor.start
  forwarder.route(["http_gun"], observations)
}

Every sinal.emit of an event whose name starts with http_gun then returns as soon as the event is handed off. A package that owns its forwarder (one per database, say) calls forwarder.emit instead, which returns the Refusal when the event is dropped.

Operational limits

• Best-effort delivery. A send beyond capacity (1,024 events by default, queued plus executing) is refused with CapacityExceeded and counted as rejected. A send to a forwarder that is not running (not yet started, or mid-restart) is refused with ForwarderUnavailable and counted as unavailable. Neither blocks or retries, and a routed event never falls back to inline delivery.
• Per-producer FIFO, no global order. A single producer’s forwarded events run in the order it sent them. Interleaving across producers is unspecified, as it already is for native telemetry handler order.
• The forwarder is the handler’s self(). Handlers run inside the forwarder process. Process-dictionary context on the producer is not carried across the hop, and a native span cannot be forwarded (its start and stop must share one process to measure duration); spans and this module’s own drop report ignore routes.
• Shutdown does not drain. Pending events are lost when the process stops. A restart takes a best-effort snapshot of the in-flight count and reports it as lost. This count is not a delivery receipt.
• Capacity belongs to one incarnation. Each running process owns a one-row named ETS table pairing its event subject with its capacity and fresh admission counters, published after initialisation. Before that, producers receive ForwarderUnavailable. A delayed producer keeps the old subject and counters; it cannot enqueue into the replacement. The table’s name is the forwarder’s process name in the separate ETS namespace; a pre-existing ETS table of that name, or a capacity below 1, makes the start fail with InitFailed.
• One set of counters per name. The drop counters live in persistent_term under the forwarder’s process name, so every Forwarder value built from one process.Name shares them. Create the name once at application start, never per request: each name keeps its counters for the life of the node.
• A handler’s exit, not just its raise, can take the forwarder down. Native telemetry isolates a handler that raises. A handler that links to another process and receives an exit signal, or is sent kill, can still stop the forwarder.
• Concurrent emitters may see a spurious, safe rejection near the boundary. Admission is a single atomic increment-then-check: it never admits past capacity, but under concurrent load it can reject a send that would have fit under a different ordering.
• Routed ordering holds per route. A producer’s events that resolve to the same forwarder keep its send order, and so do its unrouted ones. Events split across routes, or across a route change, have no relative order.
• Routes are node-global, set-up-time values. They live in persistent_term: an emit reads them without locking, and a node with no routes pays one lookup per sinal.emit. Changing a route is expensive and belongs at application start and shutdown.

When drops are reported

The three counts survive every incarnation. Only a running incarnation reports them, as dropped_event from its own process, and each report resets exactly the counts it carries:

• When an incarnation starts, it drains all three counts and reports them, if any is nonzero. Unavailable drops made while it was down arrive here.
• After a drop while it runs, the next drain reports rejected and unavailable together (lost is zero outside a start). Concurrent drops in one drain fold into one report.
• If no incarnation starts again, the counts are never reported. The ForwarderUnavailable refusal is the only signal meanwhile.
• A report can be lost with its incarnation, if it is stopped before handling the report it queued at start.
• Reports never loop. A drop report is emitted directly, never through a route, so reporting a drop never causes one.

pub type Dropped {
  Dropped(rejected: Int, lost: Int, unavailable: Int)
}

pub type DroppedMetadata {
  DroppedMetadata(forwarder: String)
}

pub opaque type Forwarder

pub opaque type Message

pub type Refusal {
  CapacityExceeded
  ForwarderUnavailable
}

pub const default_capacity: Int

pub fn describe_refusal(refusal: Refusal) -> String

pub fn dropped_event() -> sinal.Event(Dropped, DroppedMetadata)

pub fn emit(
  forwarder: Forwarder,
  event: sinal.Event(m, d),
  measurements: m,
  metadata: d,
) -> Result(Nil, Refusal)

pub fn new(name: process.Name(Message)) -> Forwarder

pub fn route(prefix: List(String), forwarder: Forwarder) -> Nil

pub fn supervised(
  forwarder: Forwarder,
) -> supervision.ChildSpecification(Forwarder)

pub fn unroute(prefix: List(String)) -> Nil

pub fn with_capacity(
  forwarder: Forwarder,
  capacity: Int,
) -> Forwarder