This venue's subscription lifecycle — internal, and the place where its two transports are joined.

Subscribing here is two operations on two protocols

Market data arrives over MQTT; subscriptions are HTTP calls. Each MQTT session is joined to its own HTTP subscriptions by one value: the session_id a shard generates, registers as its MQTT client id, and then names in every HTTP subscribe for that shard.

A consumer calls subscribe/3 with symbols. It never learns that a socket was dialled, how many, that an HTTP call followed, or that transport and subscription had to agree on an identifier — which is the facade doing precisely what D12 asks of it, on the venue where it costs the most to deliver.

Re-subscribing after a reconnect is this package's job

If the connection is dropped due to network issues, previous subscriptions are not automatically restored. You must re-subscribe after reconnecting.

So a reconnect is followed by a replay of everything wanted on that shard. A consumer that had to notice reconnects and replay its own subscriptions would be doing the venue's bookkeeping through an interface designed to hide reconnects entirely.

Coverage is observed, never intended

A symbol enters the coverage map when a payload for it arrives — not when it is subscribed, and not when the HTTP subscribe returns 200. On this venue those are three genuinely different moments, and only the last one means data.

Coverage by kind, because this venue's streamed kinds really are independent

Every subscribe asks the venue for SNAPSHOT, QUOTE and TICK (see Subscription), and the three arrive on separate MQTT topics decoded by Socket into three different structs: snapshot becomes Core.Types.Quote (kind :quotes, a traded price), quote becomes Core.Types.TopOfBook (kind :top_of_book, bid/ask), tick becomes Core.Types.Trade (kind :trades, one print). coverage/1 folds all three into one :stream per symbol, so a symbol whose snapshot topic goes dark while its quote topic keeps arriving — or any other combination — is invisible there; coverage_by_kind/1 exists to split exactly that apart, one kind map per struct type actually observed. kind_for/1 derives the kind from the struct that arrived rather than assuming it, so a future new kind reaching this clause without a matching case here is caught (logged, loudly) instead of silently folded into an existing kind.

Sharded — one session tops out at 100 symbols, this package's scope does not

A single MQTT session caps out at the venue's own stated ceiling — "Maximum number of subscribe tickers:100", confirmed against a real collection run that hit it (dp-exchange-core issue #13). A consumer with more than 100 symbols on this venue could not reach full coverage through one session no matter how the HTTP calls were split, because the limit is per-session, not per-request.

This is not new ground for the family — DpExchange.Coinbase.Feed shards for the identical reason, and its shape (recompute from the full wanted set, touch only what changed, one shard synchronous per call and the rest staggered) is the template this adapts. What differs is the leaf operation: Coinbase subscribes channels on an always-usable socket; this venue's shard identity is also its MQTT session, so a brand-new shard has to wait for its own CONNACK (see below) before its first HTTP subscribe means anything to the venue.

@pairs_per_socket is exactly the venue's own stated 100 — not a guessed margin below it. See dp_exchange_core's docs/design/closed/2026-09-04_webull-sharding-and-fake-injection.md §3.1 for why padding a number the venue already stated would be exactly the unlabeled guess this family's own conventions rule out.

Five connections per App Key is the hard ceiling this can never exceed — a consumer cannot cause a sixth socket, because a consumer cannot ask for sockets at all. Five shards of 100 is 500 symbols; a universe larger than that on this venue needs a second App Key, not a bigger number here.

A shard that rejects a batch is this package's problem to solve, not the host's

If a shard's HTTP subscribe comes back TOO_MANY_SYMBOLS_SUBSCRIPTION despite this package's own accounting — a bug, a race, or the venue's real ceiling turning out lower in practice than its own stated one — the affected symbols are moved to another shard with room (opening one if needed, within the five-connection ceiling) and retried internally. The host is never handed a session_id or a shard index to reason about; it only ever sees whether its symbols ended up covered. Only running out of shards entirely — five sessions full and the venue still refuses — is a genuine capacity ceiling this package cannot paper over, and that surfaces as a real refusal.

A socket process is not a connected socket

Socket.start_link/1 returns once the WebSocket is up; MQTT is not authenticated until the venue answers the CONNECT with a CONNACK, which arrives later as a :link_up notice carrying the shard's session_id. The first subscribe against a fresh shard does not call the HTTP endpoint itself — it waits for that notice, exactly as a reconnect already did, so the session id it names is one the venue has actually registered.

A subscribed, connected session can still go quiet on its own

Not a reconnect, not an error, not an unsubscribe — the venue simply stops pushing to an otherwise-healthy session shortly after each subscribe, with nothing on the wire to say so. dp_crypto_management's own pre-existing MQTT client found this the hard way, empirically: a blind, unconditional resubscribe on a timer, independent of whether the wanted set had changed, took its live coverage from 47 symbols back to ~240 (DpCryptoManagement's issue #17). reshard/4 alone cannot recover from this — it only touches a shard whose wanted symbol set changed, and re-asking for exactly what is already wanted computes an empty diff and asserts nothing.

So every connected shard's current subscription is re-issued unconditionally every @resubscribe_interval_ms, regardless of whether anything is believed to have changed — the same shape Coinbase's Feed already carries for its own reconnect case, applied here to a steady-state failure mode Coinbase does not have.

The resubscribe timer must never fail-fast

A moduledoc worth carrying from dp_exchange_robinhood's Feed, which named this exact shape first (acquire, not check — DpCryptoManagement's issue #16). This package reproduced it independently, live, at a worse scale: DpCryptoManagement's issue #23 — a node restart, all 4 MQTT shards linking up cleanly, then 58 consecutive blind-resubscribe failures across 13 minutes, every one the identical refusal:

{:exchange_error, :webull, "Throttled by our own rate limiter (not the venue) —
 retry after 1s; callers that can wait should set rate_limit_blocking: true"}

Core.HttpClient's own message names the fix. The refusal asks for a one-second wait; @resubscribe_interval_ms is 60,000. Fail-fast (check/3) on this timer means the request is dropped for a whole minute to avoid waiting a second — a self-inflicted outage the venue never asked for. Measured consumer impact: 0 of 342 pairs streaming, every one of them falling back to REST polling, for as long as the rate limiter stayed contended.

It compounds with the venue's own transient INVALID_SYMBOL rejections on an initial subscribe (ordinarily self-healing, since the very next resubscribe tick re-asks for the same symbol) — because recovery from that runs through this same blind resubscribe, a recoverable error became permanent for exactly as long as the timer could never issue a request at all.

Documenting that design was not the same as wiring it — again. :rate_limit_blocking — the option Core.HttpClient.check_rate_limits/1 reads to choose acquire/3 over check/3 — was missing from every allowlist on the path a blind resubscribe actually takes: this module's own resubscribe_opts (built once in init/1), replayable/2 (what carries it forward across every later subscribe), and Subscription.request_opts/1 (the last allowlist before Core.HttpClient itself). Fixing only the two in this module and leaving Subscription's allowlist untouched would have shipped a change that reads as a fix and does nothing: the option would still be stripped one layer down, silently, with every test that stops at "the keyword list contains :rate_limit_blocking" passing regardless.

All three now forward it. Only Feed's own opts — resubscribe_opts at init/1, and replayable/2 on every call after — default it to true: the resubscribe timer runs off a 60-second clock with no caller waiting on its result, so blocking for as long as a second is free. Subscription.request_opts/1 forwards the option without defaulting it, on purpose — a caller invoking Subscription.subscribe/3 directly, one-off, may legitimately want fail-fast, and this module must not decide that for it.

A generic resubscribe failure is reported too, latched per shard

The blind resubscribe timer already had two callers-visible outcomes when the venue's answer was structured: :oversubscribed rebalances silently (see handle_subscribe_result/3 — it is a capacity measurement, not a failure), and {:error, {:invalid_symbols, symbols}} gets its own :refusal notice (see "A venue-rejected symbol is excluded, timed, and reported" above). Everything else an {:error, reason} could be — the rate-limiter throttling of DpCryptoManagement's issue #23, an HTTP 5xx, a transport error — fell through handle_subscribe_result/3's catch-all clause to a Logger.warning and nothing else: a real, ongoing failure with no facade-level signal a consumer could act on.

Issue #23 is the concrete incident this closes visibility on: a node restart, all 4 shards linking up cleanly, then 58 consecutive blind-resubscribe failures across 13 minutes, every one the identical rate-limiter refusal — discovered only by grepping this module's own log for the sentence it had been repeating the whole time. PollingFeed's own :on_notice (DpCryptoManagement's issue #21, the poll-feed sibling of this same gap) is the pattern this follows: a Core.Notice{kind: :coverage_change} fires the instant a shard's blind resubscribe crosses INTO this generic failure, severity: :info fires the instant it crosses back OUT, and neither fires again while the shard's own state stays put.

Latched per shard, not globally — state.resubscribe_failed, a MapSet of shard indices currently in this state — because each shard is its own independent MQTT session with its own independent failure and recovery schedule. A global latch would either swallow a second shard's own transition while the first stayed latched, or (unlatched entirely) fire a fresh notice from up to five shards every single @resubscribe_interval_ms during a widespread outage — a notice storm being its own defect, exactly as issue #21's design established. The existing Logger.warning above keeps firing every tick regardless, unchanged — this notice is additive, not a replacement for it.

A shard's socket crash is contained to that shard

Socket.start_link/1 links to Feed — ordinary WebSockex.start_link/4 behaviour — so an uncaught exception inside a socket's own callbacks, or any other abnormal exit, propagates as a linked EXIT. Feed traps exits for exactly this reason: without it, one shard's crash killed every shard's connection and every symbol's coverage, not just the one that failed — the opposite of the isolation reshard/4, resync/1 and the resubscribe timer above all work to provide. A crashed shard's socket is reopened at the same index with the same wanted symbols; the venue issues a fresh session, and the ordinary CONNACK-then-resubscribe path brings it back exactly as a first open would. A caller with a reply pending on the crashed shard is answered {:error, {:shard_crashed, reason}} rather than left to time out.

Control-plane HTTP never runs on the mailbox that also carries ticks

Every Subscription.subscribe/unsubscribe call this module makes — reconciling a shard's diff, replaying a reconnect, or the unconditional 60s resubscribe above — runs in its own short-lived, supervised task rather than inline inside a handle_call or handle_info. A caller's subscribe/3 still does not get its reply until the real HTTP round trip finishes (the same observable contract as before, kept via a deferred GenServer.reply/2), but the Feed process itself stays free to keep draining incoming ticks — from this shard and every other one — while that round trip is in flight. Before this, a single blind resubscribe tick chained up to five sequential ~118ms HTTP calls inside one message, stalling delivery for every shard, every 60 seconds, by design.

A venue-rejected symbol is excluded, timed, and reported — DpCryptoManagement's issue #24

Subscription's INVALID_SYMBOL handling (see its own moduledoc) hands this module {:error, {:invalid_symbols, [canonical_symbol, ...]}} instead of an opaque string. The reason this module, not Subscription, has to be the one to act on it: rejection here is per-request, so one symbol the venue's streaming category does not carry fails the entire shard's batch — measured live as 17 of one consumer's 342 symbols, named byte-for-byte identically every 60-second resubscribe tick, taking stream_covered to 0/342 and every pair to REST polling (the sustained 429 storm DpCryptoManagement's issue #23 first surfaced).

Handled the same way :oversubscribed already is — see handle_subscribe_result/3 and the retry branch in reshard_step/4 — because the shape is the same: a structured venue answer this package can act on automatically rather than a caller-visible failure. The named symbols are recorded in state.rejected with an expiry and excluded from plan_reshard/1's effective wanted set (active_rejections/1), so the next chunk built for that shard carries only the symbols the venue actually accepts, and every other shard is untouched.

state.wanted is never pruned. Only what plan_reshard/1 treats as wantable right now shrinks — a rejected symbol stays in wanted for exactly the reason coverage/1 is observed rather than intended: the host asked for it, and whether the venue currently carries it is a separate, time-bound fact. active_rejections/1 returning it to eligibility the moment its entry expires is what lets it flow straight back into a shard on the very next reshard-triggering event — the same 60-second resubscribe tick that discovered the rejection, once resync/1 runs off it again — with nobody calling update_symbols/2.

@rejected_symbol_ttl_ms defaults to 24 hours, deliberately matching DpCryptoManagement.Data.Collection.VenueRefusals' own TTL for exactly this shape of fact: a venue's streaming catalogue is true at a point in time, not permanently, and a symbol it refuses today can be listed later. Picked to be the same order of magnitude as that consumer-side cache rather than independently guessed — two different TTLs for the same underlying fact would mean the two layers disagree about how stale a "the venue refuses this" belief is allowed to get. Overridable per call via opts[:rejected_symbol_ttl_ms] for a consumer with a documented reason to want a different number.

Reported, not just filtered. A filtered symbol that only ever disappears from shard composition is coverage silently shrinking — this module's data stream never reports it either way (see "Coverage is observed, never intended" above: it was never delivering, so it was never in coverage/1 to begin with), so the only way a consumer learns 17 of its 342 symbols stopped being tried is a Core.Notice. Emitted as :refusal — Core's own documented kind for exactly this ("a symbol the venue will not carry"), not a package-invented one — naming the rejected symbols in canonical form, the same reason Subscription converts them before this module ever sees them: a notice is public, gets pasted into issues, and must never carry a venue-native string a consumer has no mapping for.

Shard assignment is sticky, not recomputed from scratch

A symbol already assigned to a shard keeps that shard for as long as it stays wanted, even as other symbols are added or removed. Deriving shards by sorting the entire wanted set and cutting it into fixed-size chunks — the original approach — meant one newly-added symbol that happened to sort early could shift every symbol after it across every shard boundary, so an unrelated add unsubscribed and resubscribed symbols that were already healthy. derive_shards/3 instead starts from what each shard already carries, drops only what stopped being wanted or no longer fits that shard's measured capacity, and places everything else — new symbols, and anything just evicted by a capacity reduction — into whichever shard (in index order) still has room. reshard/4's touched computation is what actually limits HTTP calls to changed shards; this is what makes that set small in the first place.