change the approach to make opt in

Author: filipecabaco

ERROR_CODES.md

@@ -15,6 +15,7 @@ defmodule RealtimeWeb.Channels.Payloads.Config do
     embeds_one :presence, Presence
     embeds_many :postgres_changes, PostgresChange
     field :private, FlexibleBoolean, default: false
+    field :replication_ready, FlexibleBoolean, default: false
   end
 
   def changeset(config, attrs) do
@@ -28,7 +29,7 @@ defmodule RealtimeWeb.Channels.Payloads.Config do
       |> Map.new()
 
     config
-    |> cast(attrs, [:private], message: &Join.error_message/2)
+    |> cast(attrs, [:private, :replication_ready], message: &Join.error_message/2)
     |> cast_embed(:broadcast, invalid_message: "unable to parse, expected a map")
     |> cast_embed(:presence, invalid_message: "unable to parse, expected a map")
     |> cast_embed(:postgres_changes, invalid_message: "unable to parse, expected an array of maps")

lib/realtime_web/channels/payloads/config.ex

@@ -116,7 +116,8 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       RealtimeWeb.Endpoint.subscribe(tenant_topic, metadata: metadata)
       RealtimeWeb.Endpoint.subscribe("realtime:operations:" <> tenant_id, metadata: metadata)
-      send(self(), :notify_replication_ready)
+
+      replication_ready_opt_in? = !!params["config"]["replication_ready"]
 
       is_new_api = new_api?(params)
       presence_enabled? = socket.assigns.presence_enabled?
@@ -146,11 +147,21 @@ defmodule RealtimeWeb.RealtimeChannel do
         self_broadcast: Join.self_broadcast?(join),
         tenant_topic: tenant_topic,
         channel_name: sub_topic,
-        presence_enabled?: presence_enabled?,
-        replication_ready_notified?: false,
-        replication_ready_deadline: System.monotonic_time(:millisecond) + replication_ready_timeout()
+        presence_enabled?: presence_enabled?
       }
 
+      assigns =
+        if replication_ready_opt_in? do
+          Process.send_after(self(), :notify_replication_ready, @replication_ready_check_interval)
+
+          Map.merge(assigns, %{
+            replication_ready_notified?: false,
+            replication_ready_deadline: System.monotonic_time(:millisecond) + replication_ready_timeout()
+          })
+        else
+          assigns
+        end
+
       socket =
         socket
         |> assign_counter(tenant)
@@ -314,7 +325,7 @@ defmodule RealtimeWeb.RealtimeChannel do
 
     cond do
       match?({:ok, _replication_conn}, Connect.replication_status(tenant_id)) ->
-        push_system_message("broadcast", socket, "ok", "Replication connection established", channel_name)
+        push_system_message("system", socket, "ok", "Replication connection established", channel_name)
         {:noreply, assign(socket, :replication_ready_notified?, true)}
 
       System.monotonic_time(:millisecond) >= deadline ->

lib/realtime_web/channels/realtime_channel.ex

@@ -274,24 +274,16 @@ 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();
-  const subscribeMs = await waitForSubscribed(channel);
-  await waitFor(() => replicationReady ? true : null, "replication established");
-  return subscribeMs;
+  return waitForSubscribed(channel);
 }
 
 // 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, extension }: { status: string; extension?: string }) => {
-    if (status === "ok" && extension === "postgres_changes") systemOk = true;
-  });
+  channel.on("system", "*", ({ status }: { status: string }) => { if (status === "ok") systemOk = true; });
   const subscribeMs = await openChannel(channel);
   const { latencyMs: systemMs } = await waitFor(() => systemOk ? true : null, "system ok");
   return { subscribeMs, systemMs: performance.now() - start };
@@ -1009,6 +1001,7 @@ 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");
 
@@ -1038,6 +1031,7 @@ 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");
@@ -1068,6 +1062,7 @@ 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");
@@ -1776,6 +1771,7 @@ 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

@@ -76,13 +76,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
 
       # Join events
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
-
-      assert_receive %Message{
-                       topic: ^topic,
-                       event: "system",
-                       payload: %{"extension" => "postgres_changes", "status" => "ok"}
-                     },
-                     50000
+      assert_receive %Message{topic: ^topic, event: "system"}, 5000
 
       # Wait for RateCounter to run
       RateCounterHelper.tick_tenant_rate_counters!(tenant.external_id)
@@ -116,7 +110,6 @@ 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"}
 
@@ -127,7 +120,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
         assert_receive %Message{topic: ^topic, event: "broadcast", payload: ^payload}
       end
 
-      refute_receive %Message{topic: ^topic, event: "broadcast"}
+      refute_receive _any
 
       # Wait for RateCounter to run
       RateCounterHelper.tick_tenant_rate_counters!(tenant.external_id)
@@ -207,25 +200,13 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
 
       # Join events
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
-
-      assert_receive %Message{
-                       topic: ^topic,
-                       event: "system",
-                       payload: %{"extension" => "postgres_changes", "status" => "ok"}
-                     },
-                     5000
+      assert_receive %Message{topic: ^topic, event: "system"}, 5000
 
       # Add second user to test the "multiplication" of billable events
       {socket, _} = get_connection(tenant, serializer)
       WebsocketClient.join(socket, topic, %{config: config})
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
-
-      assert_receive %Message{
-                       topic: ^topic,
-                       event: "system",
-                       payload: %{"extension" => "postgres_changes", "status" => "ok"}
-                     },
-                     5000
+      assert_receive %Message{topic: ^topic, event: "system"}, 5000
 
       tenant = Tenants.get_tenant_by_external_id(tenant.external_id)
       {:ok, conn} = Database.connect(tenant, "realtime_test", :stop)
@@ -239,7 +220,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
                          event: "postgres_changes",
                          payload: %{"data" => %{"schema" => "public", "table" => "test", "type" => "INSERT"}}
                        },
-                       500
+                       5000
       end
 
       # Wait for RateCounter to run
@@ -267,13 +248,7 @@ defmodule Realtime.Integration.RtChannel.BillableEventsTest do
 
       # Join events
       assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 300
-
-      assert_receive %Message{
-                       topic: ^topic,
-                       event: "system",
-                       payload: %{"extension" => "postgres_changes", "status" => "error"}
-                     },
-                     5000
+      assert_receive %Message{topic: ^topic, event: "system"}, 5000
 
       # Wait for RateCounter to run
       RateCounterHelper.tick_tenant_rate_counters!(tenant.external_id)

test/integration/rt_channel/billable_events_test.exs

@@ -68,7 +68,7 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
       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",
+    test "clients that opt in receive the replication established system message, even after streaming has started",
          %{serializer: serializer} do
       tenant = Containers.checkout_tenant(run_migrations: true)
 
@@ -82,7 +82,7 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
       {socket1, _} = get_connection(tenant, serializer, role: "authenticated")
       {socket2, _} = get_connection(tenant, serializer, role: "authenticated")
 
-      config = %{broadcast: %{self: true}, private: false}
+      config = %{broadcast: %{self: true}, private: false, replication_ready: true}
       topic1 = "realtime:#{random_string()}"
       topic2 = "realtime:#{random_string()}"
 
@@ -246,14 +246,6 @@ 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,13 +193,6 @@ 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{}
@@ -319,13 +312,6 @@ 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"
       })

test/integration/rt_channel/token_handling_test.exs

@@ -60,7 +60,7 @@ defmodule RealtimeWeb.RealtimeChannelReplicationReadyTest do
     refute_receive %Socket.Message{event: "system", payload: %{message: "Replication connection established"}}, 300
   end
 
-  test "stops the channel when replication is not established before the timeout", %{tenant: tenant} do
+  test "shuts down the channel when replication is not established before the timeout", %{tenant: tenant} do
     previous = Application.get_env(:realtime, :replication_ready_timeout)
     Application.put_env(:realtime, :replication_ready_timeout, 50)
     on_exit(fn -> Application.put_env(:realtime, :replication_ready_timeout, previous) end)
@@ -80,10 +80,21 @@ defmodule RealtimeWeb.RealtimeChannelReplicationReadyTest do
     assert_receive {:DOWN, ^ref, :process, _, _}, 500
   end
 
+  test "does not arm replication readiness notifications unless opted in", %{tenant: tenant} do
+    stub(Connect, :lookup_or_start_connection, fn _ -> {:ok, self()} end)
+    stub(Connect, :replication_status, fn _ -> {:ok, self()} end)
+
+    jwt = generate_jwt_token(tenant)
+    {:ok, socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+    assert {:ok, _, _} = subscribe_and_join(socket, "realtime:test", %{"config" => %{}})
+
+    refute_receive %Socket.Message{event: "system", payload: %{message: "Replication connection established"}}, 300
+  end
+
   defp join(tenant) do
     jwt = generate_jwt_token(tenant)
     {:ok, socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
-    subscribe_and_join(socket, "realtime:test", %{"config" => %{}})
+    subscribe_and_join(socket, "realtime:test", %{"config" => %{"replication_ready" => true}})
   end
 
   defp conn_opts(tenant, token) do

test/realtime_web/channels/realtime_channel_replication_ready_test.exs

@@ -105,12 +105,6 @@ defmodule RealtimeWeb.RealtimeChannelTest do
                "commit_timestamp" => _
              } = Jason.encode!(data) |> Jason.decode!()
 
-      assert_receive %Socket.Message{
-                       event: "system",
-                       payload: %{message: "Replication connection established"}
-                     },
-                     500
-
       refute_receive %Socket.Message{}
       refute_receive %Socket.Reply{}
     end
@@ -185,12 +179,6 @@ defmodule RealtimeWeb.RealtimeChannelTest do
                "commit_timestamp" => _
              } = Jason.encode!(data) |> Jason.decode!()
 
-      assert_receive %Socket.Message{
-                       event: "system",
-                       payload: %{message: "Replication connection established"}
-                     },
-                     500
-
       refute_receive _any
     end
 
@@ -515,12 +503,6 @@ defmodule RealtimeWeb.RealtimeChannelTest do
         }
       }
 
-      assert_receive %Socket.Message{
-                       event: "system",
-                       payload: %{message: "Replication connection established"}
-                     },
-                     500
-
       refute_receive %Socket.Message{}
     end
   end
@@ -705,12 +687,6 @@ defmodule RealtimeWeb.RealtimeChannelTest do
       assert_receive %Socket.Reply{payload: %{}, topic: "realtime:test", status: :ok}, 500
       # no presence_diff this time
 
-      assert_receive %Socket.Message{
-                       event: "system",
-                       payload: %{message: "Replication connection established"}
-                     },
-                     500
-
       refute_receive %Socket.Message{}
       refute_receive %Socket.Reply{}
     end