The MQTT-over-WebSocket connection — internal, never named above the facade.
websockex carries the frames; MqttPacket provides the framing; QuoteProto decodes
the payloads. None of those three names can appear in anything a consumer receives.
The buffer is the point
A WebSocket frame is not an MQTT packet. One frame may carry several packets, or
half of one, and the broker is under no obligation to align them. So every inbound frame
is appended to a buffer and the buffer is drained packet by packet until it returns
:incomplete.
Assuming frame boundaries are packet boundaries loses every packet after the first coalesced frame — quietly, because the socket stays up and the first message parses fine.
A :malformed_length means the stream can no longer be resynchronised. The buffer is
dropped and a notice raised rather than retaining bytes that can never parse: keeping
them would leave a live socket delivering nothing, which is the failure mode this family
ranks worst.
Keep-alive
The venue's CONNECT carries a keep-alive interval, and MQTT requires the client to send
something within it. PINGREQ goes out at half that interval — early enough that one
lost ping is not a disconnect.
What it does not do
It does not subscribe. Subscriptions on this venue are HTTP calls, made by Feed
with the same session_id this connection registered as its MQTT client id. The socket's
only job is to connect, stay connected, and turn payloads into Core.Types.Quote (the
snapshot topic), Core.Types.TopOfBook (the quote topic) or Core.Types.Trade (the
tick topic) — see emit/3 below.
tick carries no trade id, and Trade.id is nil rather than invented
The venue's Tick message (docs/reference/webull/streaming-api.md) is Basic, time,
price, volume, side — nothing identifies one print from the next. Core.Types.Trade
requires :id, so this builds the struct literally (%Trade{id: nil, ...}) rather than
through Trade.new/1, the same way Rest.get_trades/3's to_trade/2 already does for
this venue's REST tape, which has the identical gap and says so in its own comment.
Trade.new/1's validation exists to catch an accidentally-absent required field; this
absence is not accidental, so going around it here is not going around the check — it is
the one place a real, checked absence is allowed to be nil instead of failing closed.
Ending a session cleanly
It does not close itself, either — Feed decides when a shard's session ends, and
disconnect/2 is how it says so on the wire before that shard's process goes down. See
MqttPacket.disconnect/0 for why a clean DISCONNECT matters and Feed's own
terminate/2 for where this is actually called.
The transport underneath is a private fork — dp-exchange-core issue #27
Webull sometimes closes this socket with a WebSocket close frame that carries prose
instead of a 2-byte RFC 6455 status code (measured: "bye-bye!!!", no valid close code
in the first two bytes). Hex's latest websockex (0.5.1, and its unreleased upstream)
raises and kills the process on that frame before handle_disconnect/2 below ever
runs — turning a peer's protocol violation into 117 crashes in 7 minutes, live. See
DpExchange.Webull.Vendor.WebSockex's own moduledoc for the full incident, the exact
two-line fix, and why vendoring (not switching transports, not waiting on upstream) was
the right call. use WebSockex two lines below resolves to that vendored module, not
the real dependency, via the alias immediately above it — every callback in this
module is otherwise unchanged.
Summary
Functions
The connection options handed to WebSockex.start_link/4.
Sends a clean MQTT DISCONNECT on an already-open socket, synchronously.
How long to wait before the reconnect that attempt is about to make.
Functions
The connection options handed to WebSockex.start_link/4.
Exposed so the deliberate timeouts can be asserted without opening a real socket — a later refactor must not be able to drop them back to the dependency's defaults unnoticed.
Sends a clean MQTT DISCONNECT on an already-open socket, synchronously.
WebSockex.send_frame/3 is a :gen.call against the socket process itself — the only
way to put a frame on an already-running Socket from outside its own callbacks, since
MqttPacket's framing stays private to this module (Feed must not learn it — see the
moduledoc's boundary). Called by Feed's own terminate/2, once per shard still
connected when this package is shutting down cleanly.
Best-effort and never raises: a shard whose socket has already gone — crashed, already
reconnecting, already torn down by the time shutdown reaches it — must not block or
crash the shutdown asking for this. {:error, reason} says so; there is nothing a
caller mid-shutdown can usefully do with it beyond logging, which Feed does.
The pid == self() guard exists because this runs from inside Feed's own
terminate/2: a call this deep can legitimately end up with pid being the calling
process itself only through a test fixture, never in production (a shard's socket is
always a distinct Socket.start_link/1 process) — but WebSockex.send_frame/3 answers
that specific case by raising WebSockex.CallingSelfError rather than returning an
error, which the catch below alone would not stop. Checked first so this function's
own "never raises" holds regardless.
@spec reconnect_delay_ms(pos_integer()) :: non_neg_integer()
How long to wait before the reconnect that attempt is about to make.
websockex reconnects with no delay of its own. on_disconnect/5 in
deps/websockex/lib/websockex.ex calls open_connection/3 and, on failure, calls itself
with attempt + 1 — a synchronous loop with nothing between the turns. So a socket the
venue will not accept back reconnects at full connect speed, forever, and the things that
cause it are exactly the things that do not fix themselves by being retried sooner:
credentials the venue has stopped honouring, an IP it has started refusing, a maintenance
window, a 503. CLAUDE.md's own testing tiers say what a venue does about traffic like
that — "a venue that sees a package polling it on a timer will rate-limit or block" — and
a reconnect storm is that, without the timer.
Attempt 1 waits nothing. It is a live session that just dropped, and nothing about an ordinary network blip suggests waiting helps. Every attempt after it is a reconnect that has already failed at least once, so the wait doubles from 1000ms, capped at 30000ms.
The same shape, and the same two constants, as DpExchange.Schwab.Socket's
reconnect_delay_ms/1, which had this venue family's only reconnect backoff until now.
Its counter is LOGIN_DENIEDs specifically because that venue can name its own auth
rejection; here the counter is websockex's consecutive-failure count, which needs no
venue-specific signal and is already correct for every reason a reconnect can fail.