Streaming follow-ups from the #3 review

[milmazz]

Fixes #142 — all five follow-up items deferred from the final review of #141.

Framing-desync close policy

A frame-level Mint.WebSocket.decode/2 error now tears the connection down with {:closed, {:error, reason}} instead of logging and continuing: once the byte stream is desynced, everything after the bad frame is garbage. Malformed payload JSON inside a well-framed message keeps the spec'd skip behavior. Tested via a new Hunter.RawStreamingServer (bare :gen_tcp + manual upgrade handshake) that can emit framing bytes a conformant WebSock server cannot.

Opt-in auto-reconnect

Hunter.Streaming.connect/2 accepts reconnect: true or [initial_backoff:, max_backoff:, max_attempts:] (defaults 1s / 30s / :infinity). When an established connection drops for any non-local reason, the subscriber receives {:hunter_stream, pid, {:reconnecting, reason}}, the process retries with doubling backoff, replays the current subscription set (tracked across runtime subscribe/3/unsubscribe/3), and sends :reconnected. Non-breaking:

• default stays false with the v1 notify-and-exit behavior
• the initial connect is still synchronous and returns {:error, reason} on failure
• close/1 stays terminal; exhausting max_attempts delivers {:closed, reason} and exits :normal

Handshake receive in the reconnect path re-queues unrelated messages (e.g. $gen_call) instead of silently eating them.

Facade coherence

Hunter.streaming_health?/2 now delegates to Hunter.Streaming.health?/2.

README

Usage gains a short Streaming section (connect, event shape, runtime control, reconnect pointer).

Cosmetic

• integration describe "streaming" block moved above the private helpers in test/integration/mastodon_test.exs
• nginx CI config switched to the map $http_upgrade $connection_upgrade idiom; validated with nginx -t on the nginx:1.27-alpine CI image

Verification

• mix test — 2 doctests, 251 tests, 0 failures (5 new streaming tests: framing teardown, facade delegate, reconnect+resubscribe, runtime-set replay, terminal close, max_attempts give-up)
• mix credo --strict, mix dialyzer, mix docs, mix format --check-formatted all clean

🤖 Generated with Claude Code

[milmazz]

Suggested change

	with {:ok, state} <- establish(state),
	{:ok, state} <- send_subscriptions(state) do
	{:ok, state}
	else
	{:error, reason} -> {:error, reason}
	{:error, _conn, reason} -> {:error, reason}
	end
	with {:ok, state} <- establish(state) do
	send_subscriptions(state)
	end

[milmazz]

Not applied, but for a reason that supersedes it: the review's leak finding means send_subscriptions/1 now returns {:error, state, reason} so callers can close the live connection. init/1 therefore can't pass its result through unchanged — it intercepts the error tuple, closes the conn via drop_connection/1, and returns {:error, reason} (06bf27a). The redundant with/else this suggestion targeted is gone either way.

[milmazz]

@claude review once.

[claude]

Claude finished @milmazz's task in 4m 28s —— View job

---

Review: Streaming follow-ups (#150)

• Gather context (PR diff, review comments, CLAUDE.md — none present)
• Review lib/hunter/streaming/connection.ex reconnect/close logic
• Review lib/hunter/streaming.ex and lib/hunter.ex facade changes
• Review test additions (raw_streaming_server.ex, streaming_test.exs, integration test reorg)
• Review README/CHANGELOG/nginx changes
• Post final review feedback

Overall this closes out the five #141 follow-ups cleanly: the framing-desync teardown, opt-in reconnect, facade delegate, and cosmetic moves all match the PR description and read well. One real bug worth fixing before merge, plus a couple of smaller notes.

Note on environment: I wasn't able to run mix test/credo/dialyzer here — Bash calls beyond trivial ones (including git fetch, mix test) were blocked pending approval in this session. Everything below is from static reading of the diff against origin/main, not from running the suite.

Bug: leaked Mint connection on reconnect when handshake succeeds but resubscribe fails

lib/hunter/streaming/connection.ex:92-100:

def handle_info(:reconnect, %{conn: nil} = state) do
  with {:ok, state} <- establish(state),
       {:ok, state} <- send_subscriptions(state) do
    send(state.subscriber, {:hunter_stream, self(), :reconnected})
    {:noreply, %{state | attempts: 0}}
  else
    {:error, reason} -> retry_or_stop(state, reason)
  end
end

If establish(state) succeeds (opens a new Mint conn) but the following send_subscriptions(state) then fails, Elixir's with/else can't see the intermediate state bound inside the with — state in the else clause still refers to the function's original argument (conn: nil). So retry_or_stop(state, reason) runs against the old, disconnected state, and the just-opened connection from establish/1 is never passed to Mint.HTTP.close/1. Each time this happens during a flapping reconnect, one Mint/TCP connection leaks.

This is compounded by send_subscriptions/1 (lib/hunter/streaming/connection.ex:217-226) itself discarding the live state on failure:

case send_frame(state, {:text, Poison.encode!(frame)}) do
  {:ok, state} -> {:cont, {:ok, state}}
  {:error, _state, reason} -> {:halt, {:error, reason}}   # state (and its conn) is dropped here
end

(This exact pattern also exists in init/1's use of send_subscriptions, inherited unchanged from the pre-PR subscribe_initial/2 — a pre-existing minor issue, but the new handle_info(:reconnect, ...) path turns it into a repeating leak across reconnect attempts rather than a one-shot init failure.)

Suggested fix: have send_subscriptions/1 return the state alongside the error ({:error, state, reason}), and in handle_info(:reconnect, ...) close that connection before calling retry_or_stop/2 on failure.

Fix this →

Your own inline suggestion (init/1, line 58)

Your suggested simplification:

with {:ok, state} <- establish(state) do
  send_subscriptions(state)
end

is correct — send_subscriptions/1 already returns {:ok, state} | {:error, reason}, exactly what init/1 needs, so the second with clause and the pass-through else are redundant. Worth applying regardless of the bug above (they're independent).

Minor / non-blocking

• test/hunter/streaming_test.exs: there's no test closing the connection while it's in the conn: nil backoff window (between a drop and the next reconnect attempt) — only the "still connected" case is covered by "close/1 stays terminal when reconnect is enabled" (line 235). handle_call(:close, _from, %{conn: nil} = state) (connection.ex:77) handles that path but it isn't exercised.
• disconnect/3's reconnect branch (connection.ex:250-254) pattern-matches %{initial_backoff: backoff} and schedules with the raw value rather than calling the existing backoff/2 helper — equivalent to backoff(reconnect, 0), just spelled out manually. Not wrong, just a little redundant with the helper that already exists.

Everything else — the framing-desync vs. malformed-JSON distinction, the RawStreamingServer approach for producing bytes a conformant WebSock server can't, the subscription replay/dedup on subscribe/3, the await_upgrade message requeueing, README/CHANGELOG/nginx changes — looks solid and matches the PR description.

[milmazz]

All four findings addressed in 06bf27a:

• Leaked Mint connection on failed resubscribe (bug): confirmed — with/else can't see the state bound by establish/1, so the fresh connection was dropped unclosed. send_subscriptions/1 now returns {:error, state, reason}, and both callers close the connection through a new drop_connection/1 helper: handle_info(:reconnect, ...) closes it before retry_or_stop/2 (fixing the repeating leak), and init/1 closes it before returning {:error, reason} (fixing the pre-existing one-shot leak).
• Inline with simplification on init/1: superseded by the above — init now has to intercept the 3-tuple error to close the conn, so the pass-through shape is gone entirely (replied in-thread).
• close/1 during the backoff window: covered by a new test (initial_backoff: 60_000, drop the server, close during {:reconnecting, _}) asserting {:closed, :local} and a normal exit with no reconnect attempt.
• disconnect/3 raw initial_backoff: now calls the existing backoff(reconnect, 0) helper and reuses drop_connection/1 instead of the inlined close/nil sequence.

No test asserts the absence of the leak itself — there's no deterministic wire-level way to make the upgrade succeed and the first subscription frame fail (a server-side FIN still lets the client's first send succeed), so that fix is verified by inspection plus the suite. Verification: mix test (252 tests, 0 failures), mix credo --strict, mix dialyzer, mix format --check-formatted all clean.