diff --git a/lib/phoenix/transports/long_poll.ex b/lib/phoenix/transports/long_poll.ex index 4085e1abf7..d735afd19f 100644 --- a/lib/phoenix/transports/long_poll.ex +++ b/lib/phoenix/transports/long_poll.ex @@ -174,14 +174,10 @@ defmodule Phoenix.Transports.LongPoll do receive do {:messages, messages, ^ref} -> {:ok, messages} after - opts[:window_ms] -> - broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + opts[:window_ms] -> expire_listen(endpoint, server_ref, client_ref, ref, opts) end after - opts[:window_ms] -> - broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + opts[:window_ms] -> expire_listen(endpoint, server_ref, client_ref, ref, opts) end conn @@ -189,6 +185,27 @@ defmodule Phoenix.Transports.LongPoll do |> status_token_messages_json(token, messages) end + # The session drops the messages it sends, and it can still answer our flush + # after we stopped waiting. It handles our messages in order, so we ask it to + # echo a subscribe: if it answered the flush, the messages arrive before the echo. + defp expire_listen(endpoint, server_ref, client_ref, ref, opts) do + echo_ref = make_ref() + broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) + broadcast_from!(endpoint, server_ref, {:subscribe, client_ref, echo_ref}) + + receive do + # The echo follows the messages. We deliberately leave it in the mailbox, + # as waiting for it would delay the response. With Cowboy, the request + # process stops after the response. Bandit ignores unhandled messages. + {:messages, messages, ^ref} -> {:ok, messages} + {:subscribe, ^echo_ref} -> {:no_content, []} + after + # We cannot tell if the session still answers the flush, so we fail + # and the client starts a new session instead of missing messages. + opts[:pubsub_timeout_ms] -> {:gone, []} + 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/test/phoenix/integration/long_poll_channels_test.exs b/test/phoenix/integration/long_poll_channels_test.exs index 2513f61b66..c23ede9a7a 100644 --- a/test/phoenix/integration/long_poll_channels_test.exs +++ b/test/phoenix/integration/long_poll_channels_test.exs @@ -355,6 +355,51 @@ defmodule Phoenix.Integration.LongPollChannelsTest do resp end + @doc """ + Makes the session wait before it handles its next flush. + + The caller receives `{:flush_blocked, server, release}` once the session + waits. The session continues when it receives `release`. + """ + def block_next_flush(session) do + {:ok, {:v1, _id, server, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), session["token"]) + + parent = self() + release = make_ref() + + :sys.install( + server, + {fn + _state, {:in, {:flush, _client_ref, _ref}}, _proc_state -> + send(parent, {:flush_blocked, server, release}) + + receive do + ^release -> :done + end + + state, _event, _proc_state -> + state + end, nil} + ) + end + + defp await_expired(server, retries \\ 200) do + {:messages, messages} = Process.info(server, :messages) + + cond do + Enum.any?(messages, &match?({:expired, _client_ref, _ref}, &1)) -> + :ok + + retries == 0 -> + flunk("the listen did not expire") + + true -> + Process.sleep(5) + await_expired(server, retries - 1) + end + end + for %{adapter: adapter} <- [ %{adapter: Bandit.PhoenixAdapter}, %{adapter: Phoenix.Endpoint.Cowboy2Adapter} @@ -764,6 +809,40 @@ defmodule Phoenix.Integration.LongPollChannelsTest do resp = poll(:post, "/ws", @vsn, session) assert resp.body["status"] == 410 end + + test "#{@mode}: listen receives messages flushed after its window", %{topic: topic} do + session = join("/ws", topic, @vsn, "1", @mode) + block_next_flush(session) + + listen = Task.async(fn -> poll(:get, "/ws", @vsn, session) end) + assert_receive {:flush_blocked, server, release} + + # the session answers the flush only after the listen stopped waiting for it + await_expired(server) + send(server, release) + + resp = Task.await(listen) + assert resp.body["status"] == 200 + + assert [%Message{event: "phx_reply", ref: "1", topic: ^topic} | _] = + resp.body["messages"] + + resp = poll(:get, "/ws", @vsn, session) + assert resp.body["status"] == 204 + end + + test "#{@mode}: listen fails when the session does not answer after its window", + %{topic: topic} do + session = join("/ws", topic, @vsn, "1", @mode) + block_next_flush(session) + + resp = poll(:get, "/ws", @vsn, session) + assert resp.body["status"] == 410 + assert resp.body["messages"] == [] + + assert_received {:flush_blocked, server, release} + send(server, release) + end end test "publish responds with 408 when transport_dispatch times out" do