From a7e0770f19fa4864ead9939435897167b450d837 Mon Sep 17 00:00:00 2001 From: Milton Mazzarri Date: Thu, 16 Jul 2026 01:12:38 -0500 Subject: [PATCH 1/2] Streaming follow-ups from the #3 review (#142) - Frame-level decode errors now tear the connection down with {:closed, {:error, reason}} instead of logging and continuing: after a framing desync everything that follows is garbage. Malformed payload JSON inside a well-framed message is still skipped. Covered by a new raw gen_tcp WebSocket helper that can emit bytes a conformant server cannot. - Opt-in auto-reconnect: connect/2 accepts reconnect: true or [initial_backoff:, max_backoff:, max_attempts:]; non-local drops notify the subscriber with {:reconnecting, reason}, retry with exponential backoff, replay the current subscription set (tracked across runtime subscribe/unsubscribe), and send :reconnected. close/1 stays terminal; exhausting max_attempts delivers {:closed, reason} and exits. Messages skipped during a reconnect handshake are re-queued instead of dropped. - Facade coherence: Hunter.streaming_health?/2 delegates to Hunter.Streaming.health?/2. - README: add a Streaming section to Usage. - Cosmetic: move the integration describe "streaming" block above the private helpers; nginx CI config uses the map $http_upgrade idiom (validated with nginx -t on the CI image). Fixes #142 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 12 +- README.md | 21 +++ lib/hunter.ex | 15 ++ lib/hunter/streaming.ex | 26 ++- lib/hunter/streaming/connection.ex | 227 ++++++++++++++++++++------- scripts/ci/nginx.conf | 9 +- test/hunter/streaming_test.exs | 111 +++++++++++++ test/integration/mastodon_test.exs | 60 +++---- test/support/raw_streaming_server.ex | 61 +++++++ test/support/streaming_server.ex | 12 +- 10 files changed, 459 insertions(+), 95 deletions(-) create mode 100644 test/support/raw_streaming_server.ex diff --git a/CHANGELOG.md b/CHANGELOG.md index 41445fd..8054b2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,9 +18,14 @@ `health?/2`; parsed events (`Hunter.Streaming.Event` — payloads decode to `Status`, `Notification`, `Conversation`, `Announcement`, `Announcement.Reaction`, or id strings for deletes) are delivered to - a subscriber pid as `{:hunter_stream, pid, event}` messages. No - automatic reconnection: the process notifies the subscriber and - exits, so callers supervise it + a subscriber pid as `{:hunter_stream, pid, event}` messages. By + default there is no automatic reconnection (the process notifies + the subscriber and exits, so callers supervise it); an opt-in + `reconnect` mode retries dropped connections with exponential + backoff and replays the subscription set ([#142]). Frame-level + decode errors tear the connection down instead of skipping the + desynced byte stream, and `health?/2` is also reachable as + `Hunter.streaming_health?/2` on the facade ([#142]) - Account extras ([#124]): `lookup_account/2`, `accounts_by_ids/2`, `familiar_followers/2` (new `Hunter.FamiliarFollowers` entity), `account_featured_tags/2`, `register_account/2` (returns a @@ -182,6 +187,7 @@ [#116]: https://github.com/milmazz/hunter/issues/116 [#122]: https://github.com/milmazz/hunter/issues/122 [#3]: https://github.com/milmazz/hunter/issues/3 +[#142]: https://github.com/milmazz/hunter/issues/142 [#123]: https://github.com/milmazz/hunter/issues/123 [#124]: https://github.com/milmazz/hunter/issues/124 [#126]: https://github.com/milmazz/hunter/issues/126 diff --git a/README.md b/README.md index e60a117..4512a3a 100644 --- a/README.md +++ b/README.md @@ -495,6 +495,27 @@ iex> Hunter.update_credentials(conn, %{note: "Enum.random(~w(programming cycling Returns a `Hunter.Account` +### Streaming + +Real-time events are delivered over Mastodon's multiplexed streaming +WebSocket. `Hunter.Streaming.connect/2` opens a connection process linked +to the caller and sends parsed events to the subscriber's mailbox: + +```elixir +iex> {:ok, pid} = Hunter.Streaming.connect(conn, streams: ["user", {"hashtag", tag: "elixir"}]) +{:ok, #PID<0.233.0>} +iex> flush() +{:hunter_stream, #PID<0.233.0>, + %Hunter.Streaming.Event{stream: ["user"], type: "update", payload: %Hunter.Status{...}}} +``` + +Streams can also be joined and left at runtime with +`Hunter.Streaming.subscribe/3` and `Hunter.Streaming.unsubscribe/3`, and +the connection closed with `Hunter.Streaming.close/1`. A single +`{:hunter_stream, pid, {:closed, reason}}` message signals that the socket +is gone; pass `reconnect: true` to `connect/2` to retry drops with +exponential backoff instead. See the `Hunter.Streaming` docs for details. + ### Configuration Hunter uses [Req](https://hex.pm/packages/req) as its HTTP client layer. diff --git a/lib/hunter.ex b/lib/hunter.ex index caab7f9..396c2ad 100644 --- a/lib/hunter.ex +++ b/lib/hunter.ex @@ -2476,6 +2476,21 @@ defmodule Hunter do Request.request!(conn, :delete, "/api/v2/filters/statuses/#{id}", :empty) end + @doc """ + Checks the streaming server's health endpoint (Mastodon 2.5+) + + ## Parameters + + * `conn` - connection credentials + * `opts` - `url:` overrides the streaming base URL (`ws://`/`wss://` + accepted and mapped to `http://`/`https://`) + + See `Hunter.Streaming.health?/2`; streaming connections themselves are + opened with `Hunter.Streaming.connect/2`. + """ + @spec streaming_health?(Hunter.Client.t(), Keyword.t()) :: boolean + defdelegate streaming_health?(conn, opts \\ []), to: Hunter.Streaming, as: :health? + @doc """ Returns Hunter version """ diff --git a/lib/hunter/streaming.ex b/lib/hunter/streaming.ex index 081c10c..78d93c8 100644 --- a/lib/hunter/streaming.ex +++ b/lib/hunter/streaming.ex @@ -6,8 +6,17 @@ defmodule Hunter.Streaming do events arrive in the subscriber's mailbox as `{:hunter_stream, connection_pid, %Hunter.Streaming.Event{}}` and a single `{:hunter_stream, connection_pid, {:closed, reason}}` is sent - when the socket closes. There is no automatic reconnection: supervise - and restart the connection from the consuming application. + when the socket closes. By default there is no automatic reconnection: + supervise and restart the connection from the consuming application. + + With the `reconnect` option, a drop of an established connection is + retried with exponential backoff instead of being terminal: the + subscriber receives `{:hunter_stream, connection_pid, {:reconnecting, + reason}}` when the socket drops and `{:hunter_stream, connection_pid, + :reconnected}` once it is re-established and the current subscription + set has been replayed. `close/1` remains terminal, and a reconnect mode + with `max_attempts` still delivers `{:closed, reason}` and exits once + the attempts are exhausted. Instances may serve streaming from a different host than the REST API; discover it via `Hunter.instance_info/1` under @@ -36,6 +45,16 @@ defmodule Hunter.Streaming do (see the module docs for discovery) * `transport_opts` - Mint transport options, e.g. `[verify: :verify_none]` for self-signed certificates + * `reconnect` - `true` or a keyword list to reconnect automatically + when an established connection drops (see the module docs for the + subscriber messages involved), default: `false`. The initial + connection is always synchronous: `connect/2` still returns + `{:error, reason}` when it fails. Keys: + * `initial_backoff` - delay in ms before the first attempt, + doubled on each failure, default: `1_000` + * `max_backoff` - backoff ceiling in ms, default: `30_000` + * `max_attempts` - consecutive failed attempts tolerated before + giving up, default: `:infinity` """ @spec connect(Hunter.Client.t(), Keyword.t()) :: {:ok, pid} | {:error, term} @@ -44,7 +63,8 @@ defmodule Hunter.Streaming do uri: ws_uri(conn, opts), subscriber: Keyword.get(opts, :subscriber, self()), streams: Keyword.get(opts, :streams, []), - transport_opts: Keyword.get(opts, :transport_opts, []) + transport_opts: Keyword.get(opts, :transport_opts, []), + reconnect: Keyword.get(opts, :reconnect, false) ) end diff --git a/lib/hunter/streaming/connection.ex b/lib/hunter/streaming/connection.ex index 636a761..71be035 100644 --- a/lib/hunter/streaming/connection.ex +++ b/lib/hunter/streaming/connection.ex @@ -10,13 +10,27 @@ defmodule Hunter.Streaming.Connection do alias Hunter.Streaming.Event @handshake_timeout 15_000 + @reconnect_defaults %{initial_backoff: 1_000, max_backoff: 30_000, max_attempts: :infinity} # OTP's gen_server accepts {:error, reason} from init/1 as a graceful # failure that doesn't crash linked callers; Elixir's GenServer callback # spec hasn't caught up with it yet. @dialyzer {:nowarn_function, init: 1} - defstruct [:conn, :websocket, :ref, :subscriber] + # conn/websocket/ref are nil between a drop and a successful reconnect; + # subscriptions holds {stream, params_map} in subscription order so a + # reconnect can replay them. + defstruct [ + :conn, + :websocket, + :ref, + :subscriber, + :uri, + :transport_opts, + :reconnect, + subscriptions: [], + attempts: 0 + ] def start_link(opts), do: GenServer.start_link(__MODULE__, opts) @@ -26,42 +40,42 @@ defmodule Hunter.Streaming.Connection do subscriber = Keyword.fetch!(opts, :subscriber) streams = Keyword.get(opts, :streams, []) transport_opts = Keyword.get(opts, :transport_opts, []) - - {http_scheme, ws_scheme} = schemes(uri.scheme) - path = uri.path <> "?" <> uri.query - - with {:ok, conn} <- - Mint.HTTP.connect(http_scheme, uri.host, uri.port, - protocols: [:http1], - transport_opts: transport_opts - ), - {:ok, conn, ref} <- Mint.WebSocket.upgrade(ws_scheme, conn, path, []), - {:ok, conn, status, headers} <- await_upgrade(conn, ref), - # mode: :active is the default; passing it explicitly reaches new/5 - # directly because new/4's @spec trips a mint_web_socket typespec bug - # (Mint.WebSocket.t declares fragment: tuple() while the struct - # defaults it to nil), which makes dialyzer type new/4 as error-only. - {:ok, conn, websocket} <- Mint.WebSocket.new(conn, ref, status, headers, mode: :active) do - state = %__MODULE__{conn: conn, websocket: websocket, ref: ref, subscriber: subscriber} - - subscribe_initial(state, streams) + reconnect = reconnect_config(Keyword.get(opts, :reconnect, false)) + + state = %__MODULE__{ + subscriber: subscriber, + uri: uri, + transport_opts: transport_opts, + reconnect: reconnect, + subscriptions: Enum.map(streams, &normalize_stream_spec/1) + } + + with {:ok, state} <- establish(state), + {:ok, state} <- send_subscriptions(state) do + {:ok, state} else {:error, reason} -> {:error, reason} - {:error, _conn, reason} -> {:error, reason} end end @impl GenServer - def handle_call({:control, type, stream, params}, _from, state) do - frame = - params - |> Map.new(fn {key, value} -> {to_string(key), to_string(value)} end) - |> Map.merge(%{"type" => type, "stream" => stream}) + def handle_call({:control, "subscribe", stream, params}, _from, state) do + subscription = normalize_stream_spec({stream, params}) + state = %{state | subscriptions: state.subscriptions -- [subscription]} + state = %{state | subscriptions: state.subscriptions ++ [subscription]} - case send_frame(state, {:text, Poison.encode!(frame)}) do - {:ok, state} -> {:reply, :ok, state} - {:error, state, reason} -> stop_with(state, {:error, reason}, {:reply, :ok}) - end + send_control(state, "subscribe", subscription) + end + + def handle_call({:control, "unsubscribe", stream, params}, _from, state) do + subscription = normalize_stream_spec({stream, params}) + state = %{state | subscriptions: state.subscriptions -- [subscription]} + + send_control(state, "unsubscribe", subscription) + end + + def handle_call(:close, _from, %{conn: nil} = state) do + stop_with(state, :local, {:reply, :ok}) end def handle_call(:close, _from, state) do @@ -75,19 +89,46 @@ defmodule Hunter.Streaming.Connection do end @impl GenServer + 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 + + # Stragglers from the torn-down socket while waiting to reconnect. + def handle_info(_message, %{conn: nil} = state), do: {:noreply, state} + def handle_info(message, state) do case Mint.WebSocket.stream(state.conn, message) do {:ok, conn, entries} -> handle_entries(entries, %{state | conn: conn}) {:error, conn, reason, _responses} -> - stop_with(%{state | conn: conn}, {:error, reason}, :noreply) + disconnect(%{state | conn: conn}, {:error, reason}, :noreply) :unknown -> {:noreply, state} end end + defp send_control(%{conn: nil} = state, _type, _subscription) do + # Disconnected: the updated subscription set is replayed on reconnect. + {:reply, :ok, state} + end + + defp send_control(state, type, {stream, params}) do + frame = Map.merge(params, %{"type" => type, "stream" => stream}) + + case send_frame(state, {:text, Poison.encode!(frame)}) do + {:ok, state} -> {:reply, :ok, state} + {:error, state, reason} -> disconnect(state, {:error, reason}, {:reply, :ok}) + end + end + defp handle_entries(entries, state) do frames = for {:data, ref, data} <- entries, ref == state.ref do @@ -96,6 +137,8 @@ defmodule Hunter.Streaming.Connection do Enum.reduce_while(frames, {:noreply, state}, fn data, {:noreply, state} -> case decode_frames(state, data) do + # Disconnected mid-batch: the rest of the data is from the dead socket. + {:noreply, %{conn: nil} = state} -> {:halt, {:noreply, state}} {:noreply, state} -> {:cont, {:noreply, state}} stop -> {:halt, stop} end @@ -108,8 +151,7 @@ defmodule Hunter.Streaming.Connection do dispatch_frames(frames, %{state | websocket: websocket}) {:error, websocket, reason} -> - Logger.warning("Hunter.Streaming: undecodable data: #{inspect(reason)}") - {:noreply, %{state | websocket: websocket}} + disconnect(%{state | websocket: websocket}, {:error, reason}, :noreply) end end @@ -130,33 +172,51 @@ defmodule Hunter.Streaming.Connection do defp dispatch_frames([{:ping, data} | rest], state) do case send_frame(state, {:pong, data}) do {:ok, state} -> dispatch_frames(rest, state) - {:error, state, reason} -> stop_with(state, {:error, reason}, :noreply) + {:error, state, reason} -> disconnect(state, {:error, reason}, :noreply) end end defp dispatch_frames([{:close, code, _reason} | _rest], state) do - stop_with(state, {:remote, code}, :noreply) + disconnect(state, {:remote, code}, :noreply) end - defp dispatch_frames([{:error, reason} | rest], state) do - Logger.warning("Hunter.Streaming: skipping undecodable frame: #{inspect(reason)}") - dispatch_frames(rest, state) + # A frame-level decode error means the byte stream is desynced; anything + # after it is garbage, so tear down rather than skip (unlike malformed + # payload JSON, which arrives in a well-framed message and is skipped). + defp dispatch_frames([{:error, reason} | _rest], state) do + Logger.warning("Hunter.Streaming: undecodable frame, closing: #{inspect(reason)}") + disconnect(state, {:error, reason}, :noreply) end defp dispatch_frames([_other | rest], state), do: dispatch_frames(rest, state) - defp subscribe_initial(state, streams) do - Enum.reduce_while(streams, {:ok, state}, fn spec, {:ok, state} -> - {stream, params} = - case spec do - {stream, params} -> {stream, params} - stream when is_binary(stream) -> {stream, []} - end + defp establish(state) do + uri = state.uri + {http_scheme, ws_scheme} = schemes(uri.scheme) + path = uri.path <> "?" <> uri.query + + with {:ok, conn} <- + Mint.HTTP.connect(http_scheme, uri.host, uri.port, + protocols: [:http1], + transport_opts: state.transport_opts + ), + {:ok, conn, ref} <- Mint.WebSocket.upgrade(ws_scheme, conn, path, []), + {:ok, conn, status, headers} <- await_upgrade(conn, ref), + # mode: :active is the default; passing it explicitly reaches new/5 + # directly because new/4's @spec trips a mint_web_socket typespec bug + # (Mint.WebSocket.t declares fragment: tuple() while the struct + # defaults it to nil), which makes dialyzer type new/4 as error-only. + {:ok, conn, websocket} <- Mint.WebSocket.new(conn, ref, status, headers, mode: :active) do + {:ok, %{state | conn: conn, websocket: websocket, ref: ref}} + else + {:error, reason} -> {:error, reason} + {:error, _conn, reason} -> {:error, reason} + end + end - frame = - params - |> Map.new(fn {key, value} -> {to_string(key), to_string(value)} end) - |> Map.merge(%{"type" => "subscribe", "stream" => stream}) + defp send_subscriptions(state) do + Enum.reduce_while(state.subscriptions, {:ok, state}, fn {stream, params}, {:ok, state} -> + frame = Map.merge(params, %{"type" => "subscribe", "stream" => stream}) case send_frame(state, {:text, Poison.encode!(frame)}) do {:ok, state} -> {:cont, {:ok, state}} @@ -180,9 +240,44 @@ defmodule Hunter.Streaming.Connection do end end + # A non-local drop of an established connection: tear down for good, or + # schedule a reconnect when the mode is enabled. + defp disconnect(state, reason, reply_or_noreply) do + case state.reconnect do + nil -> + stop_with(state, reason, reply_or_noreply) + + %{initial_backoff: backoff} -> + send(state.subscriber, {:hunter_stream, self(), {:reconnecting, reason}}) + if state.conn, do: Mint.HTTP.close(state.conn) + state = %{state | conn: nil, websocket: nil, ref: nil, attempts: 0} + Process.send_after(self(), :reconnect, backoff) + + case reply_or_noreply do + {:reply, value} -> {:reply, value, state} + :noreply -> {:noreply, state} + end + end + end + + defp retry_or_stop(state, reason) do + attempts = state.attempts + 1 + + if state.reconnect.max_attempts != :infinity and attempts >= state.reconnect.max_attempts do + stop_with(state, {:error, reason}, :noreply) + else + Process.send_after(self(), :reconnect, backoff(state.reconnect, attempts)) + {:noreply, %{state | attempts: attempts}} + end + end + + defp backoff(%{initial_backoff: initial, max_backoff: max}, attempts) do + min(initial * Integer.pow(2, attempts), max) + end + defp stop_with(state, reason, reply_or_noreply) do send(state.subscriber, {:hunter_stream, self(), {:closed, reason}}) - Mint.HTTP.close(state.conn) + if state.conn, do: Mint.HTTP.close(state.conn) case reply_or_noreply do {:reply, value} -> {:stop, :normal, value, state} @@ -190,7 +285,22 @@ defmodule Hunter.Streaming.Connection do end end - defp await_upgrade(conn, ref, status \\ nil, headers \\ nil) do + defp reconnect_config(false), do: nil + defp reconnect_config(true), do: @reconnect_defaults + + defp reconnect_config(opts) when is_list(opts) do + Map.merge(@reconnect_defaults, Map.new(opts)) + end + + defp normalize_stream_spec({stream, params}) do + {stream, Map.new(params, fn {key, value} -> {to_string(key), to_string(value)} end)} + end + + defp normalize_stream_spec(stream) when is_binary(stream), do: {stream, %{}} + + # Messages that aren't part of the handshake (late traffic from a + # previous socket, GenServer calls) are re-queued once it completes. + defp await_upgrade(conn, ref, status \\ nil, headers \\ nil, pending \\ []) do receive do message -> case Mint.WebSocket.stream(conn, message) do @@ -199,26 +309,35 @@ defmodule Hunter.Streaming.Connection do cond do not Enum.any?(entries, &match?({:done, ^ref}, &1)) -> - await_upgrade(conn, ref, status, headers) + await_upgrade(conn, ref, status, headers, pending) is_integer(status) and is_list(headers) -> + requeue(pending) {:ok, conn, status, headers} true -> + requeue(pending) {:error, :handshake_incomplete} end {:error, _conn, reason, _responses} -> + requeue(pending) {:error, reason} :unknown -> - await_upgrade(conn, ref, status, headers) + await_upgrade(conn, ref, status, headers, [message | pending]) end after - @handshake_timeout -> {:error, :handshake_timeout} + @handshake_timeout -> + requeue(pending) + {:error, :handshake_timeout} end end + defp requeue(pending) do + pending |> Enum.reverse() |> Enum.each(&send(self(), &1)) + end + defp collect_upgrade_entries(entries, ref, status, headers) do Enum.reduce(entries, {status, headers}, fn {:status, ^ref, status}, {_status, headers} -> {status, headers} diff --git a/scripts/ci/nginx.conf b/scripts/ci/nginx.conf index 9727977..76323d9 100644 --- a/scripts/ci/nginx.conf +++ b/scripts/ci/nginx.conf @@ -1,6 +1,13 @@ events {} http { + # Standard WebSocket proxy idiom: forward the client's upgrade intent, + # falling back to "close" for plain HTTP requests (the health check). + map $http_upgrade $connection_upgrade { + default upgrade; + '' close; + } + server { listen 3000 ssl; @@ -13,7 +20,7 @@ http { proxy_pass http://streaming:4000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; + proxy_set_header Connection $connection_upgrade; proxy_set_header Host localhost; proxy_set_header X-Forwarded-Proto https; proxy_read_timeout 120s; diff --git a/test/hunter/streaming_test.exs b/test/hunter/streaming_test.exs index da3f32b..3d70646 100644 --- a/test/hunter/streaming_test.exs +++ b/test/hunter/streaming_test.exs @@ -20,6 +20,15 @@ defmodule Hunter.StreamingTest do refute Hunter.Streaming.health?(@conn) end + test "is reachable as Hunter.streaming_health?/2 on the facade" do + stub_request(fn conn -> + assert conn.request_path == "/api/v1/streaming/health" + respond_with(conn, "OK") + end) + + assert Hunter.streaming_health?(@conn) + end + test "resolves a ws(s) url override to http(s)" do stub_request(fn conn -> assert conn.host == "streaming.example" @@ -142,6 +151,21 @@ defmodule Hunter.StreamingTest do assert_receive {:DOWN, ^ref, :process, ^pid, :normal} end + @tag :capture_log + test "framing-level garbage tears the connection down" do + {_server, port} = Hunter.RawStreamingServer.start(self()) + {:ok, pid} = Hunter.Streaming.connect(client(), url: "ws://localhost:#{port}") + assert_receive {:raw_ws_connected, raw} + ref = Process.monitor(pid) + + # fin=1, reserved opcode 0x3, unmasked, empty payload: a frame no + # conformant server emits, so the socket is presumed desynced. + send(raw, {:push_raw, <<0x83, 0x00>>}) + + assert_receive {:hunter_stream, ^pid, {:closed, {:error, _reason}}} + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + end + test "server going away delivers {:closed, {:error, reason}}", %{pid: pid, ws: ws} do ref = Process.monitor(pid) Process.exit(ws, :kill) @@ -159,5 +183,92 @@ defmodule Hunter.StreamingTest do end end + describe "auto-reconnect" do + test "reconnects after a remote close, resubscribing and notifying the subscriber" do + {_server, port} = Hunter.StreamingServer.start(self()) + + {:ok, pid} = + Hunter.Streaming.connect(client(), + url: "ws://localhost:#{port}", + streams: ["user"], + reconnect: [initial_backoff: 10] + ) + + assert_receive {:ws_connected, ws} + assert_receive {:ws_frame, %{"type" => "subscribe", "stream" => "user"}} + + send(ws, {:close, 1_012}) + + assert_receive {:hunter_stream, ^pid, {:reconnecting, {:remote, 1_012}}} + assert_receive {:ws_connected, _ws2} + assert_receive {:ws_frame, %{"type" => "subscribe", "stream" => "user"}} + assert_receive {:hunter_stream, ^pid, :reconnected} + assert Process.alive?(pid) + end + + test "replays the runtime subscription set, not the initial one" do + {_server, port} = Hunter.StreamingServer.start(self()) + + {:ok, pid} = + Hunter.Streaming.connect(client(), + url: "ws://localhost:#{port}", + streams: ["user"], + reconnect: [initial_backoff: 10] + ) + + assert_receive {:ws_connected, ws} + assert_receive {:ws_frame, %{"type" => "subscribe", "stream" => "user"}} + + :ok = Hunter.Streaming.subscribe(pid, "list", list: "12") + assert_receive {:ws_frame, %{"type" => "subscribe", "stream" => "list"}} + :ok = Hunter.Streaming.unsubscribe(pid, "user") + assert_receive {:ws_frame, %{"type" => "unsubscribe", "stream" => "user"}} + + send(ws, {:close, 1_012}) + + assert_receive {:ws_connected, _ws2} + assert_receive {:ws_frame, %{"type" => "subscribe", "stream" => "list", "list" => "12"}} + assert_receive {:hunter_stream, ^pid, :reconnected} + refute_receive {:ws_frame, %{"type" => "subscribe", "stream" => "user"}}, 100 + end + + test "close/1 stays terminal when reconnect is enabled" do + {_server, port} = Hunter.StreamingServer.start(self()) + + {:ok, pid} = + Hunter.Streaming.connect(client(), + url: "ws://localhost:#{port}", + reconnect: [initial_backoff: 10] + ) + + assert_receive {:ws_connected, _ws} + ref = Process.monitor(pid) + + assert :ok = Hunter.Streaming.close(pid) + assert_receive {:hunter_stream, ^pid, {:closed, :local}} + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + end + + test "gives up with {:closed, reason} once max_attempts is exhausted" do + {_server, port} = Hunter.StreamingServer.start(self()) + + {:ok, pid} = + Hunter.Streaming.connect(client(), + url: "ws://localhost:#{port}", + reconnect: [initial_backoff: 10, max_attempts: 2] + ) + + assert_receive {:ws_connected, ws} + ref = Process.monitor(pid) + + :ok = stop_supervised(Hunter.StreamingServer) + send(ws, {:close, 1_012}) + + assert_receive {:hunter_stream, ^pid, {:reconnecting, _reason}} + assert_receive {:hunter_stream, ^pid, {:closed, {:error, _reason}}} + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + end + end + defp client, do: Hunter.new(base_url: "https://mastodon.example", access_token: "123456") end diff --git a/test/integration/mastodon_test.exs b/test/integration/mastodon_test.exs index 1945b28..3466cc8 100644 --- a/test/integration/mastodon_test.exs +++ b/test/integration/mastodon_test.exs @@ -543,6 +543,36 @@ defmodule Hunter.Integration.MastodonTest do assert claims["preferred_username"] == "hunter" end + describe "streaming" do + test "health check and live user-stream update", %{conn: conn} do + assert Hunter.Streaming.health?(conn) + + {:ok, pid} = + Hunter.Streaming.connect(conn, + streams: ["user"], + transport_opts: [verify: :verify_none] + ) + + status = Hunter.create_status(conn, "streaming test #{System.unique_integer([:positive])}") + + try do + status_id = status.id + + assert_receive {:hunter_stream, ^pid, + %Hunter.Streaming.Event{ + type: "update", + payload: %Hunter.Status{id: ^status_id} + }}, + 30_000 + + Hunter.Streaming.close(pid) + assert_receive {:hunter_stream, ^pid, {:closed, :local}}, 5_000 + after + Hunter.destroy_status(conn, status.id) + end + end + end + # Failure-path cleanup: on_exit nets that tolerate state already removed by # the test body's own assertions. defp destroy_quietly(conn, id) do @@ -602,34 +632,4 @@ defmodule Hunter.Integration.MastodonTest do rescue Hunter.Error -> :ok end - - describe "streaming" do - test "health check and live user-stream update", %{conn: conn} do - assert Hunter.Streaming.health?(conn) - - {:ok, pid} = - Hunter.Streaming.connect(conn, - streams: ["user"], - transport_opts: [verify: :verify_none] - ) - - status = Hunter.create_status(conn, "streaming test #{System.unique_integer([:positive])}") - - try do - status_id = status.id - - assert_receive {:hunter_stream, ^pid, - %Hunter.Streaming.Event{ - type: "update", - payload: %Hunter.Status{id: ^status_id} - }}, - 30_000 - - Hunter.Streaming.close(pid) - assert_receive {:hunter_stream, ^pid, {:closed, :local}}, 5_000 - after - Hunter.destroy_status(conn, status.id) - end - end - end end diff --git a/test/support/raw_streaming_server.ex b/test/support/raw_streaming_server.ex new file mode 100644 index 0000000..d17a250 --- /dev/null +++ b/test/support/raw_streaming_server.ex @@ -0,0 +1,61 @@ +defmodule Hunter.RawStreamingServer do + @moduledoc """ + Bare `:gen_tcp` WebSocket server for framing-level failure tests. + + Performs the HTTP upgrade handshake and then writes whatever bytes the + test scripts — desynced framing that a conformant WebSock server (such + as `Hunter.StreamingServer`) cannot be made to produce. + + `start/1` returns `{pid, port}`; once a client upgrades, the test + process receives `{:raw_ws_connected, pid}` and can message the pid: + + * `{:push_raw, binary}` - write raw bytes on the socket + + """ + + @ws_magic "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + @doc """ + Boots the listener on a random loopback port and accepts one client; + returns `{pid, port}`. + """ + def start(test_pid) do + {:ok, listen} = :gen_tcp.listen(0, [:binary, ip: :loopback, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + pid = spawn_link(fn -> accept(listen, test_pid) end) + {pid, port} + end + + defp accept(listen, test_pid) do + {:ok, socket} = :gen_tcp.accept(listen) + {:ok, request} = recv_request(socket, "") + :ok = :gen_tcp.send(socket, handshake_response(request)) + send(test_pid, {:raw_ws_connected, self()}) + loop(socket) + end + + defp recv_request(socket, acc) do + {:ok, data} = :gen_tcp.recv(socket, 0) + acc = acc <> data + + if String.contains?(acc, "\r\n\r\n"), do: {:ok, acc}, else: recv_request(socket, acc) + end + + defp handshake_response(request) do + [_line, key] = Regex.run(~r/sec-websocket-key:\s*(\S+)/i, request) + accept = Base.encode64(:crypto.hash(:sha, key <> @ws_magic)) + + "HTTP/1.1 101 Switching Protocols\r\n" <> + "upgrade: websocket\r\n" <> + "connection: upgrade\r\n" <> + "sec-websocket-accept: #{accept}\r\n\r\n" + end + + defp loop(socket) do + receive do + {:push_raw, data} -> + :ok = :gen_tcp.send(socket, data) + loop(socket) + end + end +end diff --git a/test/support/streaming_server.ex b/test/support/streaming_server.ex index e85afc4..fd589f0 100644 --- a/test/support/streaming_server.ex +++ b/test/support/streaming_server.ex @@ -31,14 +31,18 @@ defmodule Hunter.StreamingServer do end @doc """ - Starts the server under the test supervisor via `start_supervised!/1`; - returns `{pid, port}`. + Starts the server under the test supervisor via `start_supervised!/1` + with `#{inspect(__MODULE__)}` as the child id (so tests can + `stop_supervised/1` it); returns `{pid, port}`. """ def start(test_pid) do server = ExUnit.Callbacks.start_supervised!( - {Bandit, - plug: {__MODULE__, test_pid: test_pid}, port: 0, ip: :loopback, startup_log: false} + Supervisor.child_spec( + {Bandit, + plug: {__MODULE__, test_pid: test_pid}, port: 0, ip: :loopback, startup_log: false}, + id: __MODULE__ + ) ) {:ok, {_ip, port}} = ThousandIsland.listener_info(server) From 06bf27a78290e2d2e639cd8283ddeb25cbcd1c67 Mon Sep 17 00:00:00 2001 From: Milton Mazzarri Date: Thu, 16 Jul 2026 01:39:16 -0500 Subject: [PATCH 2/2] Close the fresh Mint conn when a reconnect's resubscribe fails From the PR review: in handle_info(:reconnect, ...) the with/else could not see the state bound by establish/1, so a handshake that succeeded but whose subscription replay failed leaked the just-opened connection on every attempt (and init/1 had the same one-shot leak). send_subscriptions/1 now returns {:error, state, reason} so both callers can close the live connection via a shared drop_connection/1, which disconnect/3 also reuses along with the backoff/2 helper instead of pattern-matching initial_backoff by hand. New test covers close/1 during the backoff window. Co-Authored-By: Claude Fable 5 --- lib/hunter/streaming/connection.ex | 53 +++++++++++++++++++++--------- test/hunter/streaming_test.exs | 20 +++++++++++ 2 files changed, 57 insertions(+), 16 deletions(-) diff --git a/lib/hunter/streaming/connection.ex b/lib/hunter/streaming/connection.ex index 71be035..32515e1 100644 --- a/lib/hunter/streaming/connection.ex +++ b/lib/hunter/streaming/connection.ex @@ -50,11 +50,19 @@ defmodule Hunter.Streaming.Connection do subscriptions: Enum.map(streams, &normalize_stream_spec/1) } - with {:ok, state} <- establish(state), - {:ok, state} <- send_subscriptions(state) do - {:ok, state} - else - {:error, reason} -> {:error, reason} + case establish(state) do + {:ok, state} -> + case send_subscriptions(state) do + {:ok, state} -> + {:ok, state} + + {:error, state, reason} -> + drop_connection(state) + {:error, reason} + end + + {:error, reason} -> + {:error, reason} end end @@ -90,12 +98,21 @@ defmodule Hunter.Streaming.Connection do @impl GenServer 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) + case establish(state) do + {:ok, state} -> + case send_subscriptions(state) do + {:ok, state} -> + send(state.subscriber, {:hunter_stream, self(), :reconnected}) + {:noreply, %{state | attempts: 0}} + + # The fresh connection is unusable; close it before counting + # the failed attempt. + {:error, state, reason} -> + retry_or_stop(drop_connection(state), reason) + end + + {:error, reason} -> + retry_or_stop(state, reason) end end @@ -220,7 +237,7 @@ defmodule Hunter.Streaming.Connection do case send_frame(state, {:text, Poison.encode!(frame)}) do {:ok, state} -> {:cont, {:ok, state}} - {:error, _state, reason} -> {:halt, {:error, reason}} + {:error, state, reason} -> {:halt, {:error, state, reason}} end end) end @@ -247,11 +264,10 @@ defmodule Hunter.Streaming.Connection do nil -> stop_with(state, reason, reply_or_noreply) - %{initial_backoff: backoff} -> + reconnect -> send(state.subscriber, {:hunter_stream, self(), {:reconnecting, reason}}) - if state.conn, do: Mint.HTTP.close(state.conn) - state = %{state | conn: nil, websocket: nil, ref: nil, attempts: 0} - Process.send_after(self(), :reconnect, backoff) + state = %{drop_connection(state) | attempts: 0} + Process.send_after(self(), :reconnect, backoff(reconnect, 0)) case reply_or_noreply do {:reply, value} -> {:reply, value, state} @@ -260,6 +276,11 @@ defmodule Hunter.Streaming.Connection do end end + defp drop_connection(state) do + if state.conn, do: Mint.HTTP.close(state.conn) + %{state | conn: nil, websocket: nil, ref: nil} + end + defp retry_or_stop(state, reason) do attempts = state.attempts + 1 diff --git a/test/hunter/streaming_test.exs b/test/hunter/streaming_test.exs index 3d70646..50e4285 100644 --- a/test/hunter/streaming_test.exs +++ b/test/hunter/streaming_test.exs @@ -249,6 +249,26 @@ defmodule Hunter.StreamingTest do assert_receive {:DOWN, ^ref, :process, ^pid, :normal} end + test "close/1 during the backoff window is terminal and skips the reconnect" do + {_server, port} = Hunter.StreamingServer.start(self()) + + {:ok, pid} = + Hunter.Streaming.connect(client(), + url: "ws://localhost:#{port}", + reconnect: [initial_backoff: 60_000] + ) + + assert_receive {:ws_connected, _ws} + ref = Process.monitor(pid) + + :ok = stop_supervised(Hunter.StreamingServer) + assert_receive {:hunter_stream, ^pid, {:reconnecting, _reason}} + + assert :ok = Hunter.Streaming.close(pid) + assert_receive {:hunter_stream, ^pid, {:closed, :local}} + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + end + test "gives up with {:closed, reason} once max_attempts is exhausted" do {_server, port} = Hunter.StreamingServer.start(self())