fix: add system message to signal replication ready

Author: filipecabaco

ERROR_CODES.md

@@ -12,9 +12,12 @@ defmodule Realtime.SynHandler do
   @postgres_cdc_scope_prefix PostgresCdc.syn_topic_prefix()
 
   @impl true
-  def on_registry_process_updated(Connect, tenant_id, pid, %{conn: conn}, :normal) when is_pid(conn) do
-    # Update that a database connection is ready
-    Endpoint.local_broadcast(Connect.syn_topic(tenant_id), "ready", %{pid: pid, conn: conn})
+  def on_registry_process_updated(Connect, tenant_id, pid, %{conn: conn} = meta, :normal) when is_pid(conn) do
+    Endpoint.local_broadcast(Connect.syn_topic(tenant_id), "ready", %{
+      pid: pid,
+      conn: conn,
+      replication_conn: Map.get(meta, :replication_conn)
+    })
   end
 
   def on_registry_process_updated(scope, tenant_id, _pid, meta, _reason) do

lib/realtime/syn_handler.ex

@@ -104,6 +104,8 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       RealtimeWeb.Endpoint.subscribe(tenant_topic, metadata: metadata)
       RealtimeWeb.Endpoint.subscribe("realtime:operations:" <> tenant_id, metadata: metadata)
+      RealtimeWeb.Endpoint.subscribe(Connect.syn_topic(tenant_id))
+      send(self(), :notify_replication_ready)
 
       is_new_api = new_api?(params)
       presence_enabled? = socket.assigns.presence_enabled?
@@ -133,7 +135,8 @@ defmodule RealtimeWeb.RealtimeChannel do
         self_broadcast: !!params["config"]["broadcast"]["self"],
         tenant_topic: tenant_topic,
         channel_name: sub_topic,
-        presence_enabled?: presence_enabled?
+        presence_enabled?: presence_enabled?,
+        replication_ready_notified?: false
       }
 
       socket =
@@ -290,6 +293,20 @@ defmodule RealtimeWeb.RealtimeChannel do
     {:noreply, assign(socket, %{pg_sub_ref: pg_sub_ref})}
   end
 
+  def handle_info(:notify_replication_ready, %{assigns: %{tenant: tenant_id}} = socket) do
+    case Connect.replication_status(tenant_id) do
+      {:ok, _replication_conn} -> {:noreply, push_replication_ready(socket)}
+      {:error, :not_connected} -> {:noreply, socket}
+    end
+  end
+
+  def handle_info(%{event: "ready", payload: %{replication_conn: pid}}, socket)
+      when is_pid(pid) do
+    {:noreply, push_replication_ready(socket)}
+  end
+
+  def handle_info(%{topic: "connect:" <> _}, socket), do: {:noreply, socket}
+
   def handle_info(_msg, %{assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{read: false}}}} = socket) do
     Logger.warning("Broadcast message ignored")
     {:noreply, socket}
@@ -723,6 +740,14 @@ defmodule RealtimeWeb.RealtimeChannel do
     {:stop, :normal, socket}
   end
 
+  defp push_replication_ready(%{assigns: %{replication_ready_notified?: true}} = socket), do: socket
+
+  defp push_replication_ready(%{assigns: %{tenant: tenant_id, channel_name: channel_name}} = socket) do
+    RealtimeWeb.Endpoint.unsubscribe(Connect.syn_topic(tenant_id))
+    push_system_message("broadcast", socket, "ok", "Replication connection established", channel_name)
+    assign(socket, :replication_ready_notified?, true)
+  end
+
   defp push_system_message(extension, socket, status, error, channel_name)
        when is_map(error) and is_map_key(error, :error_code) and is_map_key(error, :error_message) do
     push(socket, "system", %{

lib/realtime_web/channels/realtime_channel.ex

@@ -74,3 +74,59 @@ run = """
     --db-url "postgresql://postgres:postgres@127.0.0.1:54322/postgres" \
     "$@"
 """
+
+[tasks.e2e-local]
+description = "Run e2e tests against a locally-built realtime image (carries unreleased migrations). Builds the repo, swaps it in for supabase's stock realtime, then runs the suite. e.g. mise run e2e-local -- --test postgres-changes-filters"
+dir = "test/e2e"
+run = """
+  set -e
+  REF=$(grep -E '^[[:space:]]*project_id' supabase/config.toml | sed -E 's/.*"(.*)".*/\\1/')
+  IMAGE=realtime:e2e-local
+  RT="supabase_realtime_${REF}"
+
+  # 1. Build the repo into a local image (Docker layer cache keeps rebuilds cheap).
+  docker build -t "$IMAGE" ../..
+
+  # 2. Bring up the full supabase stack. We use its realtime container only as the source of
+  #    truth for env vars (JWT secret, DB creds, encryption key, etc.). A prior run may have left
+  #    realtime swapped out / removed; `supabase start` won't recreate an externally-removed
+  #    container, so if it's missing we recycle the stack to get a clean one back.
+  supabase start
+  if ! docker inspect "$RT" >/dev/null 2>&1; then
+    echo "realtime container missing (left over from a prior swap); recreating supabase stack..."
+    supabase stop >/dev/null 2>&1 || true
+    supabase start
+  fi
+
+  # 3. Swap realtime: copy supabase's exact env (via `env` inside their container, so we avoid
+  #    Go-template braces that collide with mise's templating), drop their container, and run ours
+  #    under the same name / network / aliases / healthcheck so Kong routes to it unchanged.
+  ENVFILE=$(mktemp)
+  docker exec "$RT" env > "$ENVFILE"
+  docker rm -f "$RT" >/dev/null
+  docker run -d --name "$RT" \
+    --network "supabase_network_${REF}" \
+    --network-alias realtime --network-alias realtime-dev \
+    --env-file "$ENVFILE" \
+    --health-cmd 'curl -sSfL --head -o /dev/null -H "Host:realtime-dev" http://127.0.0.1:4000/api/ping' \
+    --health-interval 10s --health-timeout 2s --health-retries 3 \
+    "$IMAGE" >/dev/null
+  rm -f "$ENVFILE"
+
+  # 4. Wait for our realtime to report healthy (boot + tenant migrations). `docker ps` with a
+  #    health filter prints the container id only once it is healthy — no Go-template needed.
+  echo "Waiting for local realtime ($IMAGE) to become healthy..."
+  for _ in $(seq 1 30); do
+    if [ -n "$(docker ps -q --filter "name=$RT" --filter health=healthy)" ]; then break; fi
+    sleep 2
+  done
+
+  # 5. Run the suite against the stack (Kong -> our realtime). Do NOT call `supabase start` again
+  #    here, or it would recreate realtime from the stock image.
+  eval "$(supabase status --output env)"
+  bun run realtime-check.ts --env local \
+    --publishable-key "$PUBLISHABLE_KEY" \
+    --secret-key "$SECRET_KEY" \
+    --db-url "postgresql://postgres:postgres@127.0.0.1:54322/postgres" \
+    "$@"
+"""

mise.toml

@@ -274,16 +274,24 @@ async function waitForSubscribed(channel: ReturnType<SupabaseClient["channel"]>)
 // Subscribes a channel and waits until it is fully joined.
 // All data operations must happen after this returns to avoid delivery races.
 async function openChannel(channel: ReturnType<SupabaseClient["channel"]>): Promise<number> {
+  let replicationReady = false;
+  channel.on("system", "*", ({ message }: { message?: string }) => {
+    if (message === "Replication connection established") replicationReady = true;
+  });
   channel.subscribe();
-  return waitForSubscribed(channel);
+  const subscribeMs = await waitForSubscribed(channel);
+  await waitFor(() => replicationReady ? true : null, "replication established");
+  return subscribeMs;
 }
 
 // Subscribes a postgres_changes channel and waits for both the join and the
 // system:ok confirmation that the server-side WAL subscription is active.
 async function openPostgresChannel(channel: ReturnType<SupabaseClient["channel"]>): Promise<{ subscribeMs: number; systemMs: number }> {
   const start = performance.now();
   let systemOk = false;
-  channel.on("system", "*", ({ status }: { status: string }) => { if (status === "ok") systemOk = true; });
+  channel.on("system", "*", ({ status, extension }: { status: string; extension?: string }) => {
+    if (status === "ok" && extension === "postgres_changes") systemOk = true;
+  });
   const subscribeMs = await openChannel(channel);
   const { latencyMs: systemMs } = await waitFor(() => systemOk ? true : null, "system ok");
   return { subscribeMs, systemMs: performance.now() - start };
@@ -1001,7 +1009,6 @@ async function runBroadcastChangesTests(_testUser: { email: string; password: st
         .on("broadcast", { event: "INSERT" }, (res) => (result = res));
 
       const subscribeMs = await openChannel(channel);
-      await sleep(500);
       await supabase.from("broadcast_changes").insert({ value, id, topic: testTopic });
       const { latencyMs: eventMs } = await waitFor(() => result, "INSERT event");
 
@@ -1031,7 +1038,6 @@ async function runBroadcastChangesTests(_testUser: { email: string; password: st
         .on("broadcast", { event: "UPDATE" }, (res) => (result = res));
 
       const subscribeMs = await openChannel(channel);
-      await sleep(100);
       await supabase.from("broadcast_changes").insert({ value: originalValue, id, topic: testTopic });
       await supabase.from("broadcast_changes").update({ value: updatedValue }).eq("id", id);
       const { latencyMs: eventMs } = await waitFor(() => result, "UPDATE event");
@@ -1062,7 +1068,6 @@ async function runBroadcastChangesTests(_testUser: { email: string; password: st
         .on("broadcast", { event: "DELETE" }, (res) => (result = res));
 
       const subscribeMs = await openChannel(channel);
-      await sleep(100);
       await supabase.from("broadcast_changes").insert({ value, id, topic: testTopic });
       await supabase.from("broadcast_changes").delete().eq("id", id);
       const { latencyMs: eventMs } = await waitFor(() => result, "DELETE event");
@@ -1439,7 +1444,6 @@ async function runBroadcastBinaryTests(supabase: SupabaseClient) {
         .on("broadcast", { event }, (msg) => (result = msg.payload));
 
       const subscribeMs = await openChannel(channel);
-      await sleep(100);
 
       await sql`SELECT realtime.send_binary(${binary}::bytea, ${event}::text, ${topic}::text, true)`;
 

test/e2e/realtime-check.ts

@@ -110,6 +110,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
 
       # Join events
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
+
       # Broadcast event
       payload = %{"event" => "TEST", "payload" => %{"msg" => 1}, "type" => "broadcast"}
 
@@ -120,7 +121,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
         assert_receive %Message{topic: ^topic, event: "broadcast", payload: ^payload}
       end
 
-      refute_receive _any
+      refute_receive %Message{topic: ^topic, event: "broadcast"}
 
       # Wait for RateCounter to run
       RateCounterHelper.tick_tenant_rate_counters!(tenant.external_id)

test/integration/rt_channel/billable_events_test.exs

@@ -12,6 +12,7 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
   alias Phoenix.Socket.Message
   alias Realtime.Integration.WebsocketClient
   alias Realtime.Tenants
+  alias Realtime.Tenants.Connect
   alias RealtimeWeb.UserSocket
 
   @moduletag :capture_log
@@ -53,6 +54,60 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
     end
   end
 
+  describe "replication connection establishment" do
+    test "Connect signals replication readiness on its syn topic with the replication_conn pid" do
+      tenant = Containers.checkout_tenant(run_migrations: true)
+
+      Phoenix.PubSub.subscribe(Realtime.PubSub, Connect.syn_topic(tenant.external_id))
+      {:ok, _db_conn} = Connect.lookup_or_start_connection(tenant.external_id)
+
+      assert_receive %Phoenix.Socket.Broadcast{event: "ready", payload: %{replication_conn: replication_conn}}
+                     when is_pid(replication_conn),
+                     5000
+
+      assert {:ok, ^replication_conn} = Connect.replication_status(tenant.external_id)
+    end
+
+    test "every joining client receives the replication established system message, even after streaming has started",
+         %{serializer: serializer} do
+      tenant = Containers.checkout_tenant(run_migrations: true)
+
+      Phoenix.PubSub.subscribe(Realtime.PubSub, Connect.syn_topic(tenant.external_id))
+      {:ok, _db_conn} = Connect.lookup_or_start_connection(tenant.external_id)
+
+      assert_receive %Phoenix.Socket.Broadcast{event: "ready", payload: %{replication_conn: replication_conn}}
+                     when is_pid(replication_conn),
+                     5000
+
+      {socket1, _} = get_connection(tenant, serializer, role: "authenticated")
+      {socket2, _} = get_connection(tenant, serializer, role: "authenticated")
+
+      config = %{broadcast: %{self: true}, private: false}
+      topic1 = "realtime:#{random_string()}"
+      topic2 = "realtime:#{random_string()}"
+
+      WebsocketClient.join(socket1, topic1, %{config: config})
+      WebsocketClient.join(socket2, topic2, %{config: config})
+
+      assert_receive %Message{event: "phx_reply", topic: ^topic1, payload: %{"status" => "ok"}}, 500
+      assert_receive %Message{event: "phx_reply", topic: ^topic2, payload: %{"status" => "ok"}}, 500
+
+      assert_receive %Message{
+                       event: "system",
+                       topic: ^topic1,
+                       payload: %{"message" => "Replication connection established"}
+                     },
+                     2000
+
+      assert_receive %Message{
+                       event: "system",
+                       topic: ^topic2,
+                       payload: %{"message" => "Replication connection established"}
+                     },
+                     2000
+    end
+  end
+
   describe "socket disconnect - tenant suspension" do
     setup [:rls_context]
 
@@ -191,6 +246,14 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
           WebsocketClient.join(socket, topic, %{config: config})
 
           assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 500
+
+          assert_receive %Message{
+                           event: "system",
+                           topic: ^topic,
+                           payload: %{"message" => "Replication connection established"}
+                         },
+                         2000
+
           topic
         end
 

test/integration/rt_channel/connection_lifecycle_test.exs

@@ -193,6 +193,13 @@ defmodule Realtime.Integration.RtChannel.TokenHandlingTest do
 
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
 
+      assert_receive %Message{
+                       event: "system",
+                       topic: ^realtime_topic,
+                       payload: %{"message" => "Replication connection established"}
+                     },
+                     2000
+
       WebsocketClient.send_event(socket, realtime_topic, "access_token", %{"access_token" => new_token})
 
       refute_receive %Message{}
@@ -312,6 +319,13 @@ defmodule Realtime.Integration.RtChannel.TokenHandlingTest do
 
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}}, 500
 
+      assert_receive %Message{
+                       event: "system",
+                       topic: ^realtime_topic,
+                       payload: %{"message" => "Replication connection established"}
+                     },
+                     2000
+
       WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
         "access_token" => "sb_publishable_-fake_key"
       })