This venue's subscription lifecycle — internal. The facade's subscribe/2,
unsubscribe/2, update_symbols/2, coverage/1 and subscribe_notices/1 are served
from here.
Coverage is observed, never intended
A symbol enters the coverage map when a payload for it arrives, never when it is subscribed. That exists because a venue once reported 325 symbols subscribed and confirmed while 174 were delivering; reporting the subscription would have said 325.
A subscribed symbol that has delivered nothing is simply absent, which the facade
documents as :not_covered.
Coverage by kind is tracked from the struct that arrived, never from the channel
delivering_by_kind buckets the same observed-arrival fact coverage/1 reports, split
by which of Core.Types.Quote or Core.Types.TopOfBook a message actually was;
coverage/1's own map is now derived from it, so the two cannot drift apart. The kind
comes from a pattern match on the struct itself — never from the wanted set, a
channel address, or anything this module ever asked for — because intent standing in
for evidence is the exact failure coverage_by_kind/1 exists to close. See
DpExchange.Gemini.coverage_by_kind/1 for the incident this closes and why it applies
to a venue with one physical stream underneath two data kinds.
A payload of a struct type this module does not recognise is still fanned out to
subscribers — this module is not the place to decide a struct is uninteresting — but
it is not counted toward delivering_by_kind, and so not toward coverage/1 either.
Counting an unrecognised kind as generic "something arrived" would be the same collapse
one level removed: a symbol would read as covered without this module being able to say
what for.
Sharding: the host shards this venue nine ways, and this package does not
gemini/feed.ex in the host opens nine sockets, ten pairs each, because the endpoint
it uses stopped accepting subscriptions past roughly that count. That is a real measured
property — of the old endpoint.
This package speaks wss://ws.gemini.com, where subscription is a single subscribe
frame carrying a list of streams, and there is no documented per-connection limit. So
there is no shard arithmetic here, and that is a claim with a hole in it worth
naming: it has been exercised with a handful of symbols, not with 346. If a limit exists
on the new endpoint it will be found by the first consumer to subscribe broadly, and the
fix belongs here, in this package, where the measurement will live.
Carrying the host's nine-way split across on the assumption that the new endpoint shares the old one's limit would be worse — nine connections where one may do, justified by a measurement of a different API.
A reconnect that does not resubscribe is a coverage collapse with no error
WebSockex reconnects a dropped socket on its own — Socket.handle_disconnect/2 returns
{:reconnect, state} — and a bare reconnect leaves it connected and subscribed to
nothing, silently: a socket that is up and receiving nothing is not itself an error.
Socket's own state carries no memory of what was subscribed (%{subscriber:, request_id:}), so there was nothing to resend even if it tried, and this module's
wanted MapSet — written on every subscribe/3 — was never read by anything. The
sequence a consumer would actually see is :link_down then :link_up, which reads as
"recovered", followed by silence until someone notices a quiet chart. This is the same
incident class dp_exchange_coinbase's Feed moduledoc records under the same heading;
the fix here is the same idea adapted to one socket instead of a shard set.
This module now re-issues the current wanted set's subscription on a timer,
unconditionally — not only after a detected reconnect, because a reconnect a consumer's
process never learns about (Socket reconnecting on its own, under WebSockex's own
handling of a dropped connection, for instance) is indistinguishable from one it does.
Re-subscribing a stream the socket already carries costs one frame the venue ignores;
not re-subscribing one it silently dropped costs this package's whole coverage until
someone notices.
A crashed socket is Feed's crash too, unless Feed catches it — and now it does
ensure_socket/1 calls Socket.start_link/1 from inside Feed's own callback, which
links the new socket to Feed the way start_link always does — this is not a
supervised sibling Feed could lose independently, it is a linked child. Before this
fix Feed never called Process.flag(:trap_exit, true), so a socket that exited
abnormally (an exception inside a WebSockex callback, or anything that killed the
socket pid directly) sent an untrappable EXIT signal along that link and crashed
Feed too — every subscriber, the whole wanted set, gone, restarted by
DpExchange.Gemini.Supervisor from the STATIC opts it was given at tree-start, which
never carry a consumer's later subscribe/3 calls. Proven by linking a real process
into a running Feed the way ensure_socket/1 does and killing it with
Process.exit(pid, :kill) — :normal would not have proven anything, since a
non-trapping process ignores a peer's normal exit.
Feed now traps exits, and a crashed socket is handled the same way a reconnect it
never even noticed would be: state.socket is cleared, delivering_by_kind is reset
(this venue has one socket carrying both streamable kinds, so a crash costs both, not a
partial set the way a per-shard venue's would), a :link_down Core.Notice reports it,
and resubscribe/1 — the same function the periodic timer already calls — attempts an
immediate reconnect and resend of wanted rather than waiting out the next
@resubscribe_interval_ms tick.
The resubscribe timer's own failure path was silent — until now
The section above fixed a silent coverage collapse by making this module re-issue its
subscription unconditionally. Its own failure path repeated the exact shape it was built
to close: before this fix, grep -n "Notice.new(" lib/dp_exchange/gemini/feed.ex matched
nothing in this file at all — a resubscribe that kept failing every 60 seconds only ever
reached a Logger.warning, and this is a file whose own moduledoc, one section up, exists
because a log line is not something a consumer can subscribe to.
DpExchange.Core.PollingFeed's moduledoc records the sibling discovery for the poll-feed
case — DpCryptoManagement's issue #21, where a feed answered nothing for hours while its
own "delivered NOTHING" log line sat ungrepped — and its fix is the shape this module now
borrows: a notice_state: :ok | :dead latch (record_success/2 and
notify_delivering_nothing/3), firing a Core.Notice on the transition INTO failure and
a recovery notice on the transition back OUT, never once per tick for as long as an
outage lasts. dp_exchange_coinbase's Feed established the sibling case for THIS
family — a channel subscribe that exhausted its retries without ever becoming delivery —
using Core's :coverage_change kind for exactly that shape of fact: subscribed intent
that did not become delivery. A periodic resubscribe that keeps failing is the same shape
one level up: the intent is "keep what was already subscribed, subscribed", and it is not
becoming delivery either.
Coinbase's own notice there is one-shot — it fires once when retries are exhausted and
carries no recovery counterpart, because that retry chain either succeeds silently or
exhausts and is left for the next unconditional cycle to revisit. This module's
resubscribe runs forever on a fixed timer rather than a bounded retry chain, so the
stricter, PollingFeed-shaped latch applies here instead: resubscribe_notice_state
tracks whether the last resubscribe attempt succeeded, a :warning notice fires exactly
once on the first failure after a success (or after boot), and an :info recovery notice
fires exactly once on the first success after a failure. The Logger.warning above is
unchanged and still fires on every failing tick — that remains the correct "loud while it
lasts" log behaviour; only a consumer-visible Notice is new, and it is deliberately
quieter than the log beside it.
ensure_socket/1 connects inside handle_call — its timeout budget is chosen, not inherited
ensure_socket/1 calls Socket.start_link/1 synchronously inside handle_call, and
Feed/SandboxFeed are named, shared processes: the whole blocking window that connect
can take is borne by every other consumer's subscribe/3, unsubscribe/2 and
coverage/1 queued behind it, not only the caller that happened to trigger it — the
Feed process itself is stuck, so the caller's own @call_timeout cannot help any of
them.
The connect was never unbounded — that was the wrong diagnosis. websockex's own
WebSockex.Conn bounds it already: measured from the vendored dependency,
@socket_connect_timeout_default is 6_000ms and @socket_recv_timeout_default is
5_000ms (deps/websockex/lib/websockex/conn.ex:10-11), and both are read from the
opts list passed to WebSockex.start_link/4 (conn.ex:98-100). The real defect was
that Socket.start_link/1 passed no opts at all, so it inherited those defaults by
accident rather than choosing them — and 6_000 + 5_000 = 11_000ms of connect, plus one
send_frame for the subscribe that follows connecting (up to @frame_window_ms,
5_000ms), is 16_000ms against this module's own 15_000ms @call_timeout: already over
budget before any other overhead in that call is counted.
Socket.start_link/1 now sets :socket_connect_timeout and :socket_recv_timeout
explicitly, chosen against that same budget rather than left to websockex's general-
purpose defaults: 3_000ms connect + 2_000ms recv + 5_000ms for the one frame send that
follows is 10_000ms, leaving 5_000ms — a third of @call_timeout — for the GenServer
call's own overhead. Both remain overridable through the opts Socket.start_link/1
already accepts, for a deployment whose real connect time needs more room than this
package's own budget assumed — Feed.start_link/1's own opts forward
:socket_connect_timeout and :socket_recv_timeout straight through, alongside :url
and :environment.
Summary
Functions
Returns a specification to start this module under a supervisor.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec coverage(GenServer.server()) :: %{ required(String.t()) => :stream | :internal_poll | :not_covered }
@spec coverage_by_kind(GenServer.server()) :: %{ required(DpExchange.Core.Capabilities.data_kind()) => %{ required(String.t()) => :stream | :internal_poll | :not_covered } }
@spec start_link(keyword()) :: GenServer.on_start()
@spec subscribe(GenServer.server(), [String.t()], keyword()) :: :ok | {:error, term()}
@spec subscribe_notices( GenServer.server(), keyword() ) :: :ok
@spec unsubscribe(GenServer.server(), [String.t()]) :: :ok | {:error, term()}
@spec update_symbols(GenServer.server(), [String.t()]) :: :ok | {:error, term()}