# `DpExchange.Core.Fanout` [🔗](https://github.com/DistortionPoint/dp-exchange-core/blob/v0.3.120/lib/dp_exchange/core/fanout.ex#L1) Pushes a venue's events to its subscribers, and refuses to grow their mailboxes without a bound. ## Why this module exists `DpExchange.Core.Venue`'s `c:DpExchange.Core.Venue.subscribe/2` doc has said this since the contract was written: > A venue pushing faster than its subscriber consumes drops beyond a stated bound and > emits a `:degraded` notice saying so. Growing a mailbox silently until the node dies is > the failure this avoids; dropping silently is the failure the notice avoids. It was a guarantee no package provided. All five venues fanned out with a bare `send/2` in five identical private `fan_out/2` functions, none of which had ever looked at a subscriber's mailbox. A consumer reading the contract was told back-pressure was handled and declared; neither half was true. That is the failure this family names most often — a statement that stays entirely plausible while only its meaning is wrong — sitting in the contract itself rather than in a venue. The exposure is not theoretical. `dp_exchange_coinbase`'s `level2` channel measured 4258 delta frames in the window that produced this family's coverage incident. A subscriber that stalls for thirty seconds against a stream like that accumulates a mailbox in the hundred-thousands, and the node dies with no notice, no log line and nothing in `coverage/1` to suggest anything was wrong: the feed was delivering perfectly the whole time. ## Newest is dropped, not oldest — and the contract used to say otherwise The wording above used to read *"drops oldest beyond a stated bound"*. **A sender cannot drop the oldest message in another process's mailbox.** Nothing in the BEAM lets one process remove a message another process has already been sent; the receiver owns its own queue. Written that way, the sentence described something no implementation could ever have honoured, which is part of why nothing implemented it. What a sender *can* do is decline to add to a queue that is already past its bound, which is what this module does. So the contract now states the achievable guarantee. Dropping the newest is also the better trade for this data: a quote or a book delta that arrives while a consumer is thirty thousand messages behind is worthless by the time it would be read, and the frames it would push out are no fresher. ## Checking is cheaper than the send it guards Measured before it was written, on the machine this family develops on: Process.info/2 : 0.029 us/call (empty mailbox) Process.info/2 : 0.028 us/call (100k backlog) send/2 : 0.097 us/call `:message_queue_len` is a counter the process already maintains, so reading it is O(1) and does not walk the queue — the 100k-backlog figure is the one that proves it. At 0.3x the cost of the `send/2` it replaces or permits, there is no case for sampling the check, latching it for N messages, or any of the other complications that would have been reasonable if it had turned out to cost more than the send. It runs on every message to every subscriber. ## One notice per transition, never one per dropped message A stalled subscriber that produced a notice per dropped message would emit them at the rate of the stream it cannot keep up with — into the same fan-out that is already overloaded, and to a notice subscriber that may be the very process that is behind. The cure would be worse than the disease. So the caller carries the set of subscribers currently being dropped, and `deliver/4` returns it back along with only the *transitions*: `:dropping` the first time a subscriber is found over its bound, `:resumed` the first time it has drained below `resume_below/1`, half the bound. Not merely back under it: a consumer hovering at the bound would otherwise flip on every message, and each flip is a notice into its full mailbox (measured 2026-10-04, 5_700 notices in five minutes). Both are worth a notice — a consumer needs to know when data loss started, and it needs to know when it stopped, because those two instants bracket exactly what it has to reconcile from a pull endpoint. The set is rebuilt from what was observed on each call rather than edited in place, so a subscriber that dies or unsubscribes while over its bound leaves it without needing to be pruned. ## Notices are never dropped `deliver/4` is for a venue's data stream. A `Core.Notice` goes out through the caller's own unbounded path, and must: the notice announcing that a subscriber is being dropped cannot be the first casualty of that same subscriber being dropped. Notices are low-volume by construction — link transitions, coverage changes, refusals — and no venue in this family has ever produced them at a rate that could bury a consumer. # `monitors` ```elixir @type monitors() :: %{required(pid()) => reference()} ``` Monitor references for subscribers that are raw pids, keyed by pid. Carried in a feed's own state. Only pids appear here — see `watch/2` for why a registered name deliberately does not. # `subscriber` ```elixir @type subscriber() :: pid() | atom() ``` A subscriber as `subscribe/2` accepts one: a pid, or a registered name to resolve at send time. # `transition` ```elixir @type transition() :: {pid(), :dropping | :resumed, non_neg_integer()} ``` What changed for one subscriber on this call, and what its queue measured. # `default_max_queue_len` ```elixir @spec default_max_queue_len() :: pos_integer() ``` The default bound: 10000 messages queued for one subscriber. Chosen as the largest number that is still unambiguously a fault rather than a busy moment. A consumer keeping up with a book stream sits in the single or double digits; one that has reached five figures is not behind, it is broken, and every message after that point is being written to memory nobody will read in time to use. Overridable per venue — see `deliver/4`'s `:max_queue_len`. A consumer with a documented reason to buffer more can say so; the point is that the number exists and is stated, not that this particular one is right for everybody. # `deliver` ```elixir @spec deliver(Enumerable.t(), term(), MapSet.t(pid()), keyword()) :: {non_neg_integer(), MapSet.t(pid()), [transition()]} ``` Sends `message` to every subscriber whose mailbox is under the bound. `dropping` is the set of pids that were over the bound on the previous call; pass `MapSet.new()` the first time. Returns `{message_sent_count, new_dropping, transitions}` — the caller stores `new_dropping` and reports `transitions`, which is what `notice_for/3` turns into a `Core.Notice`. ## Options * `:max_queue_len` — the bound, defaulting to `default_max_queue_len/0` (also when given as `nil`). Any other value that is not a positive integer raises `ArgumentError`: a string or an `:infinity` compares as larger than every queue and would switch back-pressure off without a word. A subscriber that has died, or a registered name that resolves to nothing, is skipped silently: that is not back-pressure, it is an absent consumer, and every venue in this family already treats it that way. # `forget` ```elixir @spec forget(pid(), monitors()) :: monitors() ``` Stops monitoring `pid` and drops it from `monitors`. Call it from a `:DOWN` handler and from `unsubscribe/2`. Both matter: on `:DOWN` the monitor is already spent and this only cleans the map, while on an explicit unsubscribe the monitor is still live and would otherwise deliver a `:DOWN` for a subscriber the feed has already forgotten — harmless in itself, but it leaves a reference the feed has no record of, which is the same bookkeeping drift one level down. `flush: true` on the demonitor drops any `:DOWN` already in the mailbox, so a pid that dies in the same instant it unsubscribes cannot leave a message nothing will match. # `max_queue_len!` ```elixir @spec max_queue_len!(keyword(), atom()) :: pos_integer() ``` Reads and validates a venue's `:max_queue_len` start option, or returns the default. Shared rather than written per venue so five packages cannot drift into five different ideas of what a valid bound is — and so the failure is identical everywhere: loud, at `init/1`, naming the venue and the value it was given. A bound that silently fell back to the default because it was `"10000"` rather than `10_000` would be a back-pressure setting a consumer believes it configured and did not, which is the whole class of defect this module exists inside. Any positive integer is accepted, including very small ones. A bound of 1 — drop anything arriving while the consumer has even one message pending — is a strange choice but a coherent one for a latency-critical consumer that would rather have the newest frame than a queue, and this is not the place to overrule it. # `notice_for` ```elixir @spec notice_for(transition(), atom(), pos_integer()) :: DpExchange.Core.Notice.t() ``` Turns one `t:transition/0` into the `Core.Notice` the contract promises. `:dropping` is `severity: :warning` rather than `:error`: the venue is healthy and every other subscriber is being served: it is this one consumer that has fallen behind, and the action is on the consumer's side. `:resumed` is `:info` and exists so a consumer can bracket the gap it needs to reconcile — without it, "data loss started" is an alarm with no end, and a consumer cannot tell a five-second stall from an ongoing outage. The bound and the measured queue length both travel in `:details`, because "we dropped something" without the number is a notice a consumer cannot act on. # `resolve` ```elixir @spec resolve(subscriber()) :: pid() | nil ``` Resolves a subscriber to a live pid, or `nil`. A pid that is no longer alive and a registered name nothing answers to are the same answer — there is nobody to send to. Shared here because all five venues had written it identically, and because `deliver/4` and a venue's own notice path have to agree on what counts as a reachable subscriber or a notice could be delivered to a pid the data stream considers gone. # `resume_below` ```elixir @spec resume_below(pos_integer()) :: pos_integer() ``` The queue length a subscriber being dropped must drain below before it is sent to again: half the bound, and at least 1. See `deliver/4`. Resuming one message under the bound made a consumer hovering there flip on every message, each flip a notice into its full mailbox. # `watch` ```elixir @spec watch(subscriber(), monitors()) :: monitors() ``` Monitors `subscriber` if it is a pid and is not already monitored, so a feed can drop it when it dies. ## The leak this closes `resolve/1` skips a dead subscriber at send time, so a dead consumer never accumulates *events* — which is what `Core.Venue`'s `unsubscribe/2` doc asks for, and it was true. What accumulated was the **pid**. Nothing ever removed one from a feed's subscriber set, so a supervised consumer that restarts leaves its old pid behind on every restart, for the life of the feed. That is not a rounding error on the hot path, because `deliver/4` walks the whole set once per message and asks the runtime about every entry. Re-measured on the machine this family develops on, against the delivery path as it ships today — cost of one fan-out against a set holding one live subscriber plus N dead ones, median of eleven runs of 30_000: 0 dead 0.68 us 50 dead 2.05 us 200 dead 5.90 us 1000 dead 28.56 us Linear, and at a thousand accumulated pids each message costs roughly forty times what it should. `dp_exchange_coinbase`'s `level2` channel measured 4258 frames in the window that produced this family's coverage incident; at that size the dead entries alone would be about 119 ms of runtime calls inside the one process every subscriber's data flows through. And it only ever grows. An earlier version of this table read 0.095 / 0.956 / 4.301 / 22.842 and was taken when this path asked `Process.alive?/1` per entry. It no longer does — see `delivery_pid/1` — and a dead entry is now slightly DEARER rather than cheaper, which sharpens this argument rather than softening it. ## Why a registered name is NOT monitored A pid that has died is gone permanently, so removing it is always right. **A name is not a process.** `subscribe/2` accepts a registered name precisely so a consumer can restart under it — that is the ordinary OTP arrangement the name form exists for — and a monitor on a name fires when the *current holder* dies, not when the name is abandoned. Pruning on that `:DOWN` would silently unsubscribe a consumer whose supervisor is about to bring it straight back under the same name, which is a worse failure than the leak: it is data loss with nothing to notice it by. A name also cannot leak. The set holds one atom however many times the process behind it restarts, and `resolve/1` answers `nil` for a name nothing is registered under. The unbounded growth is entirely the pid case, which is exactly the case this covers. --- *Consult [api-reference.md](api-reference.md) for complete listing*