# `Mob.Defect.Bus`
[🔗](https://github.com/genericjam/mob/blob/master/lib/mob/defect/bus.ex#L1)

The bounded record of defects the framework has emitted, and the fanout to
whoever subscribed to hear about them.

## Deduplication

Every emit goes through `fingerprint`. One thousand emits of the same bug
produce one class row whose occurrence counter reaches 1000 as each one
lands, with the first-seen capsule held for the class and `last_seen_ms`
moving forward. That is what makes the channel usable — a report system
that emitted a thousand rows per bug would train its readers to stop
reading it.

The counter and last-seen fields are read directly from the row on each
call to `classes/1`; both are updated by atomic ETS ops on the write path
(`update_counter` and `update_element`), so a listing is consistent under
concurrency down to the individual row.

A parallel bounded ring keeps the most recent 64 raw capsules, keyed by
sequence, for the case where a triager wants to walk through occurrences
rather than classes. Ring rather than unbounded because a capsule is packaged
memory in a production app, and defect-report storage that grows without
limit is worse than a bug it might have described.

Classes are bounded too, at 256. A fingerprint is only as stable as the
`fingerprint_key` its caller supplied, and app code supplies some, so a key
that carries per-occurrence data would otherwise add a class per emit for
the life of the app. Past the bound the least recently seen class is evicted
and counted (`Mob.Diag.health/0`, `class_evictions`).

## Fanout without a mailbox on the hot path

Same reasoning as `Mob.Agent.Receipts`: an `emit/1` is on the path of every
detected defect, and putting a GenServer in front of that path serialises
every writer through one mailbox. Subscribers are kept by
`Mob.Diag.Subscribers`, which monitors them and publishes the list to
`:persistent_term`; the write path reads that once and sends to each pid
directly. A subscriber lifecycle change is a rare event; a defect emit is not.

## Observation

Post-mortems that the OS hands over only once (`Mob.PostMortem.Journal`)
stay on disk until someone has seen the specific capsule. The bus answers
that per capsule, never by sequence range: `emit_delivered/1` tells the
journal's sweep how many subscribers a capsule was handed to, and
`recent/1` passes the capsules it actually returns to
`Mob.PostMortem.Journal.observed/1`, which is one `:persistent_term` read
unless post-mortems are waiting. `classes/1` and `subscribe/1` observe
nothing: a class row does not show the specific occurrence, and a new
subscriber has seen nothing yet. `emit/1` itself does no extra work.

## No default sink

A subscriber is a pid, and no pid is subscribed until an app registers one.
Per the decision record, mob owns the format and the bus; it never owns a
destination. The dev sink in `Mob.Defect.Sinks.Dev` is what a connected
agent runs at its end after `mix mob.connect`.

## Subscribers must not crash the emit path

A subscriber pid is a `send/2` target on the emit path. `send/2` never
blocks and never raises on a dead pid, so a dead subscriber does not take
down the emitter — its monitor prunes the pid from the cached list. But the
*contents* of what a subscriber does with the message must not affect the
emitter, which is the standard contract of message passing and not enforced
here.

`emit/1` itself never raises: a failure to record is counted as `lost` in
`Mob.Diag.health/0`, because a defect reporter that crashes on the defect it
is reporting destroys the report.

# `class_count`

```elixir
@spec class_count() :: non_neg_integer()
```

How many distinct defect classes are held.

# `classes`

```elixir
@spec classes(pos_integer()) :: [map()]
```

Every defect class currently held, newest first by `last_seen_at`.

A class row carries the *first* capsule seen for that fingerprint plus the
occurrence count and last-seen timestamp. Later occurrences are on the
recent ring, not layered onto the class — that keeps the class row a
bounded shape regardless of how noisy the defect gets.

# `emit`

```elixir
@spec emit(Mob.Defect.Capsule.t()) :: Mob.Defect.Capsule.t()
```

Emit a capsule.

Records the class (incrementing occurrences), appends to the recent-ring,
and fans out to every subscribed pid as `{:mob_defect, capsule}`.

Returns the capsule, so this can sit at the end of a pipeline.

# `recent`

```elixir
@spec recent(pos_integer()) :: [Mob.Defect.Capsule.t()]
```

The most recent capsules (raw occurrences), newest first.

Returning a capsule counts as observing it (see "Observation" above).

# `subscribe`

```elixir
@spec subscribe(pid()) :: {:ok, reference()}
```

Subscribe `pid` (defaults to the caller) to defect emits.

Returns `{:ok, ref}` — the caller can keep the ref for its own bookkeeping,
but does not need it to unsubscribe (unsubscription is by pid).
Idempotent: subscribing an already-subscribed pid is a no-op.

The subscriber is monitored; its exit prunes it. A subscriber on another node
whose connection drops stops receiving and resumes when the node reconnects,
without re-subscribing (see `Mob.Diag.Subscribers`). From a connected node,
pass the pid to receive on — `:rpc.call(node, Mob.Defect.Bus, :subscribe,
[self()])`. Without it, `:rpc` subscribes the short-lived process it runs the
call in, which receives nothing.

# `subscribers`

```elixir
@spec subscribers() :: [pid()]
```

The subscribers the write path will fan out to right now.

# `unsubscribe`

```elixir
@spec unsubscribe(pid()) :: :ok
```

Unsubscribe `pid` (defaults to `self()`).

---

*Consult [api-reference.md](api-reference.md) for complete listing*