Don't drop LongPoll messages flushed after the listen window

lib/phoenix/transports/long_poll.ex

@@ -174,21 +174,38 @@ 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
     |> put_status(status)
     |> 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

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