From de624333bd3add81a3a14dce18105379892d1483 Mon Sep 17 00:00:00 2001 From: Steffen Deusch Date: Fri, 6 Nov 2026 14:04:59 +0100 Subject: [PATCH] Don't drop LongPoll messages flushed after the listen window A listen that waited for `window_ms` sent `:expired` to the session and returned 204 right away. The session could still handle a flush that the listen had sent before. It then sent its buffered messages to a process that no longer read them and cleared the buffer, so the next listen received nothing. This happens when the session is busy for longer than `window_ms` while a flush of the listen is queued. The listen now waits for the session before it returns. After the window, it sends `:expired` followed by `:subscribe`, which the session already echoes. The session receives the messages of a listen in the order they were sent, so the echo tells the listen that no flush is left: * If the session answered a flush, the messages arrive before the echo and the listen returns them. * If the echo arrives first, the listen returns 204 as before. The session keeps what it buffers afterwards for the next listen. * If neither arrives within `pubsub_timeout_ms`, the listen responds with status 410. It cannot tell whether the session still answers the flush. phoenix.js then starts a new session and rejoins its channels, as it does when a poll cannot resume its session. When the messages arrive first, the echo stays unread. With Cowboy, the request process stops after the response. Bandit ignores unhandled messages. Only the listen changes. The session code is the same, so nodes with and without this change can serve the same session during a rolling deploy. A listen served by a node without the change keeps the old behavior. A poll against a session that stopped or is blocked now returns 410 after `window_ms` plus `pubsub_timeout_ms`. Before, it returned 204 after `window_ms`, and the next poll returned 410. Messages can still be lost when the response does not reach the client. That needs acknowledgements from the client, see #6885. The tests hold the session before it handles a flush. They cover a session that continues after the listen window and one that does not answer in time, for Bandit and Cowboy, with direct and PubSub delivery. The problem was reported in #6899. --- lib/phoenix/transports/long_poll.ex | 29 +++++-- .../integration/long_poll_channels_test.exs | 79 +++++++++++++++++++ 2 files changed, 102 insertions(+), 6 deletions(-) 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