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.
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. The bus's owner GenServer holds the
subscriber registry and monitors, and it publishes the cached subscriber
pid 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.
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 — the monitor in the owner catches the DOWN and 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.
Summary
Functions
How many distinct defect classes are held.
Every defect class currently held, newest first by last_seen_at.
Emit a capsule.
The most recent capsules (raw occurrences), newest first.
Subscribe the calling process to defect emits.
The subscribers the write path will fan out to right now.
Unsubscribe pid (defaults to self()).
Functions
@spec class_count() :: non_neg_integer()
How many distinct defect classes are held.
@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.
@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.
@spec recent(pos_integer()) :: [Mob.Defect.Capsule.t()]
The most recent capsules (raw occurrences), newest first.
Subscribe the calling process 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's process is monitored; a subscriber exit prunes it from the cached list.
@spec subscribers() :: [pid()]
The subscribers the write path will fan out to right now.
@spec unsubscribe(pid()) :: :ok
Unsubscribe pid (defaults to self()).