mule/telemetry

Telemetry for mule — a pure-Gleam port of Oban.Telemetry.

Where Oban builds on the BEAM :telemetry library (atom-keyed measurements/metadata maps, dispatched through an ETS handler table), this is a self-contained, typed equivalent: events are a Gleam sum type (Event), so a handler pattern-matches on the variant it cares about and the payload is statically known — no Dynamic, no map-key typos. The event taxonomy, measurements, and metadata mirror Oban’s so the mapping back to the upstream docs is one-to-one; the delivery mechanism is ours.

Dispatch model

Handlers are attached by id to a set of EventNames and invoked synchronously in the emitting process (as :telemetry does), so a handler must be quick and must not crash the caller. The registry lives in persistent_term (the same VM-global cell the leader/Sonar state uses): reads are on the hot path (every emit), writes (attach/detach) are rare and happen at startup. When no handler is attached the lookup returns an empty dict, so emitting is close to free; once handlers exist the small handler dict is copied per emit (the cost :telemetry avoids with ETS) — fine for the handful of handlers a typical app attaches, and the trade for zero new deps. attach is a read-modify-write on persistent_term, so configure handlers at boot, not concurrently from many processes.

Divergences from Oban (deliberate, given the typed model)

Metadata carries the instance name (instance), not the whole conf record — conf is a bundle of closures that can’t be meaningfully inspected or logged, and the name is the part handlers actually key on.
Measurements are in Erlang’s native time unit (nanosecond on standard BEAM), like Oban; the default logger converts to microseconds on output.
args/meta are Dynamic on the read side (decoded only inside a worker’s perform), so the default logger omits them rather than guessing an encoding.

Types

Connectivity

</>

Notifier connectivity, mirroring Oban’s Sonar status (the notifier:switch metadata). Defined here rather than imported from runtime/sonar because Sonar emits these events, and telemetry must not depend on its emitters.

pub type Connectivity {
  Isolated
  Solitary
  Clustered
}

Constructors

```
Isolated
```
```
Solitary
```
```
Clustered
```

EngineOperation

</>

The engine operation a [:mule, :engine, *] span describes.

pub type EngineOperation {
  AllJobs
  CancelAllJobs
  CancelJob
  CheckAvailable
  CheckMeta
  CompleteJob
  DeleteAllJobs
  DeleteJob
  DiscardJob
  ErrorJob
  FetchJobs
  GetJob
  InitQueue
  InsertAllJobs
  InsertJob
  PruneJobs
  PutMeta
  RefreshMeta
  RescueJobs
  RetryAllJobs
  RetryJob
  Reindex
  ShutdownMeta
  SnoozeJob
  StageJobs
  StageScheduled
  UpdateJob
  ElectLeader
  ReadLeader
  DeleteLeader
}

Constructors

```
AllJobs
```
```
CancelAllJobs
```
```
CancelJob
```
```
CheckAvailable
```
```
CheckMeta
```
```
CompleteJob
```
```
DeleteAllJobs
```
```
DeleteJob
```
```
DiscardJob
```
```
ErrorJob
```
```
FetchJobs
```
```
GetJob
```
```
InitQueue
```
```
InsertAllJobs
```
```
InsertJob
```
```
PruneJobs
```
```
PutMeta
```
```
RefreshMeta
```
```
RescueJobs
```
```
RetryAllJobs
```
```
RetryJob
```
```
Reindex
```
```
ShutdownMeta
```
```
SnoozeJob
```
```
StageJobs
```
```
StageScheduled
```
```
UpdateJob
```
```
ElectLeader
```
```
ReadLeader
```
```
DeleteLeader
```

Event

</>

Every telemetry event, mirroring Oban’s [:mule, category, type] taxonomy. instance is the Mule instance name throughout.

pub type Event {
  JobStart(instance: String, job: job.Job, system_time: Int)
  JobStop(
    instance: String,
    job: job.Job,
    state: JobOutcome,
    measurements: Measurements,
  )
  JobException(
    instance: String,
    job: job.Job,
    state: JobOutcome,
    kind: exceptions.Kind,
    error: exceptions.ExecutionError,
    stacktrace: List(String),
    measurements: Measurements,
  )
  EngineStart(
    instance: String,
    operation: EngineOperation,
    system_time: Int,
  )
  EngineStop(
    instance: String,
    operation: EngineOperation,
    duration: Int,
  )
  EngineException(
    instance: String,
    operation: EngineOperation,
    reason: String,
    duration: Int,
  )
  NotifierNotifyStart(
    instance: String,
    channel: notifier.Channel,
    system_time: Int,
  )
  NotifierNotifyStop(
    instance: String,
    channel: notifier.Channel,
    duration: Int,
  )
  NotifierNotifyException(
    instance: String,
    channel: notifier.Channel,
    reason: String,
    duration: Int,
  )
  NotifierSwitch(instance: String, status: Connectivity)
  PeerElectionStart(
    instance: String,
    node: String,
    leader: Bool,
    system_time: Int,
  )
  PeerElectionStop(
    instance: String,
    node: String,
    leader: Bool,
    was_leader: Bool,
    duration: Int,
  )
  PeerElectionException(
    instance: String,
    node: String,
    reason: String,
    duration: Int,
  )
  PluginInit(instance: String, plugin: String)
  PluginStart(instance: String, plugin: String, system_time: Int)
  PluginStop(
    instance: String,
    plugin: String,
    duration: Int,
    result: PluginResult,
  )
  PluginException(
    instance: String,
    plugin: String,
    reason: String,
    duration: Int,
  )
  ProducerStart(
    instance: String,
    queue: String,
    system_time: Int,
  )
  ProducerStop(
    instance: String,
    queue: String,
    dispatched_count: Int,
    duration: Int,
  )
  ProducerException(
    instance: String,
    queue: String,
    reason: String,
    duration: Int,
  )
  QueueShutdown(
    instance: String,
    queue: String,
    orphaned: List(Int),
    elapsed: Int,
  )
  StagerSwitch(instance: String, mode: StagerMode)
  SupervisorInit(instance: String, system_time: Int)
}

Constructors

JobStart(instance: String, job: job.Job, system_time: Int)

JobStop(
  instance: String,
  job: job.Job,
  state: JobOutcome,
  measurements: Measurements,
)

JobException(
  instance: String,
  job: job.Job,
  state: JobOutcome,
  kind: exceptions.Kind,
  error: exceptions.ExecutionError,
  stacktrace: List(String),
  measurements: Measurements,
)

EngineStart(
  instance: String,
  operation: EngineOperation,
  system_time: Int,
)

EngineStop(
  instance: String,
  operation: EngineOperation,
  duration: Int,
)

EngineException(
  instance: String,
  operation: EngineOperation,
  reason: String,
  duration: Int,
)

NotifierNotifyStart(
  instance: String,
  channel: notifier.Channel,
  system_time: Int,
)

NotifierNotifyStop(
  instance: String,
  channel: notifier.Channel,
  duration: Int,
)

NotifierNotifyException(
  instance: String,
  channel: notifier.Channel,
  reason: String,
  duration: Int,
)

NotifierSwitch(instance: String, status: Connectivity)

PeerElectionStart(
  instance: String,
  node: String,
  leader: Bool,
  system_time: Int,
)

PeerElectionStop(
  instance: String,
  node: String,
  leader: Bool,
  was_leader: Bool,
  duration: Int,
)

PeerElectionException(
  instance: String,
  node: String,
  reason: String,
  duration: Int,
)

PluginInit(instance: String, plugin: String)

PluginStart(instance: String, plugin: String, system_time: Int)

PluginStop(
  instance: String,
  plugin: String,
  duration: Int,
  result: PluginResult,
)

PluginException(
  instance: String,
  plugin: String,
  reason: String,
  duration: Int,
)

ProducerStart(instance: String, queue: String, system_time: Int)

ProducerStop(
  instance: String,
  queue: String,
  dispatched_count: Int,
  duration: Int,
)

ProducerException(
  instance: String,
  queue: String,
  reason: String,
  duration: Int,
)

QueueShutdown(
  instance: String,
  queue: String,
  orphaned: List(Int),
  elapsed: Int,
)

StagerSwitch(instance: String, mode: StagerMode)

SupervisorInit(instance: String, system_time: Int)

EventName

</>

The coarse identity of an event, used to attach a handler to a subset of events (the typed analog of :telemetry’s event-name lists).

pub type EventName {
  JobStartName
  JobStopName
  JobExceptionName
  EngineStartName
  EngineStopName
  EngineExceptionName
  NotifierNotifyStartName
  NotifierNotifyStopName
  NotifierNotifyExceptionName
  NotifierSwitchName
  PeerElectionStartName
  PeerElectionStopName
  PeerElectionExceptionName
  PluginInitName
  PluginStartName
  PluginStopName
  PluginExceptionName
  ProducerStartName
  ProducerStopName
  ProducerExceptionName
  QueueShutdownName
  StagerSwitchName
  SupervisorInitName
}

Constructors

```
JobStartName
```
```
JobStopName
```
```
JobExceptionName
```
```
EngineStartName
```
```
EngineStopName
```
```
EngineExceptionName
```
```
NotifierNotifyStartName
```
```
NotifierNotifyStopName
```
```
NotifierNotifyExceptionName
```
```
NotifierSwitchName
```
```
PeerElectionStartName
```
```
PeerElectionStopName
```
```
PeerElectionExceptionName
```
```
PluginInitName
```
```
PluginStartName
```
```
PluginStopName
```
```
PluginExceptionName
```
```
ProducerStartName
```
```
ProducerStopName
```
```
ProducerExceptionName
```
```
QueueShutdownName
```
```
StagerSwitchName
```
```
SupervisorInitName
```

JobOutcome

</>

The terminal outcome reported in a job :stop/:exception event, mirroring Oban’s :state metadata. Success, Cancelled, Snoozed, and a manual Discard ride a JobStop; Failure rides a JobException, as does an exhausted failure — reported under Discard (executor.ex emit_event remaps :exhausted to :discard).

pub type JobOutcome {
  Success
  Failure
  Cancelled
  Discard
  Snoozed
}

Constructors

```
Success
```
```
Failure
```
```
Cancelled
```
```
Discard
```
```
Snoozed
```

Measurements

</>

The four measurements attached to a job :stop/:exception, in native time units (memory in bytes, reductions a count). Mirrors Oban’s job measures.

pub type Measurements {
  Measurements(
    duration: Int,
    queue_time: Int,
    memory: Int,
    reductions: Int,
  )
}

Constructors

Measurements(
  duration: Int,
  queue_time: Int,
  memory: Int,
  reductions: Int,
)

PluginResult

</>

What a plugin accomplished in the tick a PluginStop reports — the typed analog of the per-plugin metadata Oban merges into the [:mule, :plugin, :stop] event. Defined here (not in the plugins) for the same reason as Connectivity: telemetry must not depend on its emitters. The AffectedJob lists carry the id/queue/state projection, matching the reference’s “jobs only include id, queue, state fields” note.

pub type PluginResult {
  NoPluginResult
  PrunerResult(pruned_count: Int)
  LifelineResult(
    rescued_jobs: List(engine.AffectedJob),
    discarded_jobs: List(engine.AffectedJob),
  )
  CronResult(jobs: List(job.Job))
  StagerResult(staged_queues: List(String))
}

Constructors

```
NoPluginResult
```
```
PrunerResult(pruned_count: Int)
```

LifelineResult(
  rescued_jobs: List(engine.AffectedJob),
  discarded_jobs: List(engine.AffectedJob),
)

```
CronResult(jobs: List(job.Job))
```

StagerResult(staged_queues: List(String))

StagerMode

</>

Stager mode, mirroring Oban’s stager:switch metadata. Defined here for the same reason as Connectivity.

pub type StagerMode {
  Local
  Global
}

Constructors

```
Local
```
```
Global
```

Values

attach

</>

pub fn attach(
  id id: String,
  events events: List(EventName),
  handler handler: fn(Event) -> Nil,
) -> Result(Nil, Nil)

Attach handler under id for the given event names. The handler runs synchronously in the emitting process on each matching emit, so keep it fast and non-crashing. Returns Error(Nil) when id is already attached, mirroring :telemetry.attach’s duplicate rejection.

attach_default_logger

</>

pub fn attach_default_logger() -> Result(Nil, Nil)

Attach a structured-JSON logging handler that mirrors Oban.Telemetry.attach_default_logger: one JSON line per event (with a source: "mule" field), routed through Erlang’s logger at info. Covers the same event set Oban’s logger does. Returns Error(Nil) if already attached.

default_logger_id

</>

pub const default_logger_id: String

The id the default logger attaches under (exposed for detaching / testing).

detach

</>

pub fn detach(id id: String) -> Nil

Detach the handler attached under id. A no-op when nothing is attached.

detach_default_logger

</>

pub fn detach_default_logger() -> Nil

Detach the default logger. A no-op if it was never attached.

emit

</>

pub fn emit(event: Event) -> Nil

Emit an event to every handler attached to its EventName. Called by the runtime; also useful from tests.

Each handler runs synchronously in this (the emitting) process, guarded by a catch-all: a handler that raises is detached and logged, never propagated — so a faulty third-party handler can’t crash the executor’s job process, the stager, or any other emitter. Mirrors :telemetry’s isolation of the caller from handler failures.

engine_error_to_string

</>

pub fn engine_error_to_string(
  error: engine.EngineError,
) -> String

A human-readable reason string for an EngineError, used in :engine/:peer exception events.

instrument_engine

</>

pub fn instrument_engine(
  engine engine: engine.Engine,
  instance instance: String,
) -> engine.Engine

Wrap an Engine so every operation emits an [:mule, :engine, <op>] span (:start before, :stop on Ok, :exception on Error). mule.start applies this before building Config, so all callers go through it.

instrument_notifier

</>

pub fn instrument_notifier(
  notifier base: notifier.Notifier,
  instance instance: String,
) -> notifier.Notifier

Wrap a Notifier so each notify emits an [:mule, :notifier, :notify] span. Only notify is instrumented (as in Oban); listen/unlisten are left untouched. notify cannot fail in this port, so no :exception is emitted.

job_measurements

</>

pub fn job_measurements(
  start_native start_native: Int,
  job job: job.Job,
) -> Measurements

Build the job :stop/:exception measurements for an execution that began at start_native (an erlang:monotonic_time/0 reading). memory and reductions are read from the calling process, so call this from the executor’s own process at the end of the run.

monotonic

</>

pub fn monotonic() -> Int

name_of

</>

pub fn name_of(event: Event) -> EventName

native_to_microsecond

</>

pub fn native_to_microsecond(value: Int) -> Int

system_time

</>

pub fn system_time() -> Int