From b5215b9fc1f57b35d711b03f612b9b58fd4c97f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 06:50:17 +0200 Subject: [PATCH 01/10] Stop masking unrelated FunctionClauseError in mutations `Storex.Store.__mutation__/6` rescued every `FunctionClauseError` raised under the store's `mutation/5` call and reported it as "No mutation matching ...". That message is only correct when the store itself has no clause for the incoming name and data. An error raised deeper inside a mutation that did match was reported the same way, with the real stacktrace discarded. The two cases are distinguishable: a `FunctionClauseError` carries the module, function and arity that failed to match. Only `{store, :mutation, 5}` means "no such mutation"; anything else is reraised with its original stacktrace. This changes observable behaviour. A crash inside a matching mutation used to reach the client as an error message; it now propagates, stops the store process and takes the socket with it. --- CHANGELOG.md | 1 + lib/storex/store.ex | 17 ++++++++++++++--- test/fixtures/stores/invalid_mutation.ex | 7 +++++++ test/storex/store_test.exs | 21 +++++++++++++++++++++ 4 files changed, 43 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea519aa..05af310 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer +- **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it - Removed the `:pg2` fallback, unreachable since OTP 24 - Updated dependencies diff --git a/lib/storex/store.ex b/lib/storex/store.ex index 4cb765d..f1a586b 100644 --- a/lib/storex/store.ex +++ b/lib/storex/store.ex @@ -60,12 +60,23 @@ defmodule Storex.Store do "Return value of mutation should be {:reply, message, state}, {:noreply, state} or {:error, error}"} end rescue - FunctionClauseError -> - {:error, - "No mutation matching #{inspect(name)} with data #{inspect(data)} in store #{inspect(store)}"} + error in FunctionClauseError -> + if unmatched_mutation?(error, store) do + {:error, + "No mutation matching #{inspect(name)} with data #{inspect(data)} in store #{inspect(store)}"} + else + reraise error, __STACKTRACE__ + end end end + # Only the store's own `mutation/5` failing to match means "no such mutation". + # Any other `FunctionClauseError` was raised deeper inside a mutation that did + # match, and has to keep its original stacktrace. + defp unmatched_mutation?(%FunctionClauseError{} = error, store) do + error.module == store and error.function == :mutation and error.arity == 5 + end + @doc false def __terminate__(store, session, params, state) do if :erlang.function_exported(store, :terminate, 3) do diff --git a/test/fixtures/stores/invalid_mutation.ex b/test/fixtures/stores/invalid_mutation.ex index 9725bbc..fde9374 100644 --- a/test/fixtures/stores/invalid_mutation.ex +++ b/test/fixtures/stores/invalid_mutation.ex @@ -12,4 +12,11 @@ defmodule StorexTest.Store.InvalidMutation do def mutation("error", _data, _session_id, _params, _state) do {:error, "Not allowed"} end + + # Matches, then raises a FunctionClauseError of its own from further down. + def mutation("raise", data, _session_id, _params, _state) do + {:noreply, %{counter: only_zero(data)}} + end + + defp only_zero(0), do: 0 end diff --git a/test/storex/store_test.exs b/test/storex/store_test.exs index 5a30786..d8ea96a 100644 --- a/test/storex/store_test.exs +++ b/test/storex/store_test.exs @@ -60,6 +60,27 @@ defmodule StorexTest.StoreTest do {:error, "No mutation matching \"unknown\" with data [1] in store StorexTest.Store.Counter"} end + + test "a FunctionClauseError raised inside a matching mutation is not swallowed" do + error = + assert_raise FunctionClauseError, fn -> + Storex.Store.__mutation__(InvalidMutation, "raise", 1, "session", %{}, %{}) + end + + assert error.function == :only_zero + assert error.arity == 1 + end + + test "the original stacktrace of a raise inside a mutation is preserved" do + stacktrace = + try do + Storex.Store.__mutation__(InvalidMutation, "raise", 1, "session", %{}, %{}) + rescue + _ -> __STACKTRACE__ + end + + assert [{InvalidMutation, :only_zero, [1], _location} | _rest] = stacktrace + end end describe "store server" do From f52ad5b36ca6890a637deda2a232cbb0a2a02fb1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 06:52:59 +0200 Subject: [PATCH 02/10] Register store processes in a Registry instead of naming them `Storex.Supervisor.name/2` built the store process name with `String.to_atom("#{session}_#{store}")`. Session ids come from `Nanoid.generate/0` and are unique per connection, so every join created a new atom, and atoms are never garbage collected. One reconnect is one new atom, which makes this reachable by ordinary traffic rather than abuse: a long-running node with enough connection churn exhausts the atom table and the VM dies. Nothing ever looked a store up by that name. `Storex.Registry` maps `{store, session}` to the pid and is what every operation actually uses. The name was only load-bearing in that registering it twice fails, which keeps a second process for the same `{session, store}` from starting. A `Registry` keyed by the `{session, store}` tuple keeps that guarantee (`start_link` still returns `{:error, {:already_started, pid}}`) and creates no atoms. --- CHANGELOG.md | 1 + lib/storex.ex | 1 + lib/storex/supervisor.ex | 12 +++++- test/storex/supervisor_test.exs | 70 +++++++++++++++++++++++++++++++++ 4 files changed, 83 insertions(+), 1 deletion(-) create mode 100644 test/storex/supervisor_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index 05af310..d8ebcde 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- Store processes are now registered through a `Registry` keyed by `{session, store}` instead of being named `:"#{session}_#{store}"`. Session ids are unique per connection, so the old naming created one permanent atom per session-store pair and could exhaust the atom table on a long-running node - Removed the `:pg2` fallback, unreachable since OTP 24 - Updated dependencies diff --git a/lib/storex.ex b/lib/storex.ex index 724c3b0..24c6e45 100644 --- a/lib/storex.ex +++ b/lib/storex.ex @@ -9,6 +9,7 @@ defmodule Storex do %{id: :pg, start: {:pg, :start_link, [Storex.PG]}}, {Storex.PG, []}, {Storex.Registry, []}, + {Registry, keys: :unique, name: Storex.StoreRegistry}, {Storex.Supervisor, []} ] diff --git a/lib/storex/supervisor.ex b/lib/storex/supervisor.ex index dac27d7..43501b1 100644 --- a/lib/storex/supervisor.ex +++ b/lib/storex/supervisor.ex @@ -15,8 +15,18 @@ defmodule Storex.Supervisor do ) end + @doc false + # The name a store process registers under. It is deliberately a `:via` tuple + # and not an atom: session ids are unique per connection, so naming processes + # `:"#{session}_#{store}"` created one permanent atom per session-store pair + # and eventually exhausted the atom table on a long-running node. + # + # Nothing looks a store up by this name — `Storex.Registry` maps to the pid + # for that. It exists so that starting the same `{session, store}` twice + # fails with `{:error, {:already_started, pid}}` instead of silently + # producing a second process. def name(session, store) do - String.to_atom("#{session}_#{store}") + {:via, Registry, {Storex.StoreRegistry, {session, store}}} end def add_store(store, session, session_pid, params \\ %{}) do diff --git a/test/storex/supervisor_test.exs b/test/storex/supervisor_test.exs new file mode 100644 index 0000000..1ea876e --- /dev/null +++ b/test/storex/supervisor_test.exs @@ -0,0 +1,70 @@ +defmodule StorexTest.SupervisorTest do + use ExUnit.Case + + @store "StorexTest.Store.Counter" + + defp session, do: "session-#{System.unique_integer([:positive])}" + + defp start_store(session) do + {:ok, _key} = Storex.Supervisor.add_store(@store, session, self(), %{}) + on_exit(fn -> Storex.Supervisor.remove_store(session, @store) end) + session + end + + describe "store process naming" do + test "a store is registered under a {session, store} key, not an atom" do + session = start_store(session()) + + assert [{pid, nil}] = Registry.lookup(Storex.StoreRegistry, {session, @store}) + assert pid == Storex.Registry.get_store_pid(@store, session) + end + + test "starting stores does not create atoms" do + # Warm up so that first-call code paths are not counted. + for _ <- 1..5, do: start_store(session()) + + sessions = for _ <- 1..200, do: session() + + atoms_before = :erlang.system_info(:atom_count) + for session <- sessions, do: start_store(session) + atoms_after = :erlang.system_info(:atom_count) + + assert atoms_after - atoms_before == 0 + end + + test "the same session and store cannot be started twice" do + session = start_store(session()) + + spec = %{ + id: StorexTest.Store.Counter.Server, + start: + {StorexTest.Store.Counter.Server, :start_link, + [[session: session, store: @store, params: %{}]]}, + restart: :transient + } + + assert {:error, {:already_started, pid}} = + DynamicSupervisor.start_child(Storex.Supervisor, spec) + + assert pid == Storex.Registry.get_store_pid(@store, session) + end + + test "the name is released when the store stops" do + session = start_store(session()) + + assert [{_pid, nil}] = Registry.lookup(Storex.StoreRegistry, {session, @store}) + + Storex.Supervisor.remove_store(session, @store) + + # The cast is asynchronous, so wait for the Registry to drop the entry. + Enum.reduce_while(1..100, nil, fn _, _ -> + case Registry.lookup(Storex.StoreRegistry, {session, @store}) do + [] -> {:halt, :ok} + _ -> Process.sleep(10) && {:cont, nil} + end + end) + + assert Registry.lookup(Storex.StoreRegistry, {session, @store}) == [] + end + end +end From 78b0dcba54e1d2fceceee3942674e12a761171ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 07:42:47 +0200 Subject: [PATCH 03/10] Resolve mutations against the connection's own session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Storex.Socket.message_handle/2` looked the store up by `message.session` — the session id the client put in the frame — and never compared it against `state.session`, the id the handler generated for this connection. Any client could therefore mutate any other session's store by naming its session id, and received the resulting diff (and any `{:reply, message, state}` payload) back on its own socket. The `join` clause a few lines above already used `state.session`. This brings the mutation clause in line with it. The field stays in the frame and is still echoed in the response, but it no longer resolves anything. Deliberate cross-session mutation is what `Storex.mutate/3` and `Storex.mutate/4` are for. Fixing that alone would leave a reachable crash: a mutation for a store the session has not joined makes `Storex.Registry.get_store_pid/2` return `:undefined`, which `Storex.Supervisor.mutate_store/4` piped straight into `GenServer.call/2`, exiting the connection process. It was reachable before this change by naming a session that does not exist, so it is fixed here too — the lookup miss now returns an error the client receives as a normal error frame. `get_store_state/2` has the same shape but is left alone: its only caller runs inside the success branch of the join `with`, right after `add_store` registered the process, so the miss is unreachable there. --- CHANGELOG.md | 2 + lib/storex/socket.ex | 11 +++- lib/storex/supervisor.ex | 8 ++- test/storex/socket_test.exs | 100 ++++++++++++++++++++++++++++++++ test/storex/supervisor_test.exs | 5 ++ 5 files changed, 122 insertions(+), 4 deletions(-) create mode 100644 test/storex/socket_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index d8ebcde..54f7820 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- **[SECURITY]** A `mutation` frame is now resolved against the session the server assigned to the connection, not the `session` field carried by the frame. Any client could previously mutate — and read the resulting diff of — any other session's store by naming its session id. Mutating other sessions on purpose is what `Storex.mutate/3` and `Storex.mutate/4` are for +- A `mutation` for a store the session has not joined now returns an error to the client instead of exiting the connection process - Store processes are now registered through a `Registry` keyed by `{session, store}` instead of being named `:"#{session}_#{store}"`. Session ids are unique per connection, so the old naming created one permanent atom per session-store pair and could exhaust the atom table on a long-running node - Removed the `:pg2` fallback, unreachable since OTP 24 - Updated dependencies diff --git a/lib/storex/socket.ex b/lib/storex/socket.ex index de0b635..4952d09 100644 --- a/lib/storex/socket.ex +++ b/lib/storex/socket.ex @@ -50,10 +50,15 @@ defmodule Storex.Socket do end end - def message_handle(%{type: "mutation", session: session, store: store} = message, state) do + # The `session` carried by the frame is never used to resolve the store. It is + # supplied by the client, so trusting it would let any socket mutate any other + # session's store. Only `state.session`, generated by the handler for this + # connection, addresses a store. Mutating another session deliberately is what + # `Storex.mutate/3` and `Storex.mutate/4` are for. + def message_handle(%{type: "mutation", store: store} = message, %{session: session} = state) do Storex.Supervisor.mutate_store( - message.session, - message.store, + session, + store, message.data.name, message.data.data ) diff --git a/lib/storex/supervisor.ex b/lib/storex/supervisor.ex index 43501b1..dc1a0cc 100644 --- a/lib/storex/supervisor.ex +++ b/lib/storex/supervisor.ex @@ -64,7 +64,13 @@ defmodule Storex.Supervisor do def mutate_store(session, store, name, data) do Storex.Registry.get_store_pid(store, session) - |> GenServer.call({name, data}) + |> case do + :undefined -> + {:error, "Store '#{store}' is not joined in this session."} + + pid -> + GenServer.call(pid, {name, data}) + end end def remove_store(session, store) do diff --git a/test/storex/socket_test.exs b/test/storex/socket_test.exs new file mode 100644 index 0000000..2c84a11 --- /dev/null +++ b/test/storex/socket_test.exs @@ -0,0 +1,100 @@ +defmodule StorexTest.SocketTest do + use ExUnit.Case + + @store "StorexTest.Store.Counter" + + defp session, do: "session-#{System.unique_integer([:positive])}" + + defp joined_session do + session = session() + {:ok, _key} = Storex.Supervisor.add_store(@store, session, self(), %{}) + on_exit(fn -> Storex.Supervisor.remove_store(session, @store) end) + session + end + + defp mutation(session, name) do + %Storex.Message{ + type: "mutation", + store: @store, + session: session, + data: %{name: name, data: []}, + request: "request-id" + } + end + + describe "mutation" do + test "the session in the frame cannot address another session's store" do + attacker = joined_session() + victim = joined_session() + + # The frame claims the victim's session; the socket belongs to the attacker. + message = mutation(victim, "increase") + state = %{session: attacker, pid: self()} + + assert {:text, response, ^state} = Storex.Socket.message_handle(message, state) + + assert Storex.Supervisor.get_store_state(victim, @store) == %{counter: 0} + assert Storex.Supervisor.get_store_state(attacker, @store) == %{counter: 1} + + assert %{"session" => ^attacker, "diff" => [%{"p" => ["counter"], "t" => 1}]} = + Jason.decode!(response) + end + + test "the response always echoes the session of the socket" do + session = joined_session() + state = %{session: session, pid: self()} + + assert {:text, response, ^state} = + Storex.Socket.message_handle(mutation("some-other-session", "increase"), state) + + assert %{"session" => ^session} = Jason.decode!(response) + end + + test "a store the session has not joined returns an error instead of crashing" do + session = session() + state = %{session: session, pid: self()} + + assert {:text, response, ^state} = + Storex.Socket.message_handle(mutation(session, "increase"), state) + + assert %{ + "type" => "error", + "session" => ^session, + "store" => @store, + "error" => error, + "request" => "request-id" + } = Jason.decode!(response) + + assert error == "Store 'StorexTest.Store.Counter' is not joined in this session." + end + + test "a reply from the store is passed through" do + session = joined_session() + state = %{session: session, pid: self()} + + assert {:text, response, ^state} = + Storex.Socket.message_handle(mutation(session, "decrease"), state) + + assert %{"message" => "decreased", "diff" => [%{"p" => ["counter"], "t" => -1}]} = + Jason.decode!(response) + end + + test "a mutation pushed by Storex.mutate/3 is handled" do + session = joined_session() + state = %{session: session, pid: self()} + + # The shape the handlers build in handle_info/2: a plain map, no request id. + pushed = %{ + type: "mutation", + session: session, + store: @store, + data: %{data: [], name: "increase"} + } + + assert {:text, response, ^state} = Storex.Socket.message_handle(pushed, state) + + assert %{"request" => nil, "diff" => [%{"p" => ["counter"], "t" => 1}]} = + Jason.decode!(response) + end + end +end diff --git a/test/storex/supervisor_test.exs b/test/storex/supervisor_test.exs index 1ea876e..3e74556 100644 --- a/test/storex/supervisor_test.exs +++ b/test/storex/supervisor_test.exs @@ -49,6 +49,11 @@ defmodule StorexTest.SupervisorTest do assert pid == Storex.Registry.get_store_pid(@store, session) end + test "mutating a store that was never started returns an error" do + assert Storex.Supervisor.mutate_store(session(), @store, "increase", []) == + {:error, "Store 'StorexTest.Store.Counter' is not joined in this session."} + end + test "the name is released when the store stops" do session = start_store(session()) From baee36ba635b1a839a0fb35f2ca13a4be2909c8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 09:27:07 +0200 Subject: [PATCH 04/10] Reject binary websocket frames instead of decoding them `Storex.Handler.Cowboy` decoded `:binary` frames with `:erlang.binary_to_term/1`. Two problems in one line. It creates atoms out of bytes the client controls. A hand-built 25-byte external term format frame is enough to add one, so a single open socket can grow the atom table as fast as it can write, with no connection churn needed. It is also the only unauthenticated `binary_to_term` in the library. It also bypassed `Storex.Message.cast/1`, the allowlist every other entry point runs its payload through, and handed a raw term straight to `Storex.Socket.message_handle/2`. The `rescue ArgumentError` only covered a malformed encoding; a well-formed term of an unexpected shape raised `FunctionClauseError` and closed the connection with a 1011. Nothing uses binary frames. The client only ever sends `JSON.stringify` output, no test sends one (the `send_binary_frame/3` helper had no callers until now), and the README never documented the path. Keeping it would mean maintaining a second set of `cast/1` clauses for atom-keyed maps, for no caller. So the frame is refused with a 1003 instead. `Storex.Handler.Plug` had no `:binary` clause at all, which made any binary frame raise `FunctionClauseError` and kill the connection with a 1011 on the default transport. It now refuses the frame the same way. --- CHANGELOG.md | 2 ++ lib/storex/handler/cowboy.ex | 15 +++++++-------- lib/storex/handler/plug.ex | 6 ++++++ test/storex/handler/cowboy_test.exs | 29 +++++++++++++++++++++++++++++ test/storex/handler/plug_test.exs | 29 +++++++++++++++++++++++++++++ 5 files changed, 73 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 54f7820..a68fcd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- **[BREAKING]** **[SECURITY]** `Storex.Handler.Cowboy` no longer accepts `:binary` frames. It decoded them with `:erlang.binary_to_term/1`, which creates atoms out of bytes the client controls — a single 25-byte frame is enough — and passed the resulting term straight to `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1` allowlist every other entry point goes through. Binary frames are now closed with `1003` +- `Storex.Handler.Plug` had no clause for `:binary` frames at all, so one raised `FunctionClauseError` and closed the connection with `1011`. It now closes with `1003` like the cowboy handler - **[SECURITY]** A `mutation` frame is now resolved against the session the server assigned to the connection, not the `session` field carried by the frame. Any client could previously mutate — and read the resulting diff of — any other session's store by naming its session id. Mutating other sessions on purpose is what `Storex.mutate/3` and `Storex.mutate/4` are for - A `mutation` for a store the session has not joined now returns an error to the client instead of exiting the connection process - Store processes are now registered through a `Registry` keyed by `{session, store}` instead of being named `:"#{session}_#{store}"`. Session ids are unique per connection, so the old naming created one permanent atom per session-store pair and could exhaust the atom table on a long-running node diff --git a/lib/storex/handler/cowboy.ex b/lib/storex/handler/cowboy.ex index ce37da0..9e4434b 100644 --- a/lib/storex/handler/cowboy.ex +++ b/lib/storex/handler/cowboy.ex @@ -26,14 +26,13 @@ defmodule Storex.Handler.Cowboy do :ok end - def websocket_handle({:binary, frame}, state) do - try do - :erlang.binary_to_term(frame) - |> Socket.message_handle(state) - |> map_response() - rescue - ArgumentError -> {:reply, {:close, 1007, "Payload is malformed."}, state} - end + # Binary frames are not part of the protocol: the client only ever sends text. + # They used to be decoded with `:erlang.binary_to_term/1`, which creates atoms + # out of bytes the client controls and handed the resulting term straight to + # `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1` + # allowlist that every other entry point goes through. + def websocket_handle({:binary, _frame}, state) do + {:reply, {:close, 1003, "Binary frames are not supported."}, state} end def websocket_handle({:text, frame}, state) do diff --git a/lib/storex/handler/plug.ex b/lib/storex/handler/plug.ex index 2a71293..7a364c4 100644 --- a/lib/storex/handler/plug.ex +++ b/lib/storex/handler/plug.ex @@ -34,6 +34,12 @@ defmodule Storex.Handler.Plug do end end + # See the note in `Storex.Handler.Cowboy`. Without this clause a binary frame + # raises `FunctionClauseError` here and takes the connection down with a 1011. + def handle_in({_message, [opcode: :binary]}, state) do + {:stop, :normal, {1003, "Binary frames are not supported."}, state} + end + def handle_info({:mutate, store, mutation, data}, %{session: session} = state) do %{ type: "mutation", diff --git a/test/storex/handler/cowboy_test.exs b/test/storex/handler/cowboy_test.exs index 9219533..45b2bff 100644 --- a/test/storex/handler/cowboy_test.exs +++ b/test/storex/handler/cowboy_test.exs @@ -174,6 +174,35 @@ defmodule StorexTest.Handler.Cowboy do end end + describe "binary frames" do + test "are rejected with 1003", context do + client = tcp_client(context) + http1_handshake(client) + + send_binary_frame(client, :erlang.term_to_binary(%{type: "ping", request: "r"})) + + assert recv_connection_close_frame(client) == + {:ok, <<1003::16, "Binary frames are not supported."::binary>>} + end + + test "cannot create atoms", context do + client = tcp_client(context) + http1_handshake(client) + + name = "storex_binary_frame_probe_#{System.unique_integer([:positive])}" + + # External term format for an atom that does not exist in this VM yet. It + # is built by hand because `:erlang.term_to_binary/1` would create the atom + # here first, in the test process. + send_binary_frame(client, <<131, 118, byte_size(name)::16, name::binary>>) + + assert recv_connection_close_frame(client) == + {:ok, <<1003::16, "Binary frames are not supported."::binary>>} + + assert_raise ArgumentError, fn -> String.to_existing_atom(name) end + end + end + # Simple WebSocket client def tcp_client(context) do diff --git a/test/storex/handler/plug_test.exs b/test/storex/handler/plug_test.exs index d4617cb..41da0f3 100644 --- a/test/storex/handler/plug_test.exs +++ b/test/storex/handler/plug_test.exs @@ -178,6 +178,35 @@ defmodule StorexTest.Handler.Plug do end end + describe "binary frames" do + test "are rejected with 1003", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + send_binary_frame(client, :erlang.term_to_binary(%{type: "ping", request: "r"})) + + assert recv_connection_close_frame(client) == + {:ok, <<1003::16, "Binary frames are not supported."::binary>>} + end + + test "cannot create atoms", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + name = "storex_binary_frame_probe_#{System.unique_integer([:positive])}" + + # External term format for an atom that does not exist in this VM yet. It + # is built by hand because `:erlang.term_to_binary/1` would create the atom + # here first, in the test process. + send_binary_frame(client, <<131, 118, byte_size(name)::16, name::binary>>) + + assert recv_connection_close_frame(client) == + {:ok, <<1003::16, "Binary frames are not supported."::binary>>} + + assert_raise ArgumentError, fn -> String.to_existing_atom(name) end + end + end + # Simple WebSocket client def tcp_client(context) do From 11d4f4fe7c32ad086d12b6a460e55dda6a22ab07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 09:31:56 +0200 Subject: [PATCH 05/10] Share store name resolution between both transports A store name arrives as a string from the client, so resolving it to a module needs three checks together: `Module.safe_concat/1` so no new atom is created, `Code.ensure_compiled/1` so the name maps to a real module, and a check that the module declares the `Storex.Store` behaviour. `Storex.Socket` did all three. `Storex.HTTP` kept its own copy that had drifted down to `Module.safe_concat/1` alone, and then called `apply(module, :init, ["SSR", params])` on whatever came back. So `GET /storex?store=Any.Module¶ms=%7B%7D` invoked `init/2` on any loaded module whose name resolved and serialised the return value to the caller, and a module whose `init/2` returned an unexpected shape raised, which made the endpoint a probe for which modules exist. The defect is the duplication, not the missing lines, so both callers now share `Storex.Store.resolve/1` and the copies are gone. `Storex.HTTP.get_state/2` had the same problem on a smaller scale: its own `apply/3` plus a three clause `result/1` where `Storex.Store.__init__/3` already exists. It now uses the shared dispatcher, so SSR accepts the same return values as the WebSocket path and reports the same error for anything else, instead of a bare `FunctionClauseError`. The SSR error message interpolated the store name with `inspect/1`, rendering it double quoted. It now matches the WebSocket wording. --- CHANGELOG.md | 1 + lib/storex/http.ex | 35 ++++++++----------------- lib/storex/socket.ex | 25 +----------------- lib/storex/store.ex | 30 ++++++++++++++++++++++ test/fixtures/not_a_store.ex | 10 ++++++++ test/storex/http_test.exs | 50 ++++++++++++++++++++++++++++++++++++ test/storex/store_test.exs | 19 ++++++++++++++ 7 files changed, 121 insertions(+), 49 deletions(-) create mode 100644 test/fixtures/not_a_store.ex create mode 100644 test/storex/http_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index a68fcd5..d8dadfd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- **[SECURITY]** The SSR/HTTP path now runs the same store-name checks as the WebSocket path. `Storex.HTTP` carried its own copy of the resolution logic with only `Module.safe_concat/1`, missing `Code.ensure_compiled/1` and the `Storex.Store` behaviour check, so `GET /storex?store=Any.Module¶ms=%7B%7D` called `init/2` on any loaded module whose name resolved and serialised the result back to the caller. Both transports now share `Storex.Store.resolve/1` - **[BREAKING]** **[SECURITY]** `Storex.Handler.Cowboy` no longer accepts `:binary` frames. It decoded them with `:erlang.binary_to_term/1`, which creates atoms out of bytes the client controls — a single 25-byte frame is enough — and passed the resulting term straight to `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1` allowlist every other entry point goes through. Binary frames are now closed with `1003` - `Storex.Handler.Plug` had no clause for `:binary` frames at all, so one raised `FunctionClauseError` and closed the connection with `1011`. It now closes with `1003` like the cowboy handler - **[SECURITY]** A `mutation` frame is now resolved against the session the server assigned to the connection, not the `session` field carried by the frame. Any client could previously mutate — and read the resulting diff of — any other session's store by naming its session id. Mutating other sessions on purpose is what `Storex.mutate/3` and `Storex.mutate/4` are for diff --git a/lib/storex/http.ex b/lib/storex/http.ex index 4ce594e..6772f0d 100644 --- a/lib/storex/http.ex +++ b/lib/storex/http.ex @@ -1,6 +1,6 @@ defmodule Storex.HTTP do def init_store(store, params) do - with {:store, {:ok, store_module}} <- {:store, store |> get_module()}, + with {:store, {:ok, store_module}} <- {:store, Storex.Store.resolve(store)}, {:params, {:ok, params}} <- {:params, params |> get_params()}, {:state, {:ok, result}} <- {:state, get_state(store_module, params)} do {:ok, @@ -17,7 +17,7 @@ defmodule Storex.HTTP do type: "error", session: "SSR", store: store, - error: "Store '#{inspect(store)}' is not defined or can't be compiled." + error: "Store '#{store}' is not defined or can't be compiled." }} {:state, {:error, message}} -> @@ -40,35 +40,20 @@ defmodule Storex.HTTP do end end - defp get_module(store) do - try do - module = Module.safe_concat([store]) - {:ok, module} - rescue - ArgumentError -> {:error, :not_exists} - end - end - defp get_params(params) do params |> Jason.decode() end + # The SSR path is init-only, so the key a store may return is dropped. Using + # the shared dispatcher keeps the accepted return values, and the error raised + # for anything else, identical to the websocket path. defp get_state(module, params) do module - |> apply(:init, ["SSR", params]) - |> result() - end - - defp result({:ok, state}) do - {:ok, state} - end - - defp result({:ok, state, _}) do - {:ok, state} - end - - defp result({:error, error_message}) do - {:error, error_message} + |> Storex.Store.__init__("SSR", params) + |> case do + {:ok, state, _key} -> {:ok, state} + {:error, reason} -> {:error, reason} + end end end diff --git a/lib/storex/socket.ex b/lib/storex/socket.ex index 4952d09..af8c4da 100644 --- a/lib/storex/socket.ex +++ b/lib/storex/socket.ex @@ -21,7 +21,7 @@ defmodule Storex.Socket do end def message_handle(%{type: "join"} = message, state) do - with {:get_module, {:ok, _}} <- {:get_module, get_store_module(message.store)}, + with {:get_module, {:ok, _}} <- {:get_module, Storex.Store.resolve(message.store)}, {:add_store, {:ok, _}} <- {:add_store, Storex.Supervisor.add_store(message.store, state.session, state.pid, message.data)} do @@ -94,27 +94,4 @@ defmodule Storex.Socket do |> Jason.encode!() |> (&{:text, &1, state}).() end - - defp safe_concat(store) do - try do - module = Module.safe_concat([store]) - {:ok, module} - rescue - ArgumentError -> {:error, :not_exists} - end - end - - defp get_store_module(store) do - with {:ok, module} <- safe_concat(store), - {:module, module} <- Code.ensure_compiled(module), - true <- - Storex.Store in (module.module_info(:attributes) - |> Keyword.get_values(:behaviour) - |> List.flatten()) do - {:ok, module} - else - false -> {:error, :not_store} - _ -> {:error, :not_exists} - end - end end diff --git a/lib/storex/store.ex b/lib/storex/store.ex index f1a586b..d2e60f6 100644 --- a/lib/storex/store.ex +++ b/lib/storex/store.ex @@ -23,6 +23,36 @@ defmodule Storex.Store do @callback terminate(session_id :: binary(), params :: %{binary() => any()}, state :: any()) :: any() @optional_callbacks terminate: 3 + @doc false + # Resolves the store module from the name a client sent. All three checks + # belong together: `Module.safe_concat/1` refuses to create new atoms, + # `Code.ensure_compiled/1` refuses names that do not resolve to a real module, + # and the behaviour check refuses modules that are not stores. Dropping any of + # them lets client input reach arbitrary modules, which is what the SSR path + # did for as long as it carried its own copy of only the first check. + def resolve(store) do + with {:ok, module} <- safe_concat(store), + {:module, module} <- Code.ensure_compiled(module), + true <- storex_store?(module) do + {:ok, module} + else + false -> {:error, :not_store} + _ -> {:error, :not_exists} + end + end + + defp safe_concat(store) do + {:ok, Module.safe_concat([store])} + rescue + ArgumentError -> {:error, :not_exists} + end + + defp storex_store?(module) do + __MODULE__ in (module.module_info(:attributes) + |> Keyword.get_values(:behaviour) + |> List.flatten()) + end + @doc false def __init__(store, session, params) do apply(store, :init, [session, params]) diff --git a/test/fixtures/not_a_store.ex b/test/fixtures/not_a_store.ex new file mode 100644 index 0000000..e4da980 --- /dev/null +++ b/test/fixtures/not_a_store.ex @@ -0,0 +1,10 @@ +defmodule StorexTest.NotAStore do + @moduledoc """ + A plain module that exports `init/2` but does not declare the `Storex.Store` + behaviour. The SSR path used to resolve and call it. + """ + + def init(_session, _params) do + raise "init/2 must not be called on a module that is not a Storex.Store" + end +end diff --git a/test/storex/http_test.exs b/test/storex/http_test.exs new file mode 100644 index 0000000..8ab7548 --- /dev/null +++ b/test/storex/http_test.exs @@ -0,0 +1,50 @@ +defmodule StorexTest.HTTPTest do + use ExUnit.Case + + describe "init_store/2" do + test "returns the initial state of a store" do + assert {:ok, + %{ + type: "join", + session: "SSR", + store: "StorexTest.Store.Counter", + data: %{counter: 0} + }} = Storex.HTTP.init_store("StorexTest.Store.Counter", "{}") + end + + test "params are forwarded to the store" do + assert {:ok, %{data: "custom"}} = + Storex.HTTP.init_store("StorexTest.Store.Text", ~s({"initial_value": "custom"})) + end + + test "a module that is not a store is refused without being called" do + # StorexTest.NotAStore exports init/2 and raises if it is ever reached. + assert {:error, + %{ + type: "error", + session: "SSR", + store: "StorexTest.NotAStore", + error: "Store 'StorexTest.NotAStore' is not defined or can't be compiled." + }} = Storex.HTTP.init_store("StorexTest.NotAStore", "{}") + end + + test "a name that does not resolve to a module is refused" do + assert {:error, %{type: "error", error: error}} = + Storex.HTTP.init_store("StorexTest.Store.NotExisting", "{}") + + assert error == + "Store 'StorexTest.Store.NotExisting' is not defined or can't be compiled." + end + + test "an {:error, reason} from the store is passed through" do + assert {:error, %{type: "error", error: "Unauthorized"}} = + Storex.HTTP.init_store("StorexTest.Store.ErrorInit", "{}") + end + + test "an unsupported return value from the store raises" do + assert_raise RuntimeError, + "Return value of store init should be {:ok, state}, {:ok, state, key} or {:error, reason}", + fn -> Storex.HTTP.init_store("StorexTest.Store.InvalidInit", "{}") end + end + end +end diff --git a/test/storex/store_test.exs b/test/storex/store_test.exs index d8ea96a..442b914 100644 --- a/test/storex/store_test.exs +++ b/test/storex/store_test.exs @@ -8,6 +8,25 @@ defmodule StorexTest.StoreTest do alias StorexTest.Store.KeyInit alias StorexTest.Store.Text + describe "resolve/1" do + test "resolves a module that declares the behaviour" do + assert Storex.Store.resolve("StorexTest.Store.Counter") == {:ok, Counter} + end + + test "refuses a module that does not declare the behaviour" do + assert Storex.Store.resolve("StorexTest.NotAStore") == {:error, :not_store} + end + + test "refuses a name that does not resolve to a module" do + assert Storex.Store.resolve("StorexTest.Store.NotExisting") == {:error, :not_exists} + end + + test "refuses a name that is not an existing atom" do + assert Storex.Store.resolve("Never.Compiled.#{System.unique_integer([:positive])}") == + {:error, :not_exists} + end + end + describe "init dispatch" do test "{:ok, state} is normalized with a nil key" do assert Storex.Store.__init__(Counter, "session", %{}) == {:ok, %{counter: 0}, nil} From 1a69ca97d70c53f5e2057e11c81152f9c839e19b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 09:38:27 +0200 Subject: [PATCH 06/10] Remove dead message shapes and dead code Three unrelated pieces of code that were either unreachable or wrong. `Storex.Message.cast/1` accepted an `error` frame, but `Storex.Socket.message_handle/2` has no clause for one, so a client could kill its connection process with a well-formed frame. The direction is server to client, and those frames are built as plain maps in `Storex.Socket` rather than through the struct, so nothing produced or consumed the shape. It is out of the allowlist, which sends the frame down the same path as any other unknown type: closed with 1007. `Storex.Registry.session_pid/1` had no callers anywhere, and could not have worked if it did: it matched `{:_, session, :_, :"$1"}`, a 4-tuple, against a table of 5-tuples, with `session` in the `store_pid` position. Its `:DOWN` handler also replaced the `%{}` state with the atom `:ok`, which is harmless only for as long as nothing reads it. `Storex.Handler.Cowboy.websocket_init/3` is the cowboy 1.x callback signature. Cowboy 2.x calls `websocket_init/1`, and `init/2` already returns the real state, so it never ran. Testing the first of those turned up a fourth: the two handlers closed a malformed payload differently. `Storex.Handler.Plug` passed the reason as the process exit reason rather than the close payload, so the client saw a bare 1007 while cowboy sent 1007 plus the reason, and the connection process exited abnormally every time. It now uses the same shape as the rest of its close responses. --- CHANGELOG.md | 3 + lib/storex/handler/cowboy.ex | 4 -- lib/storex/handler/plug.ex | 2 +- lib/storex/message.ex | 15 ++--- lib/storex/registry.ex | 16 +----- test/storex/handler/cowboy_test.exs | 22 ++++++++ test/storex/handler/plug_test.exs | 22 ++++++++ test/storex/message_test.exs | 85 +++++++++++++++++++++++++++++ 8 files changed, 139 insertions(+), 30 deletions(-) create mode 100644 test/storex/message_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index d8dadfd..f0c8444 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,9 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- A client could kill its connection process with a well-formed `error` frame: `Storex.Message.cast/1` accepted the shape but `Storex.Socket.message_handle/2` had no clause for it. `error` frames only travel server to client, so the shape is no longer accepted and the frame is refused with `1007` like any other unknown type +- `Storex.Handler.Plug` closed a malformed payload with a bare `1007`, passing the reason as the process exit reason instead of the close payload. It now sends `1007` with the reason, matching `Storex.Handler.Cowboy` +- Removed dead code: `Storex.Registry.session_pid/1` (no callers, and its `:ets.match/2` pattern was a 4-tuple against 5-tuple records, so it never matched anything) and `Storex.Handler.Cowboy.websocket_init/3` (a cowboy 1.x callback, unreachable on cowboy 2.x) - **[SECURITY]** The SSR/HTTP path now runs the same store-name checks as the WebSocket path. `Storex.HTTP` carried its own copy of the resolution logic with only `Module.safe_concat/1`, missing `Code.ensure_compiled/1` and the `Storex.Store` behaviour check, so `GET /storex?store=Any.Module¶ms=%7B%7D` called `init/2` on any loaded module whose name resolved and serialised the result back to the caller. Both transports now share `Storex.Store.resolve/1` - **[BREAKING]** **[SECURITY]** `Storex.Handler.Cowboy` no longer accepts `:binary` frames. It decoded them with `:erlang.binary_to_term/1`, which creates atoms out of bytes the client controls — a single 25-byte frame is enough — and passed the resulting term straight to `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1` allowlist every other entry point goes through. Binary frames are now closed with `1003` - `Storex.Handler.Plug` had no clause for `:binary` frames at all, so one raised `FunctionClauseError` and closed the connection with `1011`. It now closes with `1003` like the cowboy handler diff --git a/lib/storex/handler/cowboy.ex b/lib/storex/handler/cowboy.ex index 9e4434b..f9e36f3 100644 --- a/lib/storex/handler/cowboy.ex +++ b/lib/storex/handler/cowboy.ex @@ -9,10 +9,6 @@ defmodule Storex.Handler.Cowboy do {:cowboy_websocket, request, %{session: session, pid: request.pid}} end - def websocket_init(_type, req, _opts) do - {:ok, req, %{status: "inactive"}} - end - def terminate(_reason, _req, %{session: session}) do Storex.Registry.session_stores(session) |> Enum.each(fn {store, _, session, _, _} -> diff --git a/lib/storex/handler/plug.ex b/lib/storex/handler/plug.ex index 7a364c4..45c8de9 100644 --- a/lib/storex/handler/plug.ex +++ b/lib/storex/handler/plug.ex @@ -30,7 +30,7 @@ defmodule Storex.Handler.Plug do |> map_response() else {:error, _} -> - {:stop, "Payload is malformed.", 1007, state} + {:stop, :normal, {1007, "Payload is malformed."}, state} end end diff --git a/lib/storex/message.ex b/lib/storex/message.ex index fc77fe2..be60485 100644 --- a/lib/storex/message.ex +++ b/lib/storex/message.ex @@ -37,17 +37,10 @@ defmodule Storex.Message do }} end - def cast(%{ - "type" => "error", - "store" => store, - "data" => data, - "request" => request, - "session" => session - }) do - {:ok, - %__MODULE__{type: "error", store: store, data: data, request: request, session: session}} - end - + # `error` frames only ever travel server to client, and are built as plain maps + # in `Storex.Socket`. Casting one here made it past the allowlist and then hit + # `Storex.Socket.message_handle/2`, which has no clause for it, so a client + # could kill its connection process with a well-formed frame. def cast(_) do {:error, "Unknown message type"} end diff --git a/lib/storex/registry.ex b/lib/storex/registry.ex index 60e5e87..20fc19d 100644 --- a/lib/storex/registry.ex +++ b/lib/storex/registry.ex @@ -15,10 +15,6 @@ defmodule Storex.Registry do {:ok, %{}} end - def session_pid(session) do - GenServer.call(@registry, {:session_pid, session}) - end - def register_store(store, store_pid, session, session_pid, key) do GenServer.call(@registry, {:register_store, store, store_pid, session, session_pid, key}) end @@ -43,14 +39,6 @@ defmodule Storex.Registry do GenServer.call(@registry, {:session_stores, session}) end - def handle_call({:session_pid, session}, _from, state) do - :ets.match(@registry, {:_, session, :_, :"$1"}) - |> case do - [] -> {:reply, :undefined, state} - [[pid] | _tail] -> {:reply, pid, state} - end - end - def handle_call({:register_store, store, store_pid, session, session_pid, key}, _from, state) do :ets.insert(@registry, {store, store_pid, session, session_pid, key}) Process.monitor(store_pid) @@ -90,9 +78,9 @@ defmodule Storex.Registry do {:reply, stores, state} end - def handle_info({:DOWN, _ref, :process, pid, _reason}, _state) do + def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do :ets.match_delete(@registry, {:_, pid, :_, :_, :_}) - {:noreply, :ok} + {:noreply, state} end end diff --git a/test/storex/handler/cowboy_test.exs b/test/storex/handler/cowboy_test.exs index 45b2bff..a03a4dd 100644 --- a/test/storex/handler/cowboy_test.exs +++ b/test/storex/handler/cowboy_test.exs @@ -174,6 +174,28 @@ defmodule StorexTest.Handler.Cowboy do end end + describe "unknown message types" do + test "an error frame is refused with 1007", context do + client = tcp_client(context) + http1_handshake(client) + + # `error` frames only travel server to client. This one used to pass + # `Storex.Message.cast/1` and then crash `message_handle/2`. + send_text_frame(client, """ + { + "type": "error", + "store": "StorexTest.Store.Counter", + "data": null, + "request": "#{random_string()}", + "session": "#{random_string()}" + } + """) + + assert recv_connection_close_frame(client) == + {:ok, <<1007::16, "Payload is malformed."::binary>>} + end + end + describe "binary frames" do test "are rejected with 1003", context do client = tcp_client(context) diff --git a/test/storex/handler/plug_test.exs b/test/storex/handler/plug_test.exs index 41da0f3..52e6a41 100644 --- a/test/storex/handler/plug_test.exs +++ b/test/storex/handler/plug_test.exs @@ -178,6 +178,28 @@ defmodule StorexTest.Handler.Plug do end end + describe "unknown message types" do + test "an error frame is refused with 1007", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + # `error` frames only travel server to client. This one used to pass + # `Storex.Message.cast/1` and then crash `message_handle/2`. + send_text_frame(client, """ + { + "type": "error", + "store": "StorexTest.Store.Counter", + "data": null, + "request": "#{random_string()}", + "session": "#{random_string()}" + } + """) + + assert recv_connection_close_frame(client) == + {:ok, <<1007::16, "Payload is malformed."::binary>>} + end + end + describe "binary frames" do test "are rejected with 1003", context do client = tcp_client(context) diff --git a/test/storex/message_test.exs b/test/storex/message_test.exs new file mode 100644 index 0000000..a0c36b8 --- /dev/null +++ b/test/storex/message_test.exs @@ -0,0 +1,85 @@ +defmodule StorexTest.MessageTest do + use ExUnit.Case + + describe "cast/1" do + test "casts a ping" do + assert Storex.Message.cast(%{"type" => "ping", "request" => "r"}) == + {:ok, %Storex.Message{type: "ping", request: "r"}} + end + + test "casts a join" do + assert Storex.Message.cast(%{ + "type" => "join", + "store" => "Store", + "data" => %{}, + "request" => "r" + }) == + {:ok, %Storex.Message{type: "join", store: "Store", data: %{}, request: "r"}} + end + + test "casts a join carrying a session" do + assert Storex.Message.cast(%{ + "type" => "join", + "store" => "Store", + "data" => %{}, + "request" => "r", + "session" => "s" + }) == + {:ok, + %Storex.Message{ + type: "join", + store: "Store", + data: %{}, + request: "r", + session: "s" + }} + end + + test "casts a mutation" do + assert Storex.Message.cast(%{ + "type" => "mutation", + "store" => "Store", + "data" => %{"name" => "increase", "data" => []}, + "request" => "r", + "session" => "s" + }) == + {:ok, + %Storex.Message{ + type: "mutation", + store: "Store", + data: %{name: "increase", data: []}, + request: "r", + session: "s" + }} + end + + test "refuses an error frame, which only travels server to client" do + assert Storex.Message.cast(%{ + "type" => "error", + "store" => "Store", + "data" => nil, + "request" => "r", + "session" => "s" + }) == {:error, "Unknown message type"} + end + + test "refuses a mutation without a name" do + assert Storex.Message.cast(%{ + "type" => "mutation", + "store" => "Store", + "data" => %{}, + "request" => "r", + "session" => "s" + }) == {:error, "Unknown message type"} + end + + test "refuses an unknown type" do + assert Storex.Message.cast(%{"type" => "whatever", "request" => "r"}) == + {:error, "Unknown message type"} + end + + test "refuses a payload that is not a message" do + assert Storex.Message.cast(%{}) == {:error, "Unknown message type"} + end + end +end From 1e505f0248d9c7cc7a8108e16744c72859d32eb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 09:43:51 +0200 Subject: [PATCH 07/10] Ask the store process for its state instead of :sys.get_state/1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Storex.Supervisor.get_store_state/2` read a store's state with `:sys.get_state/1` and then reached into the result with `Map.get(:state)`. That is a debug function, it runs on the join path in production, it uses the `:sys` timeout rather than the caller's, and it couples the supervisor to the internal state shape of the `Server` module generated in `Storex.Store` — a module in a different file that cannot change its representation without breaking this caller. The generated `Server` now answers `handle_call(:get_state, ...)` for itself, and the supervisor calls that. The return type becomes `{:ok, state} | {:error, reason}`, matching `mutate_store/4`. Besides being one contract instead of two in the same module, it closes the race left open when `mutate_store/4` was guarded: the store process can be gone between `add_store/4` returning and the join reading the state, since it is `restart: :transient`. That used to be `:sys.get_state(:undefined)`, which exits and takes the connection with it. The join now answers with an error frame. --- CHANGELOG.md | 1 + lib/storex/socket.ex | 8 ++++---- lib/storex/store.ex | 4 ++++ lib/storex/supervisor.ex | 12 ++++++++++-- test/storex/socket_test.exs | 4 ++-- test/storex/store_test.exs | 5 ++--- test/storex/storex_test.exs | 12 ++++++------ test/storex/supervisor_test.exs | 13 +++++++++++++ 8 files changed, 42 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0c8444..24bf248 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- The state a store reports on join is now read by asking the store process (`handle_call(:get_state, ...)` on the generated `Server`) instead of `:sys.get_state/1`, a debug function that was reaching into the process's internal state shape from the outside on every join. If the process is gone by the time the join reads it, the client now gets an error frame instead of the connection exiting - A client could kill its connection process with a well-formed `error` frame: `Storex.Message.cast/1` accepted the shape but `Storex.Socket.message_handle/2` had no clause for it. `error` frames only travel server to client, so the shape is no longer accepted and the frame is refused with `1007` like any other unknown type - `Storex.Handler.Plug` closed a malformed payload with a bare `1007`, passing the reason as the process exit reason instead of the close payload. It now sends `1007` with the reason, matching `Storex.Handler.Cowboy` - Removed dead code: `Storex.Registry.session_pid/1` (no callers, and its `:ets.match/2` pattern was a 4-tuple against 5-tuple records, so it never matched anything) and `Storex.Handler.Cowboy.websocket_init/3` (a cowboy 1.x callback, unreachable on cowboy 2.x) diff --git a/lib/storex/socket.ex b/lib/storex/socket.ex index af8c4da..66823a5 100644 --- a/lib/storex/socket.ex +++ b/lib/storex/socket.ex @@ -24,9 +24,9 @@ defmodule Storex.Socket do with {:get_module, {:ok, _}} <- {:get_module, Storex.Store.resolve(message.store)}, {:add_store, {:ok, _}} <- {:add_store, - Storex.Supervisor.add_store(message.store, state.session, state.pid, message.data)} do - store_state = Storex.Supervisor.get_store_state(state.session, message.store) - + Storex.Supervisor.add_store(message.store, state.session, state.pid, message.data)}, + {:store_state, {:ok, store_state}} <- + {:store_state, Storex.Supervisor.get_store_state(state.session, message.store)} do message = Map.put(message, :data, store_state) |> Map.put(:session, state.session) @@ -34,7 +34,7 @@ defmodule Storex.Socket do {:text, message, state} else - {:add_store, {:error, error_message}} -> + {step, {:error, error_message}} when step in [:add_store, :store_state] -> %{ type: "error", session: state.session, diff --git a/lib/storex/store.ex b/lib/storex/store.ex index d2e60f6..925df52 100644 --- a/lib/storex/store.ex +++ b/lib/storex/store.ex @@ -155,6 +155,10 @@ defmodule Storex.Store do {:stop, :normal, state} end + def handle_call(:get_state, _, state) do + {:reply, state.state, state} + end + def handle_call({name, data}, _, state) do Storex.Store.__mutation__(@store, name, data, state.session, state.params, state.state) |> case do diff --git a/lib/storex/supervisor.ex b/lib/storex/supervisor.ex index dc1a0cc..2a77ead 100644 --- a/lib/storex/supervisor.ex +++ b/lib/storex/supervisor.ex @@ -56,10 +56,18 @@ defmodule Storex.Supervisor do end end + # `:sys.get_state/1` is a debug function, and using it here meant reading the + # generated `Server`'s internal state shape from the outside, on the join path. + # The process answers for its own state instead. def get_store_state(session, store) do Storex.Registry.get_store_pid(store, session) - |> :sys.get_state() - |> Map.get(:state) + |> case do + :undefined -> + {:error, "Store '#{store}' is not joined in this session."} + + pid -> + {:ok, GenServer.call(pid, :get_state)} + end end def mutate_store(session, store, name, data) do diff --git a/test/storex/socket_test.exs b/test/storex/socket_test.exs index 2c84a11..0903843 100644 --- a/test/storex/socket_test.exs +++ b/test/storex/socket_test.exs @@ -33,8 +33,8 @@ defmodule StorexTest.SocketTest do assert {:text, response, ^state} = Storex.Socket.message_handle(message, state) - assert Storex.Supervisor.get_store_state(victim, @store) == %{counter: 0} - assert Storex.Supervisor.get_store_state(attacker, @store) == %{counter: 1} + assert Storex.Supervisor.get_store_state(victim, @store) == {:ok, %{counter: 0}} + assert Storex.Supervisor.get_store_state(attacker, @store) == {:ok, %{counter: 1}} assert %{"session" => ^attacker, "diff" => [%{"p" => ["counter"], "t" => 1}]} = Jason.decode!(response) diff --git a/test/storex/store_test.exs b/test/storex/store_test.exs index 442b914..707717d 100644 --- a/test/storex/store_test.exs +++ b/test/storex/store_test.exs @@ -118,9 +118,8 @@ defmodule StorexTest.StoreTest do assert {:ok, nil} = Storex.Supervisor.add_store("StorexTest.Store.Counter", session, self(), %{}) - assert Storex.Supervisor.get_store_state(session, "StorexTest.Store.Counter") == %{ - counter: 0 - } + assert Storex.Supervisor.get_store_state(session, "StorexTest.Store.Counter") == + {:ok, %{counter: 0}} end test "starts a store returning {:ok, state, key}", %{session: session} do diff --git a/test/storex/storex_test.exs b/test/storex/storex_test.exs index b8b7ff6..5e6086f 100644 --- a/test/storex/storex_test.exs +++ b/test/storex/storex_test.exs @@ -53,7 +53,7 @@ defmodule StorexTest do test "get store", %{session: session, store: store, pid: pid} do assert {:ok, _pid} = Storex.Supervisor.add_store(store, session, pid, %{}) - assert %{counter: 0} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 0}} = Storex.Supervisor.get_store_state(session, store) Storex.Supervisor.remove_store(session, store) end @@ -64,7 +64,7 @@ defmodule StorexTest do assert_receive :ok - assert %{counter: 1} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(session, store) end test "mutate store in cluster", %{session: session, store: store, pid: pid} do @@ -83,7 +83,7 @@ defmodule StorexTest do assert_receive :ok - assert %{counter: 1} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(session, store) end end @@ -117,7 +117,7 @@ defmodule StorexTest do assert_receive :ok - assert %{counter: 1} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(session, store) end test "don't mutate store for invalid key", %{session: session, store: store, pid: pid} do @@ -127,7 +127,7 @@ defmodule StorexTest do refute_receive :ok - assert %{counter: 0} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 0}} = Storex.Supervisor.get_store_state(session, store) end test "don't mutate store for invalid key in cluster", %{ @@ -150,7 +150,7 @@ defmodule StorexTest do refute_receive :ok - assert %{counter: 0} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 0}} = Storex.Supervisor.get_store_state(session, store) end end diff --git a/test/storex/supervisor_test.exs b/test/storex/supervisor_test.exs index 3e74556..b6d6683 100644 --- a/test/storex/supervisor_test.exs +++ b/test/storex/supervisor_test.exs @@ -49,6 +49,19 @@ defmodule StorexTest.SupervisorTest do assert pid == Storex.Registry.get_store_pid(@store, session) end + test "reading the state of a store that was never started returns an error" do + assert Storex.Supervisor.get_store_state(session(), @store) == + {:error, "Store 'StorexTest.Store.Counter' is not joined in this session."} + end + + test "the store answers for its own state, without :sys.get_state/1" do + session = start_store(session()) + pid = Storex.Registry.get_store_pid(@store, session) + + assert GenServer.call(pid, :get_state) == %{counter: 0} + assert Storex.Supervisor.get_store_state(session, @store) == {:ok, %{counter: 0}} + end + test "mutating a store that was never started returns an error" do assert Storex.Supervisor.mutate_store(session(), @store, "increase", []) == {:error, "Store 'StorexTest.Store.Counter' is not joined in this session."} From f477264dcdd32649e8b427ff11311513c2160e54 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 11:31:16 +0200 Subject: [PATCH 08/10] Read the registry without a GenServer round-trip The registry table is created `:protected`, which already lets any process read it, yet every accessor went through `GenServer.call` to a single process that then ran the `:ets.match` itself. Every mutation performs at least one lookup, so that process was a global serialisation point for the whole library. Reads now run in the caller. Writes and the `:DOWN` cleanup stay in the process, which still owns the table. Measured at 500 lookups per reader, comparing the round-trip against reading directly: readers GenServer direct 1 1.7ms 0.6ms 4 6.7ms 1.8ms 16 24.8ms 10.5ms 64 98.3ms 16.2ms 256 335.8ms 56.2ms The GenServer path grows linearly with load while the direct path does not, which is the serialisation showing up rather than `:ets.match` being slow. Also removes the unreachable `{:error, _}` branch in `Storex.PG.broadcast/1`. `:pg.get_members/2` always returns a list; the error tuple was `:pg2`'s contract and `:pg2` went away in this release. That branch was the reason `broadcast/1` used a comprehension, and `send/2` returns the message it sent, so `Storex.mutate/3` and `Storex.mutate/4` were handing callers the internal broadcast envelope once per node: Storex.mutate("Some.Store", "reload", ["x"]) #=> [broadcast: {:mutate, "Some.Store", "reload", ["x"]}] They return `:ok` now. --- CHANGELOG.md | 3 ++ lib/storex/pg.ex | 18 +++++------ lib/storex/registry.ex | 51 ++++++++++++------------------- test/storex/registry_test.exs | 56 +++++++++++++++++++++++++++++++++++ test/storex/storex_test.exs | 10 +++++++ 5 files changed, 96 insertions(+), 42 deletions(-) create mode 100644 test/storex/registry_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index 24bf248..1ab198b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,9 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- `Storex.Registry` reads (`get_store/2`, `get_store_pid/2`, `get_store_instances/1`, `session_stores/1`) now run in the calling process against the `:protected` ETS table instead of a `GenServer.call`. The registry process was a global serialisation point — every mutation performs at least one lookup — and the cost grew with the number of connections. Measured at 500 lookups per reader: 98.3ms against 16.2ms with 64 concurrent readers, 335.8ms against 56.2ms with 256. Writes and the `:DOWN` cleanup still go through the process +- **[BREAKING]** `Storex.mutate/3` and `Storex.mutate/4` return `:ok`. They used to return whatever the comprehension in `Storex.PG.broadcast/1` produced, which was the internal broadcast envelope repeated once per node (`[broadcast: {:mutate, "Store", "reload", []}]`) +- Removed the unreachable `{:error, _}` branch in `Storex.PG.broadcast/1`. `:pg.get_members/2` always returns a list; the error tuple was `:pg2`'s contract, and `:pg2` support went away in this release - The state a store reports on join is now read by asking the store process (`handle_call(:get_state, ...)` on the generated `Server`) instead of `:sys.get_state/1`, a debug function that was reaching into the process's internal state shape from the outside on every join. If the process is gone by the time the join reads it, the client now gets an error frame instead of the connection exiting - A client could kill its connection process with a well-formed `error` frame: `Storex.Message.cast/1` accepted the shape but `Storex.Socket.message_handle/2` had no clause for it. `error` frames only travel server to client, so the shape is no longer accepted and the frame is refused with `1007` like any other unknown type - `Storex.Handler.Plug` closed a malformed payload with a bare `1007`, passing the reason as the process exit reason instead of the close payload. It now sends `1007` with the reason, matching `Storex.Handler.Cowboy` diff --git a/lib/storex/pg.ex b/lib/storex/pg.ex index b87722b..99da40d 100644 --- a/lib/storex/pg.ex +++ b/lib/storex/pg.ex @@ -14,17 +14,15 @@ defmodule Storex.PG do {:ok, @name} end + # `:pg.get_members/2` always returns a list — the `{:error, _}` clause this used + # to carry was `:pg2`'s contract, and `:pg2` is gone. `Enum.each/2` rather than + # a comprehension so the return value is `:ok`: `send/2` returns the message, + # so `Storex.mutate/3` used to hand back the internal broadcast envelope once + # per node. def broadcast(payload) do - :pg.get_members(Storex.PG, @name) - |> case do - {:error, _} -> - :error - - pids -> - for pid <- pids do - send(pid, {:broadcast, payload}) - end - end + Storex.PG + |> :pg.get_members(@name) + |> Enum.each(&send(&1, {:broadcast, payload})) end @impl true diff --git a/lib/storex/registry.ex b/lib/storex/registry.ex index 20fc19d..0e3bd7f 100644 --- a/lib/storex/registry.ex +++ b/lib/storex/registry.ex @@ -23,20 +23,35 @@ defmodule Storex.Registry do GenServer.call(@registry, {:unregister_store, store, session}) end + # The table is `:protected`: only the owning process writes, but every process + # reads. So reads run in the caller. Routing them through this GenServer made + # it a global serialisation point for the whole library — every mutation does + # at least one lookup — and the cost scales with the number of connections. + # Measured, 500 lookups per reader: with 64 concurrent readers, 98.3ms through + # the GenServer against 16.2ms reading directly; with 256, 335.8ms against + # 56.2ms. Writes and the `:DOWN` cleanup stay in the process. def get_store(store, session) do - GenServer.call(@registry, {:get_store, store, session}) + :ets.match_object(@registry, {store, :"$1", session, :_, :_}) + |> case do + [] -> :undefined + [object | _tail] -> object + end end def get_store_pid(store, session) do - GenServer.call(@registry, {:get_store_pid, store, session}) + :ets.match(@registry, {store, :"$1", session, :_, :_}) + |> case do + [] -> :undefined + [[pid] | _tail] -> pid + end end def get_store_instances(query) do - GenServer.call(@registry, {:get_store_instances, query}) + :ets.match_object(@registry, query) end def session_stores(session) do - GenServer.call(@registry, {:session_stores, session}) + :ets.match_object(@registry, {:_, :_, session, :_, :_}) end def handle_call({:register_store, store, store_pid, session, session_pid, key}, _from, state) do @@ -50,34 +65,6 @@ defmodule Storex.Registry do {:reply, result, state} end - def handle_call({:get_store, store, session}, _from, state) do - :ets.match_object(@registry, {store, :"$1", session, :_, :_}) - |> case do - [] -> {:reply, :undefined, state} - [object | _tail] -> {:reply, object, state} - end - end - - def handle_call({:get_store_pid, store, session}, _from, state) do - :ets.match(@registry, {store, :"$1", session, :_, :_}) - |> case do - [] -> {:reply, :undefined, state} - [[pid] | _tail] -> {:reply, pid, state} - end - end - - def handle_call({:get_store_instances, query}, _from, state) do - instances = :ets.match_object(@registry, query) - - {:reply, instances, state} - end - - def handle_call({:session_stores, session}, _from, state) do - stores = :ets.match_object(@registry, {:_, :_, session, :_, :_}) - - {:reply, stores, state} - end - def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do :ets.match_delete(@registry, {:_, pid, :_, :_, :_}) diff --git a/test/storex/registry_test.exs b/test/storex/registry_test.exs new file mode 100644 index 0000000..fd0d32a --- /dev/null +++ b/test/storex/registry_test.exs @@ -0,0 +1,56 @@ +defmodule StorexTest.RegistryTest do + use ExUnit.Case, async: false + + @registry :storex_registry + + setup do + session = "session-#{System.unique_integer([:positive])}" + {:ok, _} = Storex.Registry.register_store("Store", self(), session, self(), "key") + on_exit(fn -> Storex.Registry.unregister_store("Store", session) end) + %{session: session} + end + + test "get_store/2 returns the whole row", %{session: session} do + assert Storex.Registry.get_store("Store", session) == + {"Store", self(), session, self(), "key"} + end + + test "get_store_pid/2 returns the store pid", %{session: session} do + assert Storex.Registry.get_store_pid("Store", session) == self() + end + + test "session_stores/1 returns every row of a session", %{session: session} do + assert Storex.Registry.session_stores(session) == + [{"Store", self(), session, self(), "key"}] + end + + test "get_store_instances/1 matches on the given query", %{session: session} do + assert Storex.Registry.get_store_instances({"Store", :_, :_, :_, "key"}) + |> Enum.any?(fn + {_, _, ^session, _, _} -> true + _ -> false + end) + end + + test "a miss is :undefined", %{session: session} do + assert Storex.Registry.get_store("Never.Joined", session) == :undefined + assert Storex.Registry.get_store_pid("Never.Joined", session) == :undefined + end + + test "reads do not go through the registry process", %{session: session} do + # With the owner suspended, anything that needs a GenServer round-trip + # blocks. Reads run in the calling process, so they still answer. + :sys.suspend(@registry) + + try do + assert Storex.Registry.get_store_pid("Store", session) == self() + assert Storex.Registry.get_store("Store", session) |> elem(0) == "Store" + assert Storex.Registry.session_stores(session) != [] + + # A write still needs the process, so it times out while it is suspended. + assert catch_exit(GenServer.call(@registry, {:unregister_store, "Store", session}, 100)) + after + :sys.resume(@registry) + end + end +end diff --git a/test/storex/storex_test.exs b/test/storex/storex_test.exs index 5e6086f..972037a 100644 --- a/test/storex/storex_test.exs +++ b/test/storex/storex_test.exs @@ -166,4 +166,14 @@ defmodule StorexTest do assert {:error, "Unauthorized"} = Storex.Supervisor.add_store(store, session, self(), %{}) end end + + describe "mutate return value" do + test "mutate/3 returns :ok" do + assert Storex.mutate("StorexTest.Store.Counter", "increase", []) == :ok + end + + test "mutate/4 returns :ok" do + assert Storex.mutate("key", "StorexTest.Store.Counter", "increase", []) == :ok + end + end end From 82a02edea5c3f34c5c579b4af8e066b4c966400e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 11:44:53 +0200 Subject: [PATCH 09/10] Drop dead configuration and check formatting in CI `Storex.start/2` carried `import Supervisor.Spec, warn: false`. `Supervisor.Spec` has been deprecated since Elixir 1.5 and nothing in the function used it; the `warn: false` is what kept that quiet. `config/config.exs` still set `registry: Storex.Registry.ETS`. That module was removed in 0.4.0 and nothing reads the key. `mix format --check-formatted` failed on the `@callback terminate/3` line in `lib/storex/store.ex`, and had for as long as the file has looked like that. With it formatted the check passes on the whole project, so CI can run it. The check is a separate job on one pinned Elixir rather than a step in the test matrix. The formatter's output can change between releases, and a step in the matrix would fail all thirteen rows the first time it does. `.formatter.exs` declares no `import_deps`, so the job needs no dependencies. --- .github/workflows/main.yml | 12 ++++++++++++ CHANGELOG.md | 1 + config/config.exs | 4 +--- lib/storex.ex | 2 -- lib/storex/store.ex | 3 ++- 5 files changed, 16 insertions(+), 6 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 798f7cc..bedfc60 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -61,3 +61,15 @@ jobs: run: mix compile --warnings-as-errors - name: Run Tests run: mix test + + format: + runs-on: ubuntu-24.04 + name: Format + steps: + - uses: actions/checkout@v4 + - uses: erlef/setup-beam@v1 + with: + otp-version: "28" + elixir-version: "1.20" + - name: Check formatting + run: mix format --check-formatted diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ab198b..f326ad9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- Housekeeping: dropped the unused `import Supervisor.Spec, warn: false` from `Storex.start/2` (deprecated since Elixir 1.5, and the `warn: false` was hiding it) and the `registry: Storex.Registry.ETS` key from `config/config.exs` (that module was removed in 0.4.0 and nothing read the key). The project is now formatted, and CI checks it - `Storex.Registry` reads (`get_store/2`, `get_store_pid/2`, `get_store_instances/1`, `session_stores/1`) now run in the calling process against the `:protected` ETS table instead of a `GenServer.call`. The registry process was a global serialisation point — every mutation performs at least one lookup — and the cost grew with the number of connections. Measured at 500 lookups per reader: 98.3ms against 16.2ms with 64 concurrent readers, 335.8ms against 56.2ms with 256. Writes and the `:DOWN` cleanup still go through the process - **[BREAKING]** `Storex.mutate/3` and `Storex.mutate/4` return `:ok`. They used to return whatever the comprehension in `Storex.PG.broadcast/1` produced, which was the internal broadcast envelope repeated once per node (`[broadcast: {:mutate, "Store", "reload", []}]`) - Removed the unreachable `{:error, _}` branch in `Storex.PG.broadcast/1`. `:pg.get_members/2` always returns a list; the error tuple was `:pg2`'s contract, and `:pg2` support went away in this release diff --git a/config/config.exs b/config/config.exs index 990dd39..1af521d 100644 --- a/config/config.exs +++ b/config/config.exs @@ -1,8 +1,6 @@ import Config -config :storex, - session_id_library: Nanoid, - registry: Storex.Registry.ETS +config :storex, session_id_library: Nanoid if Mix.env() == :test do import_config "test.exs" diff --git a/lib/storex.ex b/lib/storex.ex index 24c6e45..f981f92 100644 --- a/lib/storex.ex +++ b/lib/storex.ex @@ -3,8 +3,6 @@ defmodule Storex do @doc false def start(_type, _args) do - import Supervisor.Spec, warn: false - children = [ %{id: :pg, start: {:pg, :start_link, [Storex.PG]}}, {Storex.PG, []}, diff --git a/lib/storex/store.ex b/lib/storex/store.ex index 925df52..73ecae5 100644 --- a/lib/storex/store.ex +++ b/lib/storex/store.ex @@ -20,7 +20,8 @@ defmodule Storex.Store do @doc """ Called when store session ends. """ - @callback terminate(session_id :: binary(), params :: %{binary() => any()}, state :: any()) :: any() + @callback terminate(session_id :: binary(), params :: %{binary() => any()}, state :: any()) :: + any() @optional_callbacks terminate: 3 @doc false From bcdcbfae50f800ef51fec2abfeaa1a0d89bfdb1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krystian=20Dro=C5=BCd=C5=BCy=C5=84ski?= Date: Mon, 7 Sep 2026 18:02:06 +0200 Subject: [PATCH 10/10] Cover the behaviour the suite was not testing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Coverage found four gaps, and reading the tests found three more that coverage cannot see because the code runs during teardown without anything asserting on it. - `terminate/3` had never been executed. No fixture implemented it, so the whole optional-callback path was untested. There is a fixture for it now, and tests that it runs on `remove_store/2` with the state as of the last mutation. - The registry drops a row when the store process dies. That ran during teardown but nothing asserted it, which matters because the `:DOWN` handler is one of the match patterns that has to stay in step with the shape of the table. - Closing a connection stops the session's stores. Same situation. - Ping and pong. The client sends one every 30 seconds, making it the most frequently handled frame in production and the only one with no test. - `Storex.mutate/3` reaching every session of a store. The existing tests used a single session, so the fan-out — the entire point of the function — was never exercised. Also covers a session on a different store not being reached. - Joining a store that is already joined in the session returns the existing key and starts nothing. - `Storex.Diff` comparing a struct against a plain map, in both directions. The existing struct test compared two structs. Writing the first of those turned up a real defect. `Storex.Store.__terminate__/4` gated the callback on `function_exported?/3`, which answers `false` for a module that is not loaded. Under the normal flow the store is loaded by the time it terminates, because `init/2` was applied on it, so this never showed up — but whether a documented callback runs should not depend on what the code server happens to be holding. It checks `Code.ensure_loaded?/1` first now. Every test here was confirmed to fail against the unfixed code: seven deliberate breakages produce twelve red tests, and none on the code as it stands. --- CHANGELOG.md | 1 + lib/storex/store.ex | 5 ++- test/fixtures/stores/terminating.ex | 24 +++++++++++ test/storex/diff_test.exs | 12 ++++++ test/storex/handler/cowboy_test.exs | 65 +++++++++++++++++++++++++++++ test/storex/handler/plug_test.exs | 65 +++++++++++++++++++++++++++++ test/storex/registry_test.exs | 36 ++++++++++++++++ test/storex/store_test.exs | 29 +++++++++++++ test/storex/storex_test.exs | 48 +++++++++++++++++++++ test/storex/supervisor_test.exs | 21 ++++++++++ 10 files changed, 305 insertions(+), 1 deletion(-) create mode 100644 test/fixtures/stores/terminating.ex diff --git a/CHANGELOG.md b/CHANGELOG.md index f326ad9..8e3ee13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - **[BREAKING]** Updated the minimum required version of Elixir to `1.16` - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer - **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it +- `terminate/3` is no longer skipped when the store module happens not to be loaded yet. `Storex.Store.__terminate__/4` gated the call on `function_exported?/3`, which answers `false` for an unloaded module, so whether the callback ran depended on the code server rather than on the store - Housekeeping: dropped the unused `import Supervisor.Spec, warn: false` from `Storex.start/2` (deprecated since Elixir 1.5, and the `warn: false` was hiding it) and the `registry: Storex.Registry.ETS` key from `config/config.exs` (that module was removed in 0.4.0 and nothing read the key). The project is now formatted, and CI checks it - `Storex.Registry` reads (`get_store/2`, `get_store_pid/2`, `get_store_instances/1`, `session_stores/1`) now run in the calling process against the `:protected` ETS table instead of a `GenServer.call`. The registry process was a global serialisation point — every mutation performs at least one lookup — and the cost grew with the number of connections. Measured at 500 lookups per reader: 98.3ms against 16.2ms with 64 concurrent readers, 335.8ms against 56.2ms with 256. Writes and the `:DOWN` cleanup still go through the process - **[BREAKING]** `Storex.mutate/3` and `Storex.mutate/4` return `:ok`. They used to return whatever the comprehension in `Storex.PG.broadcast/1` produced, which was the internal broadcast envelope repeated once per node (`[broadcast: {:mutate, "Store", "reload", []}]`) diff --git a/lib/storex/store.ex b/lib/storex/store.ex index 73ecae5..9f53bf3 100644 --- a/lib/storex/store.ex +++ b/lib/storex/store.ex @@ -109,8 +109,11 @@ defmodule Storex.Store do end @doc false + # `function_exported?/3` answers `false` for a module that is not loaded yet, + # so the callback would be skipped silently. `ensure_loaded?/1` first makes the + # answer depend on the store, not on what the code server happens to hold. def __terminate__(store, session, params, state) do - if :erlang.function_exported(store, :terminate, 3) do + if Code.ensure_loaded?(store) and function_exported?(store, :terminate, 3) do apply(store, :terminate, [session, params, state]) end end diff --git a/test/fixtures/stores/terminating.ex b/test/fixtures/stores/terminating.ex new file mode 100644 index 0000000..baf5b2c --- /dev/null +++ b/test/fixtures/stores/terminating.ex @@ -0,0 +1,24 @@ +defmodule StorexTest.Store.Terminating do + @moduledoc """ + Implements the optional `terminate/3` callback and reports it to the pid found + under the `"reporter"` param. The pid is kept out of the state so the state + stays JSON encodable. + """ + + use Storex.Store + + def init(_session, _params) do + {:ok, %{counter: 0}} + end + + def mutation("increase", _data, _session_id, _params, state) do + {:noreply, %{state | counter: state.counter + 1}} + end + + def terminate(session, params, state) do + case Map.get(params, "reporter") do + nil -> :ok + pid -> send(pid, {:terminated, session, params, state}) + end + end +end diff --git a/test/storex/diff_test.exs b/test/storex/diff_test.exs index 77120aa..7a45d68 100644 --- a/test/storex/diff_test.exs +++ b/test/storex/diff_test.exs @@ -43,6 +43,18 @@ defmodule StorexTest.Diff do assert Enum.member?(diff, %{a: "u", p: [:age], t: 10}) end + test "diff struct against a plain map" do + diff = Storex.Diff.check(%Struct{name: "A", age: 1}, %{name: "B", age: 1}) + + assert diff == [%{a: "u", p: [:name], t: "B"}] + end + + test "diff plain map against a struct" do + diff = Storex.Diff.check(%{name: "A", age: 1}, %Struct{name: "B", age: 1}) + + assert diff == [%{a: "u", p: [:name], t: "B"}] + end + test "diff DateTime" do assert [%{a: "u", p: [], t: "2000-02-29 23:10:00+01:00 CET Europe/Warsaw"}] = Storex.Diff.check( diff --git a/test/storex/handler/cowboy_test.exs b/test/storex/handler/cowboy_test.exs index a03a4dd..10de2a8 100644 --- a/test/storex/handler/cowboy_test.exs +++ b/test/storex/handler/cowboy_test.exs @@ -225,6 +225,71 @@ defmodule StorexTest.Handler.Cowboy do end end + describe "keepalive" do + test "a ping is answered with a pong carrying the same request id", context do + client = tcp_client(context) + http1_handshake(client) + + request = random_string() + + send_text_frame(client, """ + { + "type": "ping", + "request": "#{request}" + } + """) + + {:ok, result} = recv_text_frame(client) + + assert %{type: "pong", request: ^request} = Jason.decode!(result, keys: :atoms) + end + + test "a ping does not need a store to be joined", context do + client = tcp_client(context) + http1_handshake(client) + + send_text_frame(client, ~s({"type": "ping", "request": "#{random_string()}"})) + + assert {:ok, _} = recv_text_frame(client) + end + end + + describe "session cleanup" do + test "closing the connection stops the session's stores", context do + client = tcp_client(context) + http1_handshake(client) + + send_text_frame(client, """ + { + "type": "join", + "store": "StorexTest.Store.Counter", + "data": {}, + "request": "#{random_string()}" + } + """) + + {:ok, result} = recv_text_frame(client) + assert %{session: session} = Jason.decode!(result, keys: :atoms) + + store_pid = Storex.Registry.get_store_pid("StorexTest.Store.Counter", session) + assert is_pid(store_pid) + + :gen_tcp.close(client) + + assert Enum.reduce_while(1..200, false, fn _, _ -> + if Storex.Registry.session_stores(session) == [] do + {:halt, true} + else + Process.sleep(10) + {:cont, false} + end + end), + "the session's registry rows were not cleaned up" + + refute Process.alive?(store_pid) + end + end + # Simple WebSocket client def tcp_client(context) do diff --git a/test/storex/handler/plug_test.exs b/test/storex/handler/plug_test.exs index 52e6a41..41d0d66 100644 --- a/test/storex/handler/plug_test.exs +++ b/test/storex/handler/plug_test.exs @@ -229,6 +229,71 @@ defmodule StorexTest.Handler.Plug do end end + describe "keepalive" do + test "a ping is answered with a pong carrying the same request id", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + request = random_string() + + send_text_frame(client, """ + { + "type": "ping", + "request": "#{request}" + } + """) + + {:ok, result} = recv_text_frame(client) + + assert %{type: "pong", request: ^request} = Jason.decode!(result, keys: :atoms) + end + + test "a ping does not need a store to be joined", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + send_text_frame(client, ~s({"type": "ping", "request": "#{random_string()}"})) + + assert {:ok, _} = recv_text_frame(client) + end + end + + describe "session cleanup" do + test "closing the connection stops the session's stores", context do + client = tcp_client(context) + http1_handshake(client, Storex.Handler.Plug) + + send_text_frame(client, """ + { + "type": "join", + "store": "StorexTest.Store.Counter", + "data": {}, + "request": "#{random_string()}" + } + """) + + {:ok, result} = recv_text_frame(client) + assert %{session: session} = Jason.decode!(result, keys: :atoms) + + store_pid = Storex.Registry.get_store_pid("StorexTest.Store.Counter", session) + assert is_pid(store_pid) + + :gen_tcp.close(client) + + assert Enum.reduce_while(1..200, false, fn _, _ -> + if Storex.Registry.session_stores(session) == [] do + {:halt, true} + else + Process.sleep(10) + {:cont, false} + end + end), + "the session's registry rows were not cleaned up" + + refute Process.alive?(store_pid) + end + end + # Simple WebSocket client def tcp_client(context) do diff --git a/test/storex/registry_test.exs b/test/storex/registry_test.exs index fd0d32a..922fd0a 100644 --- a/test/storex/registry_test.exs +++ b/test/storex/registry_test.exs @@ -3,6 +3,17 @@ defmodule StorexTest.RegistryTest do @registry :storex_registry + defp eventually(check, attempts \\ 100) do + Enum.reduce_while(1..attempts, false, fn _, _ -> + if check.() do + {:halt, true} + else + Process.sleep(10) + {:cont, false} + end + end) + end + setup do session = "session-#{System.unique_integer([:positive])}" {:ok, _} = Storex.Registry.register_store("Store", self(), session, self(), "key") @@ -37,6 +48,31 @@ defmodule StorexTest.RegistryTest do assert Storex.Registry.get_store_pid("Never.Joined", session) == :undefined end + test "a row is dropped when the store process dies" do + session = "session-#{System.unique_integer([:positive])}" + store_pid = spawn(fn -> Process.sleep(:infinity) end) + + {:ok, _} = Storex.Registry.register_store("Dying", store_pid, session, self(), nil) + assert Storex.Registry.get_store_pid("Dying", session) == store_pid + + Process.exit(store_pid, :kill) + + # The registry monitors the store, so the row goes when the process does. + assert eventually(fn -> Storex.Registry.get_store_pid("Dying", session) == :undefined end) + assert Storex.Registry.session_stores(session) == [] + end + + test "only the dead process's rows are dropped", %{session: session} do + other = "session-#{System.unique_integer([:positive])}" + store_pid = spawn(fn -> Process.sleep(:infinity) end) + + {:ok, _} = Storex.Registry.register_store("Dying", store_pid, other, self(), nil) + Process.exit(store_pid, :kill) + + assert eventually(fn -> Storex.Registry.get_store_pid("Dying", other) == :undefined end) + assert Storex.Registry.get_store_pid("Store", session) == self() + end + test "reads do not go through the registry process", %{session: session} do # With the owner suspended, anything that needs a GenServer round-trip # blocks. Reads run in the calling process, so they still answer. diff --git a/test/storex/store_test.exs b/test/storex/store_test.exs index 707717d..bf0ed89 100644 --- a/test/storex/store_test.exs +++ b/test/storex/store_test.exs @@ -6,6 +6,7 @@ defmodule StorexTest.StoreTest do alias StorexTest.Store.InvalidInit alias StorexTest.Store.InvalidMutation alias StorexTest.Store.KeyInit + alias StorexTest.Store.Terminating alias StorexTest.Store.Text describe "resolve/1" do @@ -102,6 +103,22 @@ defmodule StorexTest.StoreTest do end end + describe "terminate dispatch" do + test "the callback is invoked with the session, params and state" do + params = %{"reporter" => self()} + + assert Storex.Store.__terminate__(Terminating, "session", params, %{counter: 3}) == + {:terminated, "session", params, %{counter: 3}} + + assert_received {:terminated, "session", ^params, %{counter: 3}} + end + + test "a store that does not implement it is a no-op" do + refute function_exported?(Counter, :terminate, 3) + assert Storex.Store.__terminate__(Counter, "session", %{}, %{counter: 0}) == nil + end + end + describe "store server" do setup do session = "session-#{System.unique_integer([:positive])}" @@ -132,6 +149,18 @@ defmodule StorexTest.StoreTest do Storex.Supervisor.add_store("StorexTest.Store.ErrorInit", session, self(), %{}) end + test "stopping a store runs terminate/3 with the current state", %{session: session} do + store = "StorexTest.Store.Terminating" + params = %{"reporter" => self()} + + {:ok, _} = Storex.Supervisor.add_store(store, session, self(), params) + {:ok, _} = Storex.Supervisor.mutate_store(session, store, "increase", []) + + Storex.Supervisor.remove_store(session, store) + + assert_receive {:terminated, ^session, ^params, %{counter: 1}}, 1000 + end + test "mutating through the server returns the state diff", %{session: session} do {:ok, _} = Storex.Supervisor.add_store("StorexTest.Store.Counter", session, self(), %{}) diff --git a/test/storex/storex_test.exs b/test/storex/storex_test.exs index 972037a..595dd49 100644 --- a/test/storex/storex_test.exs +++ b/test/storex/storex_test.exs @@ -67,6 +67,54 @@ defmodule StorexTest do assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(session, store) end + test "reaches every session of the store", %{session: session, store: store, pid: pid} do + # The point of mutate/3 is the fan-out. One session proves nothing. + other_session = Application.get_env(:storex, :session_id_library, Nanoid).generate() + + {:ok, other_pid} = + GenServer.start_link(FakeWebsocketServer, [self(), other_session], + name: {:global, other_session} + ) + + assert {:ok, _} = Storex.Supervisor.add_store(store, session, pid, %{}) + assert {:ok, _} = Storex.Supervisor.add_store(store, other_session, other_pid, %{}) + + Storex.mutate(store, "increase", []) + + assert_receive :ok + assert_receive :ok + + assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(session, store) + assert {:ok, %{counter: 1}} = Storex.Supervisor.get_store_state(other_session, store) + + Storex.Supervisor.remove_store(other_session, store) + end + + test "does not reach a session that joined a different store", %{ + session: session, + store: store, + pid: pid + } do + other_session = Application.get_env(:storex, :session_id_library, Nanoid).generate() + + {:ok, other_pid} = + GenServer.start_link(FakeWebsocketServer, [self(), other_session], + name: {:global, other_session} + ) + + assert {:ok, _} = Storex.Supervisor.add_store(store, session, pid, %{}) + + assert {:ok, _} = + Storex.Supervisor.add_store("StorexTest.Store.Text", other_session, other_pid, %{}) + + Storex.mutate(store, "increase", []) + + assert_receive :ok + refute_receive :ok, 200 + + Storex.Supervisor.remove_store(other_session, "StorexTest.Store.Text") + end + test "mutate store in cluster", %{session: session, store: store, pid: pid} do [node_1] = LocalCluster.start_nodes(:spawn, 1, diff --git a/test/storex/supervisor_test.exs b/test/storex/supervisor_test.exs index b6d6683..116730a 100644 --- a/test/storex/supervisor_test.exs +++ b/test/storex/supervisor_test.exs @@ -32,6 +32,27 @@ defmodule StorexTest.SupervisorTest do assert atoms_after - atoms_before == 0 end + test "joining a store twice reuses the running one", %{} do + session = start_store(session()) + pid = Storex.Registry.get_store_pid(@store, session) + + # The second add_store/4 takes the registry branch and starts nothing. + assert {:ok, nil} = Storex.Supervisor.add_store(@store, session, self(), %{}) + + assert Storex.Registry.get_store_pid(@store, session) == pid + assert length(Storex.Registry.session_stores(session)) == 1 + end + + test "joining a store twice returns the key from the first join" do + store = "StorexTest.Store.KeyInit" + session = session() + + assert {:ok, "user_id"} = Storex.Supervisor.add_store(store, session, self(), %{}) + assert {:ok, "user_id"} = Storex.Supervisor.add_store(store, session, self(), %{}) + + on_exit(fn -> Storex.Supervisor.remove_store(session, store) end) + end + test "the same session and store cannot be started twice" do session = start_store(session())