diff --git a/CHANGELOG.md b/CHANGELOG.md index 87e0d54435..e6d2e04068 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Bug fixes + * [Phoenix.Transports.LongPoll] Keep a publish when its listen has already + expired, and return that publish on the next listen * [Phoenix.ChannelTest] Fix test sockets violating the Phoenix.Socket.t typespec, which made Dialyzer treat every test join as never returning ([#5509](https://github.com/phoenixframework/phoenix/issues/5509)) diff --git a/lib/phoenix/transports/long_poll.ex b/lib/phoenix/transports/long_poll.ex index 4085e1abf7..824c0b83ae 100644 --- a/lib/phoenix/transports/long_poll.ex +++ b/lib/phoenix/transports/long_poll.ex @@ -175,13 +175,11 @@ defmodule Phoenix.Transports.LongPoll do {:messages, messages, ^ref} -> {:ok, messages} after opts[:window_ms] -> - broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + expire_listen(endpoint, server_ref, client_ref, ref) end after opts[:window_ms] -> - broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + expire_listen(endpoint, server_ref, client_ref, ref) end conn @@ -189,6 +187,19 @@ defmodule Phoenix.Transports.LongPoll do |> status_token_messages_json(token, messages) end + # A publish can arrive in the same moment the timer fires. Take it when it + # is already in this mailbox. The session keeps its buffer when this expired + # notice is already queued there. + defp expire_listen(endpoint, server_ref, client_ref, ref) do + broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) + + receive do + {:messages, messages, ^ref} -> {:ok, messages} + after + 0 -> {:no_content, []} + end + end + # Accept the token from either location so that newer clients can use the # header while older clients continue to use params. defp fetch_token(conn) do diff --git a/lib/phoenix/transports/long_poll_server.ex b/lib/phoenix/transports/long_poll_server.ex index e3a5f23d8d..41af5fbde6 100644 --- a/lib/phoenix/transports/long_poll_server.ex +++ b/lib/phoenix/transports/long_poll_server.ex @@ -73,12 +73,21 @@ defmodule Phoenix.Transports.LongPoll.Server do end def handle_info({:flush, client_ref, ref}, state) do - case state.buffer do - [] -> + cond do + # This listen already gave up. A send now would miss it and drop the buffer. + expired_queued?(client_ref, ref) -> + {:noreply, + %{ + state + | client_ref: clear_client(state.client_ref, client_ref, ref), + last_client_poll: now_ms() + }} + + state.buffer == [] -> {:noreply, %{state | client_ref: {client_ref, ref}, last_client_poll: now_ms()}} - buffer -> - broadcast_from!(state, client_ref, {:messages, Enum.reverse(buffer), ref}) + true -> + broadcast_from!(state, client_ref, {:messages, Enum.reverse(state.buffer), ref}) {:noreply, %{state | client_ref: nil, last_client_poll: now_ms(), buffer: []}} end end @@ -153,6 +162,14 @@ defmodule Phoenix.Transports.LongPoll.Server do end end + defp expired_queued?(client_ref, ref) do + {:messages, mailbox} = Process.info(self(), :messages) + Enum.any?(mailbox, &match?({:expired, ^client_ref, ^ref}, &1)) + end + + defp clear_client({client_ref, ref}, client_ref, ref), do: nil + defp clear_client(client_ref, _client_ref, _ref), do: client_ref + defp now_ms, do: System.system_time(:millisecond) defp schedule_inactive_shutdown(window_ms) do diff --git a/test/phoenix/integration/long_poll_channels_test.exs b/test/phoenix/integration/long_poll_channels_test.exs index 2513f61b66..65574618e1 100644 --- a/test/phoenix/integration/long_poll_channels_test.exs +++ b/test/phoenix/integration/long_poll_channels_test.exs @@ -281,6 +281,74 @@ defmodule Phoenix.Integration.LongPollChannelsTest do defp encode(V1.JSONSerializer, %{} = map), do: Phoenix.json_library().encode!(map) + # Queues a join behind the heartbeat, then holds the session before that + # join runs. The listen's later flush is already behind the join. + defp hold_expired_join(state, body, opcode, client, parent, release) do + cond do + String.contains?(body, "\"heartbeat\"") -> + join = ~s(["join-1","join-1","room:lobby","phx_join",{}]) + send(self(), {:dispatch, client, {join, opcode}, make_ref()}) + state + + String.contains?(body, "\"phx_join\"") -> + send(parent, :join_blocked) + + receive do + {^release, :release} -> state + end + + true -> + state + end + end + + defp await_parked_listen(pid) do + Enum.reduce_while(1..100, :waiting, fn _, _ -> + case :sys.get_state(pid) do + %{client_ref: {listen_pid, _ref}} when is_pid(listen_pid) -> + {:halt, listen_pid} + + _ -> + Process.sleep(2) + {:cont, :waiting} + end + end) + |> case do + listen_pid when is_pid(listen_pid) -> listen_pid + :waiting -> flunk("long-poll listen did not park") + end + end + + defp await_expired_flush(server) do + Enum.reduce_while(1..100, :waiting, fn _, _ -> + case Process.info(server, :messages) do + {:messages, mailbox} -> + flush? = Enum.any?(mailbox, &match?({:flush, _, _}, &1)) + expired? = Enum.any?(mailbox, &match?({:expired, _, _}, &1)) + + if flush? and expired? do + {:halt, :ok} + else + Process.sleep(20) + {:cont, :waiting} + end + + nil -> + flunk("long-poll session ended before the expired flush was queued") + end + end) + |> case do + :ok -> :ok + :waiting -> flunk("second flush and expired notice were not both queued") + end + end + + defp resume_if_suspended(pid) do + if Process.alive?(pid) and match?({:status, :suspended}, Process.info(pid, :status)) do + :erlang.resume_process(pid) + end + end + @doc """ Joins a long poll socket. @@ -788,6 +856,78 @@ defmodule Phoenix.Integration.LongPollChannelsTest do assert resp.body["status"] == 408 end + test "a reply published while a listen has expired is on the next listen" do + resp = poll(:get, "/ws", "2.0.0", %{}, nil) + assert resp.body["status"] == 410 + assert resp.status == 200 + token = resp.body["token"] + session = %{"token" => token} + + {:ok, {:v1, _id, server, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), token) + + release = make_ref() + parent = self() + + :ok = + :sys.install( + server, + {fn state, event, _name -> + case event do + {:in, {:dispatch, client, {body, opcode}, _ref}} when is_binary(body) -> + hold_expired_join(state, body, opcode, client, parent, release) + + _ -> + state + end + end, nil} + ) + + on_exit(fn -> + if Process.alive?(server) do + send(server, {release, :release}) + Process.exit(server, :kill) + end + end) + + listen = Task.async(fn -> poll(:get, "/ws", "2.0.0", session) end) + listen_pid = await_parked_listen(server) + on_exit(fn -> resume_if_suspended(listen_pid) end) + :erlang.suspend_process(listen_pid) + + poster = + Task.async(fn -> + poll(:post, "/ws", "2.0.0", session, %{ + "topic" => "phoenix", + "event" => "heartbeat", + "ref" => "hb-1", + "payload" => %{} + }) + end) + + assert_receive :join_blocked, 1_000 + :erlang.resume_process(listen_pid) + assert await_expired_flush(server) + send(server, {release, :release}) + + first = Task.await(listen, 5_000) + assert first.status == 200 + assert first.body["status"] == 204 + assert first.body["messages"] == [] + + post = Task.await(poster, 5_000) + assert post.status == 200 + assert post.body["status"] == 200 + + next = poll(:get, "/ws", "2.0.0", session) + assert next.status == 200 + assert next.body["status"] == 200 + + assert Enum.any?(next.body["messages"], fn message -> + message.ref == "join-1" or message.join_ref == "join-1" + end) + end + test "ignores ndjson batch entries after the first 100", %{topic: topic} do vsn = "2.0.0" join_ref = "1"