-
-
Notifications
You must be signed in to change notification settings - Fork 317
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: listen on notifications for error handling
* Adds Realtime.Tenants.Listen to receive error messages from functions so we can better inform our users * Changes send function to not handle any partition creation * Changes send function to pg_notify to `realtime:system` so we can capture and handle in Realtime * Increases number of future partitions to 10 days to prevent issues
- Loading branch information
1 parent
821bd77
commit 72ff20a
Showing
11 changed files
with
262 additions
and
71 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| defmodule Realtime.Tenants.Listen do | ||
| @moduledoc """ | ||
| Listen for Postgres notifications to identify issues with the functions that are being called in tenants database | ||
| """ | ||
| use GenServer, restart: :transient | ||
| require Logger | ||
| alias Realtime.Logs | ||
| alias Realtime.Api.Tenant | ||
| alias Realtime.Database | ||
| alias Realtime.PostgresCdc | ||
| alias Realtime.Registry.Unique | ||
|
|
||
| defstruct tenant_id: nil, listen_conn: nil | ||
|
|
||
| @cdc "postgres_cdc_rls" | ||
| @topic "realtime:system" | ||
| def start_link(%Tenant{} = tenant) do | ||
| name = {:via, Registry, {Unique, {__MODULE__, :tenant_id, tenant.external_id}}} | ||
| GenServer.start_link(__MODULE__, tenant, name: name) | ||
| end | ||
|
|
||
| def init(%Tenant{external_id: external_id} = tenant) do | ||
| Logger.metadata(external_id: external_id, project: external_id) | ||
|
|
||
| settings = | ||
| tenant | ||
| |> then(&PostgresCdc.filter_settings(@cdc, &1.extensions)) | ||
| |> then(&Database.from_settings(&1, "realtime_listen", :rand_exp, true)) | ||
| |> Map.from_struct() | ||
|
|
||
| name = | ||
| {:via, Registry, | ||
| {Realtime.Registry.Unique, {Postgrex.Notifications, :tenant_id, tenant.external_id}}} | ||
|
|
||
| settings = | ||
| settings | ||
| |> Map.put(:hostname, settings[:host]) | ||
| |> Map.put(:database, settings[:name]) | ||
| |> Map.put(:password, settings[:pass]) | ||
| |> Map.put(:username, "postgres") | ||
| |> Map.put(:port, String.to_integer(settings[:port])) | ||
| |> Map.put(:ssl, settings[:ssl_enforced]) | ||
| |> Map.put(:auto_reconnect, true) | ||
| |> Map.put(:name, name) | ||
| |> Enum.to_list() | ||
|
|
||
| Logger.info("Listening for notifications on #{@topic}") | ||
|
|
||
| case Postgrex.Notifications.start_link(settings) do | ||
| {:ok, conn} -> | ||
| Postgrex.Notifications.listen!(conn, @topic) | ||
| {:ok, %{tenant_id: tenant.external_id, listen_conn: conn}} | ||
|
|
||
| {:error, {:already_started, conn}} -> | ||
| Postgrex.Notifications.listen!(conn, @topic) | ||
| {:ok, %{tenant_id: tenant.external_id, listen_conn: conn}} | ||
|
|
||
| {:error, reason} -> | ||
| {:stop, reason} | ||
| end | ||
| catch | ||
| e -> {:stop, e} | ||
| end | ||
|
|
||
| @spec start(Realtime.Api.Tenant.t()) :: {:ok, pid()} | {:error, any()} | ||
| def start(%Tenant{} = tenant) do | ||
| supervisor = {:via, PartitionSupervisor, {Realtime.Tenants.Listen.DynamicSupervisor, self()}} | ||
| spec = {__MODULE__, tenant} | ||
|
|
||
| case DynamicSupervisor.start_child(supervisor, spec) do | ||
| {:ok, pid} -> {:ok, pid} | ||
| {:error, {:already_started, pid}} -> {:ok, pid} | ||
| error -> {:error, error} | ||
| end | ||
| catch | ||
| e -> {:error, e} | ||
| end | ||
|
|
||
| def handle_info({:notification, _, _, @topic, payload}, state) do | ||
| case Jason.decode(payload) do | ||
| {:ok, %{"function" => "realtime.send"} = parsed} when is_map_key(parsed, "error") -> | ||
| Logs.log_error("FailedSendFromDatabase", parsed) | ||
|
|
||
| {:error, _} -> | ||
| Logs.log_error("FailedToParseDiagnosticMessage", payload) | ||
|
|
||
| _ -> | ||
| :ok | ||
| end | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| def handle_info(_, state), do: {:noreply, state} | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
34 changes: 34 additions & 0 deletions
34
...po/migrations/20241220123912_realtime_send_handle_exceptions_remove_partition_creation.ex
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| defmodule Realtime.Tenants.Migrations.RealtimeSendHandleExceptionsRemovePartitionCreation do | ||
| @moduledoc false | ||
| use Ecto.Migration | ||
|
|
||
| # We missed the schema prefix of `realtime.` in the create table partition statement | ||
| def change do | ||
| execute(""" | ||
| CREATE OR REPLACE FUNCTION realtime.send(payload jsonb, event text, topic text, private boolean DEFAULT true ) RETURNS void | ||
| AS $$ | ||
| BEGIN | ||
| BEGIN | ||
| -- Attempt to insert the message | ||
| INSERT INTO realtime.messages (payload, event, topic, private, extension) | ||
| VALUES (payload, event, topic, private, 'broadcast'); | ||
| EXCEPTION | ||
| WHEN OTHERS THEN | ||
| -- Capture and notify the error | ||
| PERFORM pg_notify( | ||
| 'realtime:system', | ||
| jsonb_build_object( | ||
| 'error', SQLERRM, | ||
| 'function', 'realtime.send', | ||
| 'event', event, | ||
| 'topic', topic, | ||
| 'private', private | ||
| )::text | ||
| ); | ||
| END; | ||
| END; | ||
| $$ | ||
| LANGUAGE plpgsql; | ||
| """) | ||
| end | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.