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 or :malformed_header means the stream can no longer be
resynchronised. A :data_quality notice is raised and the connection is closed, as MQTT
3.1.1 section 2.2 requires of a receiver; the ordinary disconnect path then reconnects with
backoff and resubscribes. Retaining the bytes would leave a live socket delivering nothing,
and merely emptying the buffer would read the next frame from an unknown offset - both
silent, 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.
Exactly one ping chain per connection. WebSockex reconnects inside the SAME
process, and handle_connect/2 runs on every reconnect. The chain used to be a bare
:ping timer started by each CONNECT and re-armed by every :ping forever, so each
reconnect started another chain beside the old one, and nothing ever stopped either. This
venue drops this socket in bursts (the "issue #27" section below measured 117 crashes in
seven minutes), which leaves about 118 chains running: a PINGREQ every quarter-second
where the protocol asks for one every thirty. Worse, a surviving chain's :ping, queued
while the socket reconnected, could reach the NEW connection ahead of its CONNECT.
MQTT 3.1.1 requires CONNECT to be the first packet a client sends (§3.1), and a server
closes a connection that sends anything else first — so a stale ping could cause the next
drop itself.
Each CONNECT now arms a timer carrying a fresh reference, kept in state.ping. A
{:ping, ref} that does not match it belongs to a dead connection: it is dropped and NOT
re-armed, so its chain ends there. handle_disconnect/2 clears the reference, so no ping
from the old connection can reach the new one before its CONNECT.
A PINGREQ nobody answers ends the connection
MQTT 3.1.1 requires the server to answer every PINGREQ with a PINGRESP (§3.12). This
socket sent the PINGREQs and never looked for the answers. A network path that died
without either end being told therefore left a "connected" socket delivering nothing
for as long as the operating system's own timeouts allowed, since TCP notices a dead peer
only after its retransmissions give up. Every frame, a PINGRESP included, now counts as
being heard from. When a ping falls due after @silence_ms (three PINGREQ intervals,
90s) with nothing heard, the socket raises a :degraded notice
(details.reason: :silent_connection, with the session_id) and closes. That takes the
ordinary handle_disconnect/2 path: :link_down, reconnect, CONNECT, resubscribe.
Acknowledgements are owed whatever QoS the venue picks
The socket never sends a SUBSCRIBE, so the QoS a session is delivered at is the venue's
choice. A PUBLISH at QoS 1 or 2 is acknowledged (PUBACK, PUBREC, and PUBCOMP for
the PUBREL that follows), all of a frame's acknowledgements in one reply frame. A
broker's in-flight window fills with unacknowledged messages and delivery stops while the
connection, the keep-alive and the subscription all look healthy. See MqttPacket.
QoS 2 is delivered once; QoS 1 may be delivered again, and that is the spec
Acknowledging a QoS 2 PUBLISH is not enough to make it exactly-once. A broker that does
not see the PUBREC in time retransmits the PUBLISH with DUP set, and one that
reuses a packet id before the PUBREL produces a second PUBLISH the same way. Handled
naively, each is a second delivery of a price the consumer already has. MQTT 3.1.1 §4.3.3
says what a receiver does instead: until it has the PUBREL, it acknowledges every
further PUBLISH of that packet id with a PUBREC and "MUST NOT cause duplicate
messages to be delivered to any onward recipients".
So state.qos2_pending holds the packet ids that were delivered and answered with a
PUBREC and have not yet been released. A QoS 2 PUBLISH whose id is in the set is
re-acknowledged and NOT delivered. A PUBREL removes its id and is answered with
PUBCOMP whether or not the id was held, because §4.3.3 requires the PUBCOMP
regardless, and after it the same id is a new publication. The DUP flag is not consulted
and MqttPacket does not expose it: the set is the authority, and a DUP publish whose
id is not in it (the first attempt never arrived) is exactly the first delivery.
The set is bounded by the id space: a packet id is 16 bits, so at most 65,535 can be
outstanding (id 0 is not a valid identifier, §2.3.1). A PUBLISH that would need a
65,536th entry cannot come from a conforming broker; it is neither delivered nor
acknowledged, and a :data_quality notice says so. Delivering it would lose the
exactly-once guarantee silently; acknowledging it would tell the broker it is safe to
forget a message this socket cannot track. The set is cleared on disconnect, because the
CONNECT sets clean session and the broker discards its half of the exchange too.
QoS 1 is deliberately NOT deduplicated. §4.3.2: after sending the PUBACK, the
receiver "MUST treat any incoming PUBLISH packet that contains the same Packet Identifier
as being a new publication, irrespective of the setting of its DUP flag". At-least-once
is the contract at that QoS, and a retransmission is indistinguishable from a legitimate
reuse of the id. It is acknowledged again and delivered again, which for a feed where the
newest value wins costs nothing; a consumer that needs the difference has the venue's
own timestamp on every Quote, TopOfBook and Trade.
Data this socket cannot read is reported
A snapshot, quote or tick that cannot be turned into a Quote, TopOfBook or
Trade, a topic it does not decode, and a packet type it does not handle each raise a
:data_quality notice, once per reason per connection and for at most 64 distinct
reasons. The reason never carries the venue's offending value (that is in details), so
the set of reasons already reported cannot grow with what the venue sends.
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.
delay_ms shortened by a random amount of up to a quarter of itself, never lengthened.
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 jittered_delay_ms(non_neg_integer()) :: non_neg_integer()
delay_ms shortened by a random amount of up to a quarter of itself, never lengthened.
Every shard has its own socket, and the venue drops them together. Feed opens up to
five sockets (one per 100 symbols, against the venue's five-connections-per-App-Key
ceiling), and a venue-side restart or a network fault closes every one of them in the same
instant. reconnect_delay_ms/1 is a pure function of the attempt number, so all of them
then retried on identical schedules: 1s, 2s, 4s, 8s... together, a connect burst of the
whole fleet on every turn of the ladder. This package's own Feed staggers its opens by
shard_spacing_ms because a connect burst is answered with resets, and the retry ladder
undid that stagger at the one time the venue is already refusing connections. A random
shortening per socket spreads the turns apart.
Only shortened, so reconnect_delay_ms/1 stays the ceiling it documents (30 seconds).
0 stays 0: attempt 1 is a live session that just dropped and reconnects at once.
@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.