diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8afd9cd..44d253a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ jobs: uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 with: repository: ArchAstro/archastro-openapi - ref: ${{ github.event_name == 'pull_request' && 'features/calvin-archastro-06-08-2026-elixir-sdk' || 'main' }} + ref: main path: .ci/archastro-openapi - name: Setup Node (for prism + @archastro/channel-harness) diff --git a/lib/archastro/channel.ex b/lib/archastro/channel.ex index d0404b9..c202fe7 100644 --- a/lib/archastro/channel.ex +++ b/lib/archastro/channel.ex @@ -21,8 +21,13 @@ defmodule ArchAstro.SDK.Channel do ArchAstro.SDK.Codec.descriptor(), timeout() ) :: {:ok, t()} | {:error, ArchAstro.SDK.Error.reason()} - def join(socket, topic, payload, module, descriptor, timeout \\ 5_000), - do: GenServer.call(socket, {:archastro_join, topic, payload, module, descriptor}, timeout) + # Payloads are encoded here rather than in the socket process: an + # unencodable value must fail its own caller, not take down the shared + # socket and every other channel riding on it. + def join(socket, topic, payload, module, descriptor, timeout \\ 5_000) do + encoded = ArchAstro.SDK.Codec.encode(payload) + GenServer.call(socket, {:archastro_join, topic, encoded, module, descriptor}, timeout) + end @spec push( t(), @@ -31,13 +36,15 @@ defmodule ArchAstro.SDK.Channel do ArchAstro.SDK.Codec.descriptor(), timeout() ) :: {:ok, ArchAstro.SDK.JSON.t() | struct() | :ok} | {:error, ArchAstro.SDK.Error.reason()} - def push(%__MODULE__{} = channel, event, payload, descriptor, timeout \\ 5_000), - do: - GenServer.call( - channel.socket, - {:archastro_push, channel.topic, event, payload, descriptor}, - timeout - ) + def push(%__MODULE__{} = channel, event, payload, descriptor, timeout \\ 5_000) do + encoded = ArchAstro.SDK.Codec.encode(payload) + + GenServer.call( + channel.socket, + {:archastro_push, channel.topic, event, encoded, descriptor}, + timeout + ) + end @spec subscribe(t(), String.t(), pid(), ArchAstro.SDK.Codec.descriptor()) :: :ok def subscribe(%__MODULE__{} = channel, event, subscriber, descriptor) diff --git a/lib/archastro/client.ex b/lib/archastro/client.ex index f7f601a..969efc6 100644 --- a/lib/archastro/client.ex +++ b/lib/archastro/client.ex @@ -43,7 +43,7 @@ defmodule ArchAstro.SDK.Client do opts |> Keyword.get(:base_url, ArchAstro.SDK.GeneratedClient.default_base_url()) |> String.trim_trailing("/"), - request: Keyword.get(opts, :req, Req.new(retry: false)), + request: Keyword.get(opts, :req, Req.new(retry: false, redirect: false)), token_binding: binding, telemetry_metadata: Keyword.get(opts, :telemetry_metadata, %{}) }} diff --git a/lib/archastro/codec.ex b/lib/archastro/codec.ex index 58d5016..445ae9e 100644 --- a/lib/archastro/codec.ex +++ b/lib/archastro/codec.ex @@ -73,8 +73,21 @@ defmodule ArchAstro.SDK.Codec do def encode(%DateTime{} = value), do: DateTime.to_iso8601(value) def encode(%Date{} = value), do: Date.to_iso8601(value) + def encode(%NaiveDateTime{} = value) do + raise ArgumentError, + "cannot encode NaiveDateTime #{inspect(value)}: the API requires an " <> + "ISO 8601 datetime with a UTC offset; convert it with " <> + "DateTime.from_naive!(value, \"Etc/UTC\") first" + end + + def encode(%Time{} = value) do + raise ArgumentError, + "cannot encode Time #{inspect(value)}: the API has no time-of-day wire " <> + "format; send a full DateTime instead" + end + def encode(%module{} = value) do - if function_exported?(module, :to_map, 1), + if Code.ensure_loaded?(module) and function_exported?(module, :to_map, 1), do: module.to_map(value), else: Map.from_struct(value) end diff --git a/lib/archastro/http.ex b/lib/archastro/http.ex index a9aaaf0..ff30ca7 100644 --- a/lib/archastro/http.ex +++ b/lib/archastro/http.ex @@ -38,6 +38,11 @@ defmodule ArchAstro.SDK.HTTP do retry: false ] + request_opts = + if Keyword.get(opts, :raw, false), + do: Keyword.put(request_opts, :decode_body, false), + else: request_opts + request_opts = case Keyword.fetch(opts, :body) do {:ok, value} -> Keyword.put(request_opts, :json, Codec.encode(value)) @@ -70,7 +75,7 @@ defmodule ArchAstro.SDK.HTTP do {:ok, decoded} {:ok, response} -> - error = Error.from_response(response) + error = response |> decode_error_body() |> Error.from_response() :telemetry.execute( [:archastro, :request, :stop], @@ -92,4 +97,15 @@ defmodule ArchAstro.SDK.HTTP do {:error, error} end end + + # Raw requests suppress Req's body decoding, which would otherwise strip the + # structured error envelope off a failure response too. + defp decode_error_body(%Req.Response{body: body} = response) when is_binary(body) do + case Jason.decode(body) do + {:ok, decoded} -> %{response | body: decoded} + {:error, _reason} -> response + end + end + + defp decode_error_body(response), do: response end diff --git a/lib/archastro/query.ex b/lib/archastro/query.ex index cf542d3..0b8b046 100644 --- a/lib/archastro/query.ex +++ b/lib/archastro/query.ex @@ -18,14 +18,21 @@ defmodule ArchAstro.SDK.Query do value |> ArchAstro.SDK.Codec.encode() |> Enum.flat_map(fn - {_key, :__archastro_unset__} -> [] - {key, values} when is_list(values) -> Enum.map(values, &{key, scalar(&1)}) - {key, item} -> [{key, scalar(item)}] + # A nil param means "not provided", never the string "null". The + # sentinel still arrives un-rewritten from Codec.encode's + # Map.from_struct fallback (structs without to_map/1). + {_key, value} when value in [nil, :__archastro_unset__] -> + [] + + {key, values} when is_list(values) -> + for value <- values, value not in [nil, :__archastro_unset__], do: {key, scalar(value)} + + {key, item} -> + [{key, scalar(item)}] end) |> URI.encode_query() end - defp scalar(nil), do: "null" defp scalar(value) when is_binary(value), do: value defp scalar(value) when is_boolean(value) or is_number(value), do: to_string(value) defp scalar(value) when is_map(value) or is_list(value), do: Jason.encode!(value) diff --git a/lib/archastro/socket.ex b/lib/archastro/socket.ex index cd2ad02..1c4dea0 100644 --- a/lib/archastro/socket.ex +++ b/lib/archastro/socket.ex @@ -12,6 +12,13 @@ defmodule ArchAstro.SDK.Socket do alias ArchAstro.SDK.{Channel, Client, Codec, Error, TokenServer} + # Pushes and joins the server never acknowledges (fire-and-forget events, + # callers that give up) would otherwise accumulate forever on long-lived + # sockets. Entries are swept once they are older than a full interval, so + # the effective grace period is one to two intervals — far above any sane + # Channel call timeout (default 5s). + @pending_sweep_interval_msec 60_000 + @type option :: {:name, GenServer.name()} | {:socket_path, String.t()} @@ -29,6 +36,8 @@ defmodule ArchAstro.SDK.Socket do @impl Slipstream def init({client, opts}) do with {:ok, config} <- connection_config(client, opts) do + Process.send_after(self(), :archastro_sweep_pending, @pending_sweep_interval_msec) + socket = new_socket() |> assign( @@ -108,9 +117,15 @@ defmodule ArchAstro.SDK.Socket do end def handle_call({:archastro_push, topic, event, payload, descriptor}, from, socket) do - case push(socket, topic, event, Codec.encode(payload)) do + case push(socket, topic, event, payload) do {:ok, ref} -> - pending = %{from: from, topic: topic, descriptor: descriptor} + pending = %{ + from: from, + topic: topic, + descriptor: descriptor, + inserted_at: System.monotonic_time(:millisecond) + } + {:noreply, update(socket, :pending_pushes, &Map.put(&1, ref, pending))} {:error, reason} -> @@ -175,9 +190,10 @@ defmodule ArchAstro.SDK.Socket do defp queue_join(socket, topic, payload, module, descriptor, from, nil) do pending = %{ froms: [from], - payload: Codec.encode(payload), + payload: payload, module: module, - descriptor: descriptor + descriptor: descriptor, + inserted_at: System.monotonic_time(:millisecond) } update(socket, :pending_joins, &Map.put(&1, topic, pending)) @@ -208,22 +224,41 @@ defmodule ArchAstro.SDK.Socket do def handle_join(topic, response, socket) do case Map.pop(socket.assigns.pending_joins, topic) do {nil, pending_joins} -> - {:ok, assign(socket, :pending_joins, pending_joins)} + socket = assign(socket, :pending_joins, pending_joins) + + # No waiter and no established channel means the join outlived + # everyone who wanted it (swept, or its callers gave up). Leaving + # returns the topic to a closed state; stranding it as joined-but- + # unowned would make every later join a silent no-op. + if Map.has_key?(socket.assigns.channels, topic) do + {:ok, socket} + else + {:ok, if(joined?(socket, topic), do: leave(socket, topic), else: socket)} + end {pending, pending_joins} -> - channel = %Channel{ - socket: self(), - topic: topic, - module: pending.module, - join_response: Codec.decode(response, pending.descriptor) - } - - Enum.each(pending.froms, &GenServer.reply(&1, {:ok, channel})) - - {:ok, - socket - |> assign(:pending_joins, pending_joins) - |> update(:channels, &Map.put(&1, topic, channel))} + case safe_decode(response, pending.descriptor, %{topic: topic, kind: :join}) do + {:ok, join_response} -> + channel = %Channel{ + socket: self(), + topic: topic, + module: pending.module, + join_response: join_response + } + + Enum.each(pending.froms, &GenServer.reply(&1, {:ok, channel})) + + {:ok, + socket + |> assign(:pending_joins, pending_joins) + |> update(:channels, &Map.put(&1, topic, channel))} + + {:error, error} -> + Enum.each(pending.froms, &GenServer.reply(&1, {:error, error})) + socket = assign(socket, :pending_joins, pending_joins) + socket = if joined?(socket, topic), do: leave(socket, topic), else: socket + {:ok, socket} + end end end @@ -234,7 +269,7 @@ defmodule ArchAstro.SDK.Socket do {:ok, assign(socket, :pending_pushes, pending)} {request, pending} -> - GenServer.reply(request.from, decode_reply(reply, request.descriptor)) + GenServer.reply(request.from, decode_reply(reply, request)) {:ok, assign(socket, :pending_pushes, pending)} end end @@ -244,13 +279,55 @@ defmodule ArchAstro.SDK.Socket do channel = socket.assigns.channels[topic] Enum.each(socket.assigns.subscriptions[{topic, event}] || [], fn subscription -> - decoded = Codec.decode(message, subscription.descriptor) - send(subscription.subscriber, {:archastro_channel, channel, event, decoded}) + case safe_decode(message, subscription.descriptor, %{ + topic: topic, + event: event, + kind: :message + }) do + {:ok, decoded} -> + send(subscription.subscriber, {:archastro_channel, channel, event, decoded}) + + {:error, _error} -> + :ok + end end) {:ok, socket} end + @impl Slipstream + def handle_info(:archastro_sweep_pending, socket) do + Process.send_after(self(), :archastro_sweep_pending, @pending_sweep_interval_msec) + cutoff = System.monotonic_time(:millisecond) - @pending_sweep_interval_msec + + {expired_pushes, live_pushes} = + Enum.split_with(socket.assigns.pending_pushes, fn {_ref, request} -> + request.inserted_at <= cutoff + end) + + Enum.each(expired_pushes, fn {_ref, request} -> + GenServer.reply(request.from, {:error, no_reply_error("push")}) + end) + + {expired_joins, live_joins} = + Enum.split_with(socket.assigns.pending_joins, fn {_topic, pending} -> + pending.inserted_at <= cutoff + end) + + Enum.each(expired_joins, fn {_topic, pending} -> + Enum.each(pending.froms, &GenServer.reply(&1, {:error, no_reply_error("join")})) + end) + + {:noreply, + socket + |> assign(:pending_pushes, Map.new(live_pushes)) + |> assign(:pending_joins, Map.new(live_joins))} + end + + defp no_reply_error(kind) do + %Error{message: "channel #{kind} was never acknowledged", code: "no_reply"} + end + @impl Slipstream def handle_topic_close(topic, reason, socket) do error = Error.channel(reason) @@ -362,14 +439,35 @@ defmodule ArchAstro.SDK.Socket do end) end - defp decode_reply({:ok, value}, descriptor), do: {:ok, Codec.decode(value, descriptor)} - defp decode_reply({:error, reason}, _descriptor), do: {:error, Error.channel(reason)} + defp decode_reply({:ok, value}, request), do: safe_decode_reply(value, request) + defp decode_reply({:error, reason}, _request), do: {:error, Error.channel(reason)} + + # Slipstream collapses a payload-less `{:reply, :error, socket}` to the bare + # atom :error; it is an error verdict, not a payload to decode. + defp decode_reply(:error, _request), do: {:error, Error.channel(%{})} - defp decode_reply(%{"status" => "ok", "response" => value}, descriptor), - do: {:ok, Codec.decode(value, descriptor)} + defp decode_reply(%{"status" => "ok", "response" => value}, request), + do: safe_decode_reply(value, request) - defp decode_reply(%{"status" => "error", "response" => reason}, _descriptor), + defp decode_reply(%{"status" => "error", "response" => reason}, _request), do: {:error, Error.channel(reason)} - defp decode_reply(value, descriptor), do: {:ok, Codec.decode(value, descriptor)} + defp decode_reply(value, request), do: safe_decode_reply(value, request) + + defp safe_decode_reply(value, request), + do: safe_decode(value, request.descriptor, %{topic: request.topic, kind: :reply}) + + defp safe_decode(value, descriptor, metadata) do + {:ok, Codec.decode(value, descriptor)} + rescue + exception -> + :telemetry.execute([:archastro, :channel, :decode_failure], %{}, metadata) + + {:error, + %Error{ + message: "ArchAstro channel payload failed to decode", + code: "decode_failure", + reason: exception + }} + end end diff --git a/lib/archastro/token_server/default.ex b/lib/archastro/token_server/default.ex index 59f034d..a10346e 100644 --- a/lib/archastro/token_server/default.ex +++ b/lib/archastro/token_server/default.ex @@ -89,7 +89,7 @@ defmodule ArchAstro.SDK.TokenServer.Default do table: table, base_url: String.trim_trailing(Keyword.get(opts, :base_url, "https://platform.archastro.ai"), "/"), - req: Keyword.get(opts, :req, Req.new(retry: false)), + req: Keyword.get(opts, :req, Req.new(retry: false, redirect: false)), store: opts[:store], inflight: %{} }} diff --git a/test/codec_test.exs b/test/codec_test.exs index 0aac0b3..e0e9446 100644 --- a/test/codec_test.exs +++ b/test/codec_test.exs @@ -82,4 +82,16 @@ defmodule ArchAstro.SDK.CodecTest do descriptor = {:union, [:datetime, {:enum, ["not-a-date"]}]} assert "not-a-date" = ArchAstro.SDK.Codec.decode("not-a-date", descriptor) end + + test "encoding a NaiveDateTime raises a clear error instead of leaking a struct" do + assert_raise ArgumentError, ~r/NaiveDateTime.*DateTime/s, fn -> + ArchAstro.SDK.Codec.encode(%{"scheduled_at" => ~N[2026-08-16 12:00:00]}) + end + end + + test "encoding a Time raises a clear error instead of leaking a struct" do + assert_raise ArgumentError, ~r/Time/, fn -> + ArchAstro.SDK.Codec.encode(~T[12:00:00]) + end + end end diff --git a/test/codec_unloaded_modules_test.exs b/test/codec_unloaded_modules_test.exs new file mode 100644 index 0000000..532e99b --- /dev/null +++ b/test/codec_unloaded_modules_test.exs @@ -0,0 +1,79 @@ +# Copyright (c) 2026 ArchAstro Inc. Licensed under the MIT License. + +defmodule ArchAstro.SDK.CodecUnloadedModulesTest do + # Purging module code is VM-global state; keep this suite out of the + # concurrent test phase. + use ExUnit.Case, async: false + + alias ArchAstro.SDK.Codec + + @sentinel :__archastro_unset__ + + test "generated input/params structs encode identically when their modules are not loaded" do + modules = generated_input_modules() + assert length(modules) > 150 + + for module <- modules do + {:module, ^module} = Code.ensure_loaded(module) + value = struct(module) + expected = Codec.encode(value) + + unload(module) + actual = Codec.encode(value) + + assert actual == expected, + "#{inspect(module)} encoded differently when its module was not loaded: " <> + "#{inspect(actual)} vs #{inspect(expected)}" + + assert_wire_clean(actual, module) + {:module, ^module} = Code.ensure_loaded(module) + end + end + + test "renamed fields keep their wire names when the module is not loaded" do + module = ArchAstro.SDK.Types.Operations.PostApiV1AuthRegister.AuthInput + {:module, ^module} = Code.ensure_loaded(module) + value = struct(module, alias_: "my-alias") + + unload(module) + encoded = Codec.encode(value) + {:module, ^module} = Code.ensure_loaded(module) + + assert encoded["alias"] == "my-alias" + refute Map.has_key?(encoded, "alias_") + refute Map.has_key?(encoded, :alias_) + end + + defp generated_input_modules do + {:ok, modules} = :application.get_key(:archastro, :modules) + + Enum.filter(modules, fn module -> + name = Atom.to_string(module) + + String.starts_with?(name, "Elixir.ArchAstro.SDK.") and + Regex.match?(~r/\.(?:[A-Za-z0-9]*Input|Params)$/, name) + end) + end + + defp unload(module) do + :code.purge(module) + :code.delete(module) + refute function_exported?(module, :to_map, 1) + end + + defp assert_wire_clean(value, module) when is_map(value) do + Enum.each(value, fn {key, item} -> + assert is_binary(key), "#{inspect(module)} leaked non-string map key #{inspect(key)}" + assert_wire_clean(item, module) + end) + end + + defp assert_wire_clean(value, module) when is_list(value) do + Enum.each(value, &assert_wire_clean(&1, module)) + end + + defp assert_wire_clean(value, module) do + refute value in [@sentinel, "__archastro_unset__"], + "#{inspect(module)} leaked the unset sentinel onto the wire" + end +end diff --git a/test/http_test.exs b/test/http_test.exs new file mode 100644 index 0000000..aa794b6 --- /dev/null +++ b/test/http_test.exs @@ -0,0 +1,76 @@ +# Copyright (c) 2026 ArchAstro Inc. Licensed under the MIT License. + +defmodule ArchAstro.SDK.HTTPTest do + use ExUnit.Case, async: true + + alias ArchAstro.SDK.{Client, TokenServer} + + defp client(plug) do + server = + start_supervised!( + {TokenServer.Default, mode: {:system_user, publishable_key: "pk", access_token: "token"}} + ) + + {:ok, client} = Client.for_server(server, req: Req.new(plug: plug)) + client + end + + defp json_plug(body) do + fn conn -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.send_resp(200, body) + end + end + + test "raw requests deliver an application/json body as the raw binary" do + raw_body = ~s({"steps":[{"n":1}],"note":"blob"}) + client = client(json_plug(raw_body)) + + assert {:ok, %Req.Response{} = response} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/trajectories/tra_1/contents", + raw: true + ) + + assert response.body == raw_body + end + + test "raw requests keep structured API errors on failure responses" do + error_plug = fn conn -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.send_resp( + 404, + ~s({"error":{"message":"Trajectory not found","code":"not_found"}}) + ) + end + + client = client(error_plug) + + assert {:error, %ArchAstro.SDK.Error{} = error} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/trajectories/tra_1/contents", + raw: true + ) + + assert error.status == 404 + assert error.message == "Trajectory not found" + assert error.code == "not_found" + end + + test "raw requests tolerate a non-JSON failure body" do + error_plug = fn conn -> Plug.Conn.send_resp(conn, 502, "upstream exploded") end + client = client(error_plug) + + assert {:error, %ArchAstro.SDK.Error{status: 502} = error} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/files/file_1/avatar", raw: true) + + assert error.message == "ArchAstro API returned HTTP 502" + end + + test "typed requests still auto-decode application/json bodies" do + client = client(json_plug(~s({"note":"blob"}))) + + assert {:ok, %{"note" => "blob"}} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/things/thing_1", decode: :unknown) + end +end diff --git a/test/query_test.exs b/test/query_test.exs index 8e444d3..78ea8f2 100644 --- a/test/query_test.exs +++ b/test/query_test.exs @@ -32,4 +32,25 @@ defmodule ArchAstro.SDK.QueryTest do assert ArchAstro.SDK.Query.append("https://example.test/path", %Params{tags: []}) == "https://example.test/path" end + + test "omits nil params instead of sending the string null" do + assert ArchAstro.SDK.Query.encode(%{"cursor" => nil, "limit" => 5}) == "limit=5" + + assert ArchAstro.SDK.Query.append("https://example.test/path", %{"cursor" => nil}) == + "https://example.test/path" + end + + test "omits nil elements inside repeated params" do + assert ArchAstro.SDK.Query.encode(%{"tags" => ["one", nil, "two"]}) == "tags=one&tags=two" + end + + defmodule SentinelParams do + # A params struct with no to_map/1: Codec.encode falls back to + # Map.from_struct, which does not rewrite unset sentinels. + defstruct limit: :__archastro_unset__, cursor: :__archastro_unset__ + end + + test "omits unset sentinels that reach encoding un-rewritten" do + assert ArchAstro.SDK.Query.encode(%SentinelParams{limit: 5}) == "limit=5" + end end diff --git a/test/socket_test.exs b/test/socket_test.exs index abf939f..a9c9ee0 100644 --- a/test/socket_test.exs +++ b/test/socket_test.exs @@ -212,4 +212,260 @@ defmodule ArchAstro.SDK.SocketTest do assert left_socket.assigns.pending_leaves == %{} refute Map.has_key?(left_socket.assigns.channels, "room:1") end + + describe "outbound payload encoding" do + test "encode failures raise in the caller process, never inside the socket" do + test_pid = self() + # A stand-in socket that reports anything it receives and never replies, + # so a payload reaching the socket process is visible as a message. + socket = spawn_link(fn -> relay_forever(test_pid) end) + channel = %ArchAstro.SDK.Channel{socket: socket, topic: "room:1", module: __MODULE__} + bad_payload = %{"scheduled_at" => ~N[2026-08-16 12:00:00]} + + assert_raise ArgumentError, ~r/NaiveDateTime/, fn -> + ArchAstro.SDK.Channel.push(channel, "evt", bad_payload, :string, 200) + end + + assert_raise ArgumentError, ~r/NaiveDateTime/, fn -> + ArchAstro.SDK.Channel.join(socket, "room:1", bad_payload, __MODULE__, :string, 200) + end + + refute_receive {:socket_received, _message}, 50 + end + end + + defp relay_forever(test_pid) do + receive do + message -> + send(test_pid, {:socket_received, message}) + relay_forever(test_pid) + end + end + + describe "pending entry sweep" do + test "sweeps pushes and joins that were never acknowledged, keeping fresh ones" do + stale_push_tag = make_ref() + fresh_push_tag = make_ref() + stale_join_tag = make_ref() + + now = System.monotonic_time(:millisecond) + stale = now - 200_000 + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:subscriptions, %{}) + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{ + from: {self(), stale_push_tag}, + topic: "room:1", + descriptor: :string, + inserted_at: stale + }, + "2" => %{ + from: {self(), fresh_push_tag}, + topic: "room:1", + descriptor: :string, + inserted_at: now + } + }) + |> Slipstream.Socket.assign(:pending_joins, %{ + "room:2" => %{ + froms: [{self(), stale_join_tag}], + payload: %{}, + module: __MODULE__, + descriptor: :string, + inserted_at: stale + } + }) + + assert {:noreply, swept} = + ArchAstro.SDK.Socket.handle_info(:archastro_sweep_pending, socket) + + assert Map.keys(swept.assigns.pending_pushes) == ["2"] + assert swept.assigns.pending_joins == %{} + + assert_receive {^stale_push_tag, {:error, %ArchAstro.SDK.Error{code: "no_reply"}}} + assert_receive {^stale_join_tag, {:error, %ArchAstro.SDK.Error{code: "no_reply"}}} + refute_received {^fresh_push_tag, _reply} + end + + test "new pushes and queued joins carry an insertion timestamp" do + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + |> Slipstream.Socket.assign(:pending_leaves, %{"room:1" => true}) + + assert {:noreply, queued} = + ArchAstro.SDK.Socket.handle_call( + {:archastro_join, "room:1", %{}, __MODULE__, {:object, []}}, + {self(), make_ref()}, + socket + ) + + assert is_integer(queued.assigns.pending_joins["room:1"].inserted_at) + end + end + + describe "join responses with no waiter" do + test "a join acked after its waiters are gone leaves the topic instead of stranding it" do + socket = + Slipstream.Socket.new() + |> Map.put(:channel_pid, self()) + |> Map.put(:joins, %{ + "room:1" => %Slipstream.Socket.Join{ + topic: "room:1", + params: %{}, + status: :joined, + rejoin_counter: 0 + } + }) + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + + assert {:ok, _socket} = ArchAstro.SDK.Socket.handle_join("room:1", %{}, socket) + + assert_receive {:__slipstream_command__, %Slipstream.Commands.LeaveTopic{topic: "room:1"}} + end + + test "an established topic rejoined after reconnect is kept, not left" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Map.put(:channel_pid, self()) + |> Map.put(:joins, %{ + "room:1" => %Slipstream.Socket.Join{ + topic: "room:1", + params: %{}, + status: :joined, + rejoin_counter: 1 + } + }) + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + + assert {:ok, kept} = ArchAstro.SDK.Socket.handle_join("room:1", %{}, socket) + + assert Map.has_key?(kept.assigns.channels, "room:1") + refute_received {:__slipstream_command__, %Slipstream.Commands.LeaveTopic{}} + end + end + + describe "decode failures do not crash the socket" do + setup do + handler_id = "socket-decode-failure-#{inspect(make_ref())}" + + :telemetry.attach( + handler_id, + [:archastro, :channel, :decode_failure], + fn event, _measurements, metadata, pid -> send(pid, {:telemetry, event, metadata}) end, + self() + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + :ok + end + + test "a malformed broadcast is dropped with telemetry instead of raising" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:subscriptions, %{ + {"room:1", "evt"} => [%{subscriber: self(), descriptor: :string}] + }) + + assert {:ok, _socket} = + ArchAstro.SDK.Socket.handle_message("room:1", "evt", %{"bad" => true}, socket) + + refute_received {:archastro_channel, _channel, _event, _payload} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", event: "evt", kind: :message}} + end + + test "a well-formed broadcast is still delivered" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:subscriptions, %{ + {"room:1", "evt"} => [%{subscriber: self(), descriptor: :string}] + }) + + assert {:ok, _socket} = + ArchAstro.SDK.Socket.handle_message("room:1", "evt", "hello", socket) + + assert_received {:archastro_channel, ^channel, "evt", "hello"} + end + + test "an undecodable reply fails only that push" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{from: {self(), tag}, topic: "room:1", descriptor: :string} + }) + + assert {:ok, updated} = + ArchAstro.SDK.Socket.handle_reply( + "1", + %{"status" => "ok", "response" => 42}, + socket + ) + + assert updated.assigns.pending_pushes == %{} + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{code: "decode_failure"}}} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", kind: :reply}} + end + + test "an empty error reply resolves the push as a channel error" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{from: {self(), tag}, topic: "room:1", descriptor: :string} + }) + + assert {:ok, updated} = ArchAstro.SDK.Socket.handle_reply("1", :error, socket) + + assert updated.assigns.pending_pushes == %{} + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{} = error}} + assert error.message == "ArchAstro channel operation failed" + end + + test "an undecodable join response fails the join instead of the socket" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{ + "room:1" => %{ + froms: [{self(), tag}], + payload: %{}, + module: __MODULE__, + descriptor: :string + } + }) + + assert {:ok, updated} = + ArchAstro.SDK.Socket.handle_join("room:1", %{"bad" => true}, socket) + + assert updated.assigns.pending_joins == %{} + refute Map.has_key?(updated.assigns.channels, "room:1") + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{code: "decode_failure"}}} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", kind: :join}} + end + end end diff --git a/test/token_server_test.exs b/test/token_server_test.exs index b153a68..9bd6ecb 100644 --- a/test/token_server_test.exs +++ b/test/token_server_test.exs @@ -537,4 +537,15 @@ defmodule ArchAstro.SDK.TokenServerTest do assert {:ok, authorization} = TokenServer.authorization(client.token_binding) assert {"x-archastro-api-key", "sk_test"} in authorization.headers end + + test "default requests never follow redirects" do + # A 3xx from the platform is always wrong; following it can silently turn a + # token-refresh POST into a body-less GET against the Location target. + server = start_supervised!({TokenServer.Default, mode: {:secret_key, "sk_test"}}) + + assert :sys.get_state(server).req.options[:redirect] == false + + assert {:ok, client} = Client.for_server(server) + assert client.request.options[:redirect] == false + end end