Mob.Defect.Bus (mob v0.8.3)

Copy Markdown View Source

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

class_count()

@spec class_count() :: non_neg_integer()

How many distinct defect classes are held.

classes(limit \\ 20)

@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(capsule)

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(limit \\ 20)

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

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

subscribe(pid \\ self())

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

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.

subscribers()

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

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

unsubscribe(pid \\ self())

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

Unsubscribe pid (defaults to self()).