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: TICK is in the default sub_types again since dp_exchange_webull issue #7, although the venue has so far published none for us-crypto). The topics arrive separately and Socket decodes each into its own struct: 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 every kind 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.

:trades is delivered but never counted here. capabilities().streamable does not declare :trades (the venue publishes no TICK for us-crypto, measured twice — see Subscription), and Core's coverage_by_kind/1 contract is that every reported key is a declared kind, with coverage/1 the exact union of them. A Trade that does arrive is still delivered to subscribers, but record_quote/3 records no coverage for it: counting it would either break that invariant or declare a capability nobody measured. It moves into coverage together with the declaration, when a run shows ticks arriving.

A reconcile that fails leaves its shard dirty, not believed-subscribed

touch_primary_shard/7 and touch_background_shard/5 record the shard's NEW symbol list before the HTTP reconcile has answered. Found 2026-10-10 by reading the path: when that reconcile failed for any reason that does not tear the shard down (a transport error, a :reconcile_timeout, :reconcile_pending, a 429, :oversubscribed, INVALID_SYMBOL) the list was never put back, so a retry diffed the wanted set against a shard that already "had" it, found nothing to send and answered :ok while the symbol stayed unsubscribed. The failure now reverts the delta (added leaves the shard's list, removed returns to it) and records the shard in state.unsettled, which makes the 60-second tick re-plan it even when nobody calls. reconcile_by_session/4 also attempts the subscribe when the unsubscribe half failed, so one failing half no longer cancels the other.

A result is only ever applied to the session it was asked of. The context travels with the attempt (state.settled) and an answer from a session the shard has since replaced is dropped: an INVALID_SESSION or :oversubscribed measured against a dead session says nothing about its healthy successor and used to tear it down or cap it.

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 caller parked on a connecting shard is always answered

A subscribe/2, unsubscribe/2 or update_symbols/2 whose symbols land on a shard that is not yet connected waits for that shard's link_up (shard.reply_to). Two things could leave such a caller unanswered until its @call_timeout ran out, at which point it EXITED and took its own process down:

• Only the branch that OPENS a shard armed a :connack_timeout. A caller that arrived while an existing shard was reconnecting parked with no timer at all, and Socket.handle_disconnect/2 backs off for up to 30s, twice @call_timeout.
• reply_to held ONE caller, and a second caller parking on the same shard overwrote the first, who then got no answer from anything.

reply_to is now a list, and every path that answers a parked caller answers all of them, each with its own capacity overflow. Every park arms its own :connack_timeout, so no caller waits past @connack_timeout_ms, two thirds of @call_timeout.

A caller waiting on a reconcile is answered before its call times out

On a shard that is already connected, subscribe/2, unsubscribe/2 and update_symbols/2 wait for the HTTP subscribe their call started (a :primary reconcile). That reconcile is allowed @reconcile_timeout_ms (60s), because Core.HttpClient retries inside it. The caller's own GenServer.call gives up at @call_timeout (15s), so a slow venue made the caller EXIT and take its process down. The comment on @reconcile_timeout_ms said as much ("a reconcile that outlives the caller waiting on it has already lost that caller") and left it there.

Each call arms ONE {:caller_due, from} when it arrives (reshard/3), at @connack_timeout_ms, the same two-thirds-of-@call_timeout budget a parked caller gets. If the caller is still unanswered then, it is answered: {:error, {:reconcile_pending, ms}} when it rides a primary reconcile, {:error, {:connack_timeout, ms}} when it is parked on a shard (and it is removed from that shard's reply_to). The work carries on and its result is applied. Its own later reply to that from is dropped by gen, because the call has already returned.

The timer is keyed by from, not by shard index and not per reconcile. It used to be armed for each primary reconcile and matched on the index, so an :oversubscribed or INVALID_SYMBOL retry that moved the primary to another index, or parked the caller on a connecting shard, left it matching nothing, and the timers armed by the retry could fall after the caller's own @call_timeout.

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/3 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.

One tick, one attempt

That unconditional re-issue is the only recovery this venue has for a session it has silently stopped publishing to, and for a long time it could not complete once a shard actually needed it. @reconcile_timeout_ms and @resubscribe_interval_ms are both 60,000, and handle_info(:resubscribe, _) re-arms its own timer BEFORE spawning, so at every tick boundary the tick was enqueued ahead of the deadline armed during the previous one. The tick overwrote reconciling[{:resubscribe, index}] with a fresh task; the previous deadline then arrived, found that successor under the tag it had been armed for, and killed it milliseconds old. Issue #3 measured the consequence over a 2h04m boot: 449 reconcile timeouts — against 496 ticks that a four-shard, 60-second interval predicts, so very nearly all of them — producing 12 recoveries.

Two changes, because it was two defects wearing one symptom. A tick whose previous attempt is still running is now SKIPPED rather than stacked on top of it: a second identical subscribe for a session the venue has not answered for yet is load, not a safety net, and validate_resubscribe_interval_ms!/1 already refuses a configured interval under a second in those words. And a deadline now names the ATTEMPT, not merely the tag it ran under, so a timer can only ever tear down the attempt it was armed for — see spawn_reconcile/3.

A tag is not an identity. Every blind resubscribe for shard 2 runs under {:resubscribe, 2}, and anything keyed only by the tag — a deadline, a monitor, a reply — is keyed to the SLOT rather than to the thing occupying it. So an attempt mints a token of its own, and BOTH of the things that conclude one carry it: its deadline and its answer. The answer needs it as much as the deadline does, because killing a superseded attempt cannot recall a result it has already put in the mailbox.

Core.PollingFeed had already answered the first half of this and said why: rescheduling happens in its finish/3 and nowhere else, "which is what bounds the queue — a symbol cannot have a second tick pending until its first has finished". This module re-arms at the TOP of handle_info(:resubscribe, _) instead, which is the opposite, and the two other venues that carry a resubscribe timer are immune only because their re-issue is a synchronous socket send inside the callback rather than an async HTTP call. This feed is the only one in the family whose periodic re-assert is a task tracked by a reusable key with a deadline on it, which is why the defect landed here and nowhere else. Re-arming at the top is kept — the timer must survive a crash in the body — and the in-flight skip in resubscribe_quiet_shards/1 is what supplies the bound instead.

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 map from shard index to that shard's count of consecutive failures — 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/3, 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.

state.rejected is pruned of expired entries on the resubscribe tick, after the re-plan that returns them to a shard. active_rejections/1 always ignored an expired entry, but nothing removed it, so the map only grew.

A symbol list that cannot be used is refused, not crashed on

subscribe/3, unsubscribe/3 and update_symbols/3 take a list of non-empty strings and answer {:error, {:invalid_symbols, symbols}} (the argument exactly as given) before any state moves. This is one named process for every caller, and a raise inside handle_call/3 restarted it from its static options with everyone's symbols gone. The facade has long returned the same tuple when the VENUE refused a symbol, which is what a caller who matches on it wants either way. DpExchange.Webull.Fake answers alike.

A shard that is not connected is never sent an HTTP call by a re-plan

A re-plan that changes a shard other than the one a caller is waiting on used to send its subscribe or unsubscribe at once, whatever that shard was doing. For a shard still connecting, or reconnecting after a drop (the socket backs off for up to 30s), the session was not registered with the venue; the answer was INVALID_SESSION, which this module takes to mean the session is dead, and the shard was torn down and reopened on a fresh session. touch_background_shard/5 now records the new list and sends nothing, as touch_primary_shard/7 always did: on_link_up/2 replays the whole list on the CONNACK. The deferred {:reconcile_shard, ...} message re-checks the session and connected? when it runs, for the same reason.

An unsubscribe that races a whole-list subscribe is sent again

A blind resubscribe and a link-up replay send the shard's list as it stood when they were spawned. An unsubscribe/3 that completes while one is in flight can be overtaken at the venue by the older, larger subscribe, leaving a symbol subscribed that nothing here tracks and nothing will ever diff: it holds one of the session's 100 slots and can make a later TOO_MANY_SYMBOLS cap the shard too low. When such an attempt answers :ok, whatever it sent that the shard no longer carries is unsubscribed once more (unsubscribe_stale/4).

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_step/4's touched computation is what actually limits HTTP calls to changed shards; this is what makes that set small in the first place.