feat: broadcast persistence by leandrocp · Pull Request #2022 · supabase/realtime

leandrocp · 2026-07-14T20:19:57Z

What kind of change does this PR introduce?

Introduce Broadcast Persistence allowing users to opt-in to :persistence to store broadcasted messages that can be replayed or manipulated in realtime.messages.

Closes REAL-956

Notes

Always try to broadcast first
Ack true await persistence and return message id on response
Ack false do not wait persistence, just fire and forget
Feature flag broadcast_persistence

Ref REAL-956

Testing

1. Enable feature flag

{:ok, flag} = Api.upsert_feature_flag(%{name: "broadcast_persistence", enabled: true})
FeatureFlags.Cache.update_cache(flag)

2. Create policies

-- policies
create policy demo_read on realtime.messages for select to authenticated
  using (realtime.topic() like 'persisted:%');

create policy demo_send on realtime.messages for insert to authenticated
  with check (realtime.messages.extension = 'broadcast' and realtime.topic() like 'persisted:%');

create policy demo_persist on realtime.messages for insert to authenticated
  with check (realtime.messages.extension = 'persistence' and realtime.topic() like 'persisted:%');

3. Set token

claims = %{"role" => "authenticated", "sub" => "demo", "exp" => System.system_time(:second) + 100000000000}
token = Joken.generate_and_sign!(%{}, claims, Joken.Signer.create("HS256", "dev"))

4. Broadcast from WS

# broadcast
{"topic":"realtime:persisted:demo","event":"phx_join","payload":{"config":{"private":true,"broadcast":{"self":true,"ack":true}}},"ref":"1"}
{"topic":"realtime:persisted:demo","event":"broadcast","payload":{"type":"broadcast","event":"hello","payload":{"msg":1}},"ref":"2"}

5. Broadcast from API

TOKEN="..."

curl -i -X POST "http://realtime-dev.localhost:4000/api/broadcast/persisted:demo/events/hello?private=true" \
  -H "authorization: Bearer $TOKEN" \
  -H "content-type: application/json" \
  -d '{"msg":2}'

6. Broadcast from DB

set time zone 'UTC';
select realtime.send('{"msg":3}'::jsonb, 'hello', 'persisted:demo', true);

7. Replay all messages

# replay
{"topic":"realtime:persisted:demo","event":"phx_join","payload":{"config":{"private":true,"broadcast":{"replay":{"since":0,"limit":100}}}},"ref":"1"}

github-actions · 2026-07-14T20:50:42Z

CRAP Score Report

Summary: files=234 functions=1285 scored=1285 worst_score=15944.76

** (Mix) CRAP threshold failed: max_score=30.00
High scores: 30
  lib/extensions/postgres_cdc_rls/subscription_manager.ex Extensions.PostgresCdcRls.SubscriptionManager.handle_info/2 score=37.16
  lib/realtime/adapters/postgres/oid_database.ex Realtime.Adapters.Postgres.OidDatabase.name_for_type_id/1 score=15944.76
  lib/realtime/application.ex Realtime.Application.setup_region_mapping/0 score=47.11
  lib/realtime/nodes.ex Realtime.Nodes.default_region_mapping/1 score=157.20
  lib/realtime/operations.ex Realtime.Operations.rebalance/0 score=42.00
  lib/realtime/operations.ex Realtime.Operations.kill_connections_to_tenant_id/2 score=90.00
  lib/realtime/tenants/connect.ex Realtime.Tenants.Connect.handle_info/2 score=40.25
  lib/realtime_web/channels/realtime_channel.ex RealtimeWeb.RealtimeChannel.handle_info/2 score=60.51
  lib/realtime_web/channels/realtime_channel.ex RealtimeWeb.RealtimeChannel.handle_in/3 score=132.36
  lib/realtime_web/dashboard/feature_flags.ex RealtimeWeb.Dashboard.FeatureFlags.handle_event/3 score=112.70
  lib/realtime_web/dashboard/node_info.ex RealtimeWeb.Dashboard.NodeInfo.fetch_node_data/2 score=42.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.handle_event/3 score=552.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.handle_info/2 score=210.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.render_value/2 score=42.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.load_module_functions/1 score=42.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.parse_and_start/2 score=110.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.format_value/1 score=702.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.sort_entries/2 score=42.00
  lib/realtime_web/dashboard/recon_trace.ex RealtimeWeb.Dashboard.ReconTrace.parse_module/1 score=90.00
  lib/realtime_web/dashboard/sql_inspector.ex RealtimeWeb.Dashboard.SqlInspector.handle_event/3 score=72.00
  lib/realtime_web/dashboard/sql_inspector.ex RealtimeWeb.Dashboard.SqlInspector.execute_read_only/1 score=72.00
  lib/realtime_web/dashboard/sql_inspector.ex RealtimeWeb.Dashboard.SqlInspector.mask_sensitive_columns/1 score=42.00
  lib/realtime_web/dashboard/sql_inspector.ex RealtimeWeb.Dashboard.SqlInspector.compare_cells/2 score=42.00
  lib/realtime_web/dashboard/sql_inspector.ex RealtimeWeb.Dashboard.SqlInspector.format_cell/1 score=72.00
  lib/realtime_web/dashboard/tenant_migrations.ex RealtimeWeb.Dashboard.TenantMigrations.handle_info/2 score=156.00
  lib/realtime_web/live/components.ex RealtimeWeb.Components.input/1 score=90.00
  lib/realtime_web/live/inspector_live/conn_component.ex RealtimeWeb.InspectorLive.ConnComponent.handle_event/3 score=53.83
  lib/realtime_web/live/inspector_live/event_log_component.ex RealtimeWeb.InspectorLive.EventLogComponent.category_variant/1 score=35.00
  lib/realtime_web/live/inspector_live/event_log_component.ex RealtimeWeb.InspectorLive.EventLogComponent.event_label/1 score=76.13
  lib/realtime_web/live/status_live/index.ex RealtimeWeb.StatusLive.Index.handle_event/3 score=42.00

coveralls · 2026-07-14T20:52:08Z

Coverage is 90.841% — feat-broadcast-storage into main. No base build found for main.

leandrocp · 2026-08-06T20:42:58Z

    end
  end

-  defp result_to_single_struct(


- check `:persistence` and store as `:broadcast` - add logger meta `tenant_id` - simplify `with` to have smaller diff - json passthrough using Jason.Fragment - fix docs

edgurgel · 2026-08-18T02:16:39Z

-    with {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id) do
-      auth_params = %{auth_params | topic: topic}
-      Authorization.get_write_authorizations(db_conn, auth_params)
+    with {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id),


should we just move this Connect clause out so you don't need to return db_conn here and keep this function just checking for the write authorizations?

I'm not sure if you meant out of the with clauses or out of handle_private_message but on main handle_private_message already calls Connect.lookup_or_start_connection/1 indirectly and now it just moved to to be more explicit and reused by permissions_for_message and maybe_persist so I'm not sure there's something else we can change here. 🤔

I've added a comment above to be more clear

edgurgel · 2026-08-18T02:17:28Z

+          map() | binary(),
          content_type()
        ) :: :ok | {:error, term()} | {:error, atom(), String.t()}
  def broadcast(_auth_params, %Tenant{suspend: true}, _topic, _event, _private, _payload, _content_type) do


So if I understand well this means that every API request will try to persist?

I think this should be an option you pass from the API request. And if that's the case let's do this just on the new API which uses SingleBroadcast and leave BatchBroadcast alone?

Good point!

I think this should be an option you pass from the API request

Added a query param persist (similar to private) to enable persistence.

leandrocp · 2026-08-18T16:39:57Z

-      Authorization.get_write_authorizations(db_conn, auth_params)
+  defp maybe_persist(%BroadcastPolicies{persist: true}, db_conn, tenant, topic, event, payload) do
+    if FeatureFlags.broadcast_persistence_enabled?(tenant.external_id) do
+      Task.Supervisor.start_child(Realtime.TaskSupervisor, fn ->


Same behavior as ack=false.

It might useful to add a ack query param but not on this PR.

Hmm yeah we might want to consider something like persist=ack or something like this? As ack here is only related to the case when something is persisted?

What if we add the ack param (defaults to false) into the API as well?

As ack here is only related to the case when something is persisted?

Currently there's no ack right? So the only possible logic is to try to persist in a best-effort case, but having ack we could wait and respond with message id. Not sure is there's much value tho, need to discuss.

It really depends because we are mixing things here I think?

API is always ACK (single broadcast at least) as it will return error if broadcasting failed for example if the payload size is above the limit. See the status codes the single broadcast API can return

On the websocket if you don't want ack it will not reply if it exceeded a payload size.

Now persisting is sort of another layer on top of this not the only thing

WDYT?

leandrocp · 2026-08-18T16:40:24Z

        example: false,
        description: "Whether this is a private broadcast (requires RLS authorization). Defaults to false."
      ]
+      # TODO: uncomment when broadcast persistence is public and ready


To not expose it yet.

edgurgel · 2026-08-19T02:13:09Z

@@ -152,11 +155,12 @@ defmodule Realtime.Tenants.SingleBroadcast do

  defp handle_private_message(tenant, auth_params, topic, event, payload, content_type, rate_counter) do
    case permissions_for_message(tenant, auth_params, topic) do


Calling {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id) here, and then you don't need to get db_conn from "permissions_for_message" now that the connection is being used for two different things it makes less sense to have Connect being called from permissions_for_message

edgurgel · 2026-08-19T04:31:08Z

+  defp convert_to_persistable_fields(%{"event" => event, "payload" => payload}), do: {:ok, event, payload}
+
+  defp convert_to_persistable_fields({event, :json, user_payload, _metadata}),
+    do: {:ok, event, Jason.Fragment.new(user_payload)}


Oh so Jason.Fragment will be properly stored as JSON without decoding/re-encoding by Ecto?

Yes although it's a bit tricky to test that exact path, but still tested indirectly by all other tests.

edgurgel · 2026-08-19T04:37:16Z

+      :ok
+    end
+
+    test "public broadcast is delivered but not stored", %{


Should we make one of these tests send a binary payload?

Good point, added it!

edgurgel · 2026-08-19T04:38:02Z

      |> expect(:get_write_authorizations, 2, fn
-        _, %{topic: ^topic} -> %Policies{broadcast: %BroadcastPolicies{write: true}}
-        _, _ -> %Policies{broadcast: %BroadcastPolicies{write: false}}
+        _, %{topic: ^topic} -> {:ok, %Policies{broadcast: %BroadcastPolicies{write: true}}}


huh how was this test not breaking?

Because the code is:

case Authorization.get_write_authorizations(db_conn, auth_params) do {:ok, policies} -> policies {:error, :not_found} -> nil error -> error end

So it just falls into the error clause returning the struct which is the exact same return when it pass {:ok, policies} (the first clause).

Since Authorization.get_write_authorizations/2 returns only {:ok, ...} | {:error, ...} (per spec) we should probably remove error -> error. Wdyt?

edgurgel · 2026-08-19T04:39:49Z

    end
  end

+  describe "message persistence" do


There are no more changes on batch_broadcast 🤔 How are these tests passing?

Because the only test now is test "the batch API never stores messages, even when authorized to persist" to make it explicit that batch broadcast doesn't persist messages.

ooooh sorry I should've read it...

edgurgel · 2026-08-19T04:42:24Z

+      assert :ok = SingleBroadcast.broadcast(auth_params, tenant, topic, "event", true, %{"a" => "b"}, :json, true)
+
+      assert {:ok, []} = Repo.all(db_conn, messages_for(topic), Message)


This test could be passing if the task supervisor has not inserted right? That's why eventually is used above, correct?

I wonder if we should check that there are no children under the task supervisor? 🤔

Or us a refute eventually with retries like down below?

For now I'm avoiding adding too much code around tasks because that will be most likely temporary, we need something better to handle hard limits and pressure which is to be done in a following PR so I think it's okay for this version.

edgurgel · 2026-08-19T04:43:28Z

+          boolean()
        ) :: :ok | {:error, term()} | {:error, atom(), String.t()}
-  def broadcast(_auth_params, %Tenant{suspend: true}, _topic, _event, _private, _payload, _content_type) do
+  def broadcast(auth_params, tenant, topic, event, private, payload, content_type, persist \\ false)


I wonder if we should use opts here for private and persist? It gets harder to read boolean arguments like this. WDYT?

Yep I prefer opts as well, changed it. 👍🏻

edgurgel · 2026-08-19T05:01:02Z

+
+      assert_receive {:socket_push, _encoding, _data}
+      assert log =~ "UnableToPersistMessage"
+      assert {:ok, []} = Repo.all(db_conn, messages_for(topic), Message)


given the task stuff is it enough to look once for messages?

socket_fixture in this module does ack_broadcast = true by default so the broadcast handler wait for that insert, but I also added ack_broadcast: true in the test to make it explicit.

edgurgel · 2026-08-19T05:03:03Z

+      db_conn: db_conn,
+      tenant: tenant
+    } do
+      stub(GenCounter, :add, fn _ -> :ok end)


Why the stub on gencounter? 🤔

Because it shares the same setup and tenant with others tests that assert rate_counter.avg causing flaky tests.

I see... We might want fix these flaky tests then. If you see them again let's see if we can fix it!

- pass opts with :private and :persist - fix msg "private channels" - test store and replay persisted binary messages - enable persistence per tenant in tests to simplify and make tests more reliable - pass an explicit ack:true to socket_fixture in tests to make it explicit

edgurgel

I've added some comments but other than those LGTM

Adapt the persistence probe to the lazy per-extension write authorization introduced in #2089: :persistence is now a valid extension requested on its own instead of an entry appended to the checked list.

realtime-release-bot · 2026-08-31T14:49:47Z

🎉 This PR is included in version 2.132.0 🎉

The release is available on GitHub release

Your semantic-release bot 📦🚀

leandrocp and others added 4 commits July 14, 2026 16:19

feat: broadcast storage

78e0590

Merge remote-tracking branch 'origin/main' into feat-broadcast-storage

74ae972

chore: update tenant db catalog snapshots

05f2f4f

chore: update tenant db catalog snapshots

9e20562

leandrocp added 2 commits July 21, 2026 16:00

wip: benchmark possible solutions

5f7d35e

del benchs

7b3bbb2

leandrocp force-pushed the feat-broadcast-storage branch 2 times, most recently from 6401d83 to 4927967 Compare July 30, 2026 15:15

leandrocp marked this pull request as draft July 30, 2026 15:19

leandrocp force-pushed the feat-broadcast-storage branch 2 times, most recently from 10f9d60 to ea15322 Compare July 31, 2026 16:15

RLS-based persistence

52c9004

leandrocp force-pushed the feat-broadcast-storage branch from ea15322 to 52c9004 Compare July 31, 2026 16:25

leandrocp and others added 5 commits July 31, 2026 13:30

Merge remote-tracking branch 'origin/main' into feat-broadcast-storage

925ad5b

format

6ab20a0

chore: update tenant db catalog and dumps

227585e

chore: update tenant db catalog and dumps

0994fc4

fix spec

6e910fb

leandrocp changed the title ~~feat: broadcast storage~~ feat: broadcast persistence Jul 31, 2026

leandrocp marked this pull request as ready for review July 31, 2026 19:19

edgurgel reviewed Aug 4, 2026

View reviewed changes

leandrocp force-pushed the feat-broadcast-storage branch from 32e6fb2 to 859d340 Compare August 6, 2026 20:42

leandrocp commented Aug 6, 2026

View reviewed changes

Comment thread lib/realtime/tenants/repo.ex

end

end

defp result_to_single_struct(

leandrocp Aug 6, 2026

Copy link
Copy Markdown

Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dead code.

address review:

b2cd728

- check `:persistence` and store as `:broadcast` - add logger meta `tenant_id` - simplify `with` to have smaller diff - json passthrough using Jason.Fragment - fix docs

leandrocp force-pushed the feat-broadcast-storage branch from 859d340 to b2cd728 Compare August 6, 2026 21:17

api

1b069d4

leandrocp force-pushed the feat-broadcast-storage branch from d782018 to 1b069d4 Compare August 6, 2026 21:44

chore: update tenant db catalog and dumps

958d0f8

fix tests

3f94e7a

leandrocp force-pushed the feat-broadcast-storage branch from 7844cde to 3f94e7a Compare August 17, 2026 20:43

edgurgel reviewed Aug 18, 2026

View reviewed changes

leandrocp added 4 commits August 18, 2026 11:33

replace PersistencePolicies with :persist opt in BroadcastPolicies

6f98d4a

api persist param

59eae34

Merge remote-tracking branch 'origin/main' into feat-broadcast-storage

6bbc64a

cleanup

664d408

leandrocp force-pushed the feat-broadcast-storage branch from 4cac34a to 664d408 Compare August 18, 2026 16:18

revert batch

240a0b3

leandrocp commented Aug 18, 2026

View reviewed changes

edgurgel reviewed Aug 19, 2026

View reviewed changes

Comment thread lib/realtime/tenants/single_broadcast.ex Outdated

edgurgel reviewed Aug 19, 2026

View reviewed changes

Comment thread test/realtime_web/channels/realtime_channel/broadcast_handler_test.exs Outdated

edgurgel reviewed Aug 19, 2026

View reviewed changes

edgurgel approved these changes Aug 26, 2026

View reviewed changes

leandrocp added 2 commits August 28, 2026 12:56

Merge remote-tracking branch 'origin/main' into feat-broadcast-storage

ae15468

Adapt the persistence probe to the lazy per-extension write authorization introduced in #2089: :persistence is now a valid extension requested on its own instead of an entry appended to the checked list.

cleanup

19f0537

leandrocp merged commit af99cde into main Aug 31, 2026
64 of 65 checks passed

leandrocp deleted the feat-broadcast-storage branch August 31, 2026 14:49

realtime-release-bot Bot added the released label Aug 31, 2026

		@@ -152,11 +155,12 @@ defmodule Realtime.Tenants.SingleBroadcast do

		defp handle_private_message(tenant, auth_params, topic, event, payload, content_type, rate_counter) do
		case permissions_for_message(tenant, auth_params, topic) do

		assert :ok = SingleBroadcast.broadcast(auth_params, tenant, topic, "event", true, %{"a" => "b"}, :json, true)

		assert {:ok, []} = Repo.all(db_conn, messages_for(topic), Message)

Uh oh!

Conversation

leandrocp commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What kind of change does this PR introduce?

Notes

Testing

1. Enable feature flag

2. Create policies

3. Set token

4. Broadcast from WS

5. Broadcast from API

6. Broadcast from DB

7. Replay all messages

Uh oh!

github-actions Bot commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CRAP Score Report

Uh oh!

coveralls commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edgurgel Aug 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leandrocp Aug 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edgurgel Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leandrocp Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

leandrocp commented Jul 14, 2026 •

edited

Loading

github-actions Bot commented Jul 14, 2026 •

edited

Loading

coveralls commented Jul 14, 2026 •

edited

Loading

edgurgel Aug 24, 2026 •

edited

Loading

leandrocp Aug 18, 2026 •

edited

Loading

edgurgel Aug 19, 2026 •

edited

Loading

leandrocp Aug 19, 2026 •

edited

Loading

edgurgel Aug 19, 2026 •

edited

Loading