fix: message dispatcher cache (#1615)

Author: edgurgel

lib/realtime_web/channels/realtime_channel/message_dispatcher.ex

@@ -107,6 +107,7 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
 
             {:error, reason} ->
               maybe_log(log_level, join_topic, reason, tenant_id)
+              {:error, reason}
           end
 
         Map.put(cache, serializer, result)

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.63.1",
+      version: "2.63.2",
       elixir: "~> 1.18",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/e2e/tests.ts

@@ -66,11 +66,7 @@ describe("broadcast extension", () => {
     while (activeChannel.state == "joining") await sleep(0.2);
 
     // Send from unsubscribed channel
-    supabase.channel(topic, config).send({
-      type: "broadcast",
-      event,
-      payload: expectedPayload,
-    });
+    supabase.channel(topic, config).httpSend(event, expectedPayload);
 
     while (result == null) await sleep(0.2);
 

test/realtime/gen_rpc_pub_sub_test.exs

@@ -23,6 +23,8 @@ defmodule Realtime.GenRpcPubSubTest do
                 def subscribe(subscriber, topic) do
                   spawn(fn ->
                     RealtimeWeb.Endpoint.subscribe(topic)
+                    2 = length(Realtime.Nodes.region_nodes("us-east-1"))
+                    2 = length(Realtime.Nodes.region_nodes("ap-southeast-2"))
                     send(subscriber, {:ready, Application.get_env(:realtime, :region)})
 
                     loop = fn f ->
@@ -91,12 +93,12 @@ defmodule Realtime.GenRpcPubSubTest do
 
         {:ok, _} = Clustered.start(@aux_mod, name: ap2_nodeY, extra_config: ap2_nodeY_extra_config, phoenix_port: 4016)
 
-        RealtimeWeb.Endpoint.subscribe(@topic)
-        :erpc.multicall(Node.list(), Subscriber, :subscribe, [self(), @topic])
-
         # Ensuring that syn had enough time to propagate to all nodes the group information
         Process.sleep(3000)
 
+        RealtimeWeb.Endpoint.subscribe(@topic)
+        :erpc.multicall(Node.list(), Subscriber, :subscribe, [self(), @topic])
+
         assert length(Realtime.Nodes.region_nodes("us-east-1")) == 2
         assert length(Realtime.Nodes.region_nodes("ap-southeast-2")) == 2
 

test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs

@@ -430,6 +430,9 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
       assert log =~ "Received message on realtime:topic with payload: #{inspect(msg, pretty: true)}"
       assert log =~ "User payload encoding is not JSON"
 
+      # Only prints once
+      assert String.split(log, "User payload encoding is not JSON") |> length() == 2
+
       # No V1 message received as binary payloads are not supported
       refute_receive {:socket_push, :text, _message_v1}