DpExchange.Core.PollingFeed (DpExchangeCore v0.3.10)

Copy Markdown View Source

A feed built from repeated fetches, for a venue package to use INSIDE its own feed.

Why this lives in the contract and not in a consumer

A venue with no streaming API still has to present a feed, or every consumer above it forks on transport — which is the drift the facade exists to undo. Robinhood has no socket and never will; its feed is this module, and nothing above the package can tell the difference.

Venues WITH sockets use it too, for the symbols their stream does not reach. That gap-filling decision belongs to the venue because only the venue can make it correctly: Webull's stream is bounded by "3 messages per second per connection", and which symbols lose that race is knowable only by watching its own sockets. Shared code had to guess, and guessed that 151 delivering-nothing symbols were a quiet market.

Shipping it in the contract rather than in a consumer means a venue package can build a feed without reaching for anything outside its own dependencies — which is the property that lets a venue be a standalone package at all.

Bulk where the venue offers it, per-symbol where it does not

A venue with an overview endpoint fetches its whole set in one request (:fetch_all). Robinhood's 87 pairs cost one call per cycle that way and 87 without, and its rate limiter is not hypothetical — dropping to per-symbol fetches would multiply this venue's request count by the size of its catalog.

Where only a per-symbol endpoint exists (:fetch), every symbol runs on its own schedule rather than the set being swept in a burst. A burst is what a rate limiter sees as an attack, and it makes the first symbol in the list permanently fresher than the last, so start times are spread across the interval to keep the request rate flat.

A fetch that fails does not stop the feed

A symbol whose fetch fails is retried on the next tick and reported as :not_covered until one succeeds. It is never dropped: this module cannot tell a delisted symbol from a network blip, and the layer that CAN — the venue answering with an explicit refusal — handles that separately.

A feed delivering NOTHING says so, loudly

Individual failures are debug-level, because a handful of them are ordinary. A feed where nothing at all is succeeding is not ordinary, and it is the most expensive failure shape in collection: it is indistinguishable from a quiet venue, which is how one venue's feed ran through its first deployment publishing nothing at all: it had been handed a credential whose key was still ciphertext. Credentials are accepted as an opaque map, so a wrong one is not an error at the boundary — it is a fetch that fails, and the per-symbol failures were invisible at debug level.

So a feed that has delivered nothing since it started escalates: it warns once it has failed a full cycle with zero successes, and keeps warning while that holds. Never a silent retry loop.

The warning above is a log line, and a log nobody greps in time is exactly how DpCryptoManagement's issue #21 stayed hidden — the feed said "has delivered NOTHING in 154 consecutive attempts" to its own log, and a human found that sentence by grepping, not because anything downstream reacted to it. So this module ALSO emits an on_notice callback — the same injected-function shape as on_refusal — carrying a Core.Notice{kind: :coverage_change} the instant it crosses INTO the delivering-nothing state, and a recovery notice the instant it crosses back out. Once per transition, not once per failed tick and not once per sweep thereafter: a 342-symbol feed retrying every symbol every cycle would otherwise turn one outage into a notice storm. on_notice defaults to a no-op, so a caller that does not wire it up gets a working feed, not a crash — the same contract on_refusal already keeps.

A fetch that never returns fails just as loudly — and never blocks a health check

Two hazards live here. They look like one and needed two separate fixes.

The hang. Nothing in this module's own code can time out a caller's fetch, so that has to be a boundary this module imposes. Without one, a single hung fetch (a socket that never closes, an HTTP client with no timeout of its own) blocks this GenServer forever. So every fetch runs inside a disposable task and a hang past :fetch_timeout_ms becomes an ordinary fetch failure — retried next tick, counted toward failures_since_ok, escalated the same way a real error would be. The default is derived from the poll interval and clamped between @min_fetch_timeout_ms and @max_fetch_timeout_ms, so this module never trades an unbounded hang for a merely very long one, and never turns a fast interval into an accidental timeout on an ordinary, slower-than-a-tick HTTP round trip.

Waiting on that task was the second hazard, and it is the one that bit. This module used to run the fetch in a task and then BLOCK on it inside handle_info. The task bounded the hang; it did nothing at all for the mailbox. With @min_fetch_timeout_ms at 30 seconds and GenServer.call/2's default timeout at five, a coverage/1 or status/1 arriving during an ordinary in-flight fetch was not unlucky — it was a guaranteed timeout. In dp_exchange_robinhood that exit propagated out of the venue Feed's own handle_call and killed it (dp-exchange-core issue #28): a read-only health check terminated the thing it was checking, the feed restarted without the subscription state its static start opts never carried, and coverage went 61 pairs to 0 and stayed there — while the process sat alive and idle, so every liveness probe kept passing.

So the fetch result now arrives as a message. handle_info starts the task and returns immediately, and coverage/1 and status/1 answer from state at any point during a fetch. A health check can no longer be what breaks the thing it observes.

Concurrency is deliberately still one. A tick arriving while a fetch is in flight is queued, not started — which is exactly what the mailbox did when the fetch was synchronous. Letting ticks overlap would quietly multiply a venue's request rate the moment a fetch grew slower than its interval, and an unexamined multiplier on request volume is a defect class this family has paid for more than once. Rescheduling still happens only after a job finishes, so there is at most one pending job per symbol and the queue cannot grow without bound.

Summary

Types

Fetches one symbol's current price.

Fetches many symbols in one call, for a venue whose upstream API answers a batch as cheaply as one symbol.

Receives a Core.Notice.t() the instant this feed crosses into or back out of the delivering-nothing state. Injected like sink and on_refusal, so the feed never reaches outside its own inputs to publish one.

Functions

Returns a specification to start this module under a supervisor.

Which symbols this poller is currently delivering.

Whether this feed is delivering, and what went wrong if it is not.

Replace the symbol set without restarting the poller.

Types

fetch()

@type fetch() :: (String.t() -> {:ok, map()} | {:error, term()} | {:refused, term()})

Fetches one symbol's current price.

Three outcomes, and the third matters. {:ok, event} publishes. {:error, reason} is retried on the next tick — this module cannot tell a delisted symbol from a network blip, so it must not decide. {:refused, reason} is the venue stating it does not carry the symbol at all, which only the adapter can recognise, and which is reported once rather than retried forever.

fetch_all()

@type fetch_all() :: ([String.t()] ->
                  {:ok, [map()]}
                  | {:error, term()}
                  | {:refused, [{String.t(), term()}]})

Fetches many symbols in one call, for a venue whose upstream API answers a batch as cheaply as one symbol.

{:ok, events} publishes every event and is the only outcome most bulk fetchers ever need. {:error, reason} is retried next cycle, same as fetch's :error — this module still cannot tell a delisted symbol from a network blip and must not decide for one.

{:refused, refusals} is fetch's :refused scaled to a batch: refusals is a [{symbol, reason}] list naming every symbol in THIS call that the venue stated it does not carry at all, reported through on_refusal once each rather than retried forever. Added because its absence was a real gap: before it existed, a batch fetcher that found one bad symbol mixed into an otherwise-live request had no outcome to return that this module's fetch_all_and_publish/1 did not already handle, other than reporting the whole batch as an ordinary {:error, reason} — which, unlike a real :refused, is retried forever and never reaches on_refusal, so the caller never learns to drop the symbol from its scope. dp_exchange_robinhood documented this exact gap as the reason it stayed on per-symbol :fetch rather than adopt this endpoint's own documented repeatable-query bulk mode.

A bulk response with zero events ({:ok, []}) is treated as delivering nothing for escalation purposes even though the call itself succeeded — see fetch_all_and_publish/1.

notice_handler()

@type notice_handler() :: (DpExchange.Core.Notice.t() -> any())

Receives a Core.Notice.t() the instant this feed crosses into or back out of the delivering-nothing state. Injected like sink and on_refusal, so the feed never reaches outside its own inputs to publish one.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

coverage(server)

@spec coverage(pid() | atom()) :: %{required(String.t()) => :internal_poll}

Which symbols this poller is currently delivering.

OBSERVED: a symbol appears only once a fetch has actually succeeded for it and is still recent. A symbol that has been asked for and never answered is absent, because reporting it as covered would be the venue asserting a delivery that never happened.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

status(server)

@spec status(pid() | atom()) :: %{
  delivering: boolean(),
  symbols: non_neg_integer(),
  covered: non_neg_integer(),
  failures_since_ok: non_neg_integer(),
  last_error: term()
}

Whether this feed is delivering, and what went wrong if it is not.

Exposed as DATA and not only as a log line, because "delivering nothing" is the condition a health check has to be able to ask about. Robinhood's feed ran a whole deployment publishing nothing — a log nobody greps in time is how that stays true for hours.

update_symbols(server, symbols)

@spec update_symbols(pid() | atom(), [String.t()]) :: :ok

Replace the symbol set without restarting the poller.