msimonborg

msimonborg

Hello,

I have been using the Postgres NOTIFY feature via Postgrex.Notifications to listen for messages emitted after insert of a table row. I’m using this to emit the event notification, rather than Phoenix.PubSub or another strategy, because for my use case I’m specifically interested in knowing when the data is available to read, and not when the transaction was made. I’m distributing the application in a global cluster on fly.io with local read replicas in each VM region, and the intent is to know when the local replica has the data.

I have this working exactly as intended when I only have one writable db instance that all nodes are connected to. Recently I added @brainlid 's excellent fly_postgres library to my project to setup local replica connections. The library can detect if a node is in the “primary” region or not, and will dynamically change the database connection and config to connect to the nearest replica if it’s outside of the primary. It also wraps the Repo functions to automatically make RPC calls to nodes in the primary for write operations.

After following the configuration steps everything is working great except that I am no longer receiving the postgres NOTIFY messages on the nodes outside of the primary region when deploying the read replicas. Nodes in the primary region still receive messages properly. I found some information on stack overflow about extra psql you must execute in order to enable triggers on the replicas, but after running the migration I still have no luck.

Does anyone know if I might be doing something wrong? Does Postgrex.Notifications work with read replicas? Is there some quirk of this configuration, or maybe the fly.io database infrastructure that could be tripping this up?

Here’s the relevant code/config. Note that fly_postgres has you rename your Repo to Repo.Local, and the standard Repo module wraps the functions in Repo.Local, changing the behavior based on the region, so if you see naming inconsistencies those are not mistakes :slightly_smiling_face: Thanks in advance!

# config/config.exs
config :my_app,
  ecto_repos: [MyApp.Repo.Local]

config :my_app, MyApp.Repo.Local,
  priv: "priv/repo"

# config/runtime.exs
database_url =
  System.get_env("DATABASE_URL") ||
    raise """
    environment variable DATABASE_URL is missing.
    For example: ecto://USER:PASS@HOST/DATABASE
    """

maybe_ipv6 = if System.get_env("ECTO_IPV6"), do: [:inet6], else: []

config :my_app, MyApp.Repo.Local,
  # ssl: true,
  url: database_url,
  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "20"),
  socket_options: maybe_ipv6

# application.ex
children = [
  Fly.RPC,
  MyApp.Repo.Local,
  {Fly.Postgres.LSN.Tracker, repo: MyApp.Repo.Local},
  {Postgrex.Notifications, Keyword.put_new(Repo.config(), :name, MyApp.Notifier)},
  MyApp.Listener
]
Supervisor.start_link(...)

# my_app/listener.ex
def init(_) do
  Process.monitor(MyApp.Notifier)
  {_, ref} = Postgrex.Notifications.listen(MyApp.Notifier, "handoff_inserted")  
  {:ok, %__MODULE__{ref: ref}}
end

def handle_info({:notification, _, ref, "handoff_inserted", payload}, %{ref: ref} = state) do
  do_stuff(payload)
  {:noreply, state}
end

# migrations
defmodule MyApp.Repo.Migrations.CreateHandoffNotifications do
  use Ecto.Migration

  def up do
    execute("""
    CREATE OR REPLACE FUNCTION notify_handoff_insertions()
    RETURNS trigger AS $$
    BEGIN
      PERFORM pg_notify(
        'handoff_inserted',
        NEW.game_id::text
      );

      RETURN NEW;
    END;
    $$ LANGUAGE plpgsql;
    """)

    execute("""
    CREATE TRIGGER handoff_inserted
    AFTER INSERT
    ON handoffs
    FOR EACH ROW
    EXECUTE PROCEDURE notify_handoff_insertions();
    """)
  end

  def down do
    execute("DROP FUNCTION notify_handoff_insertions() CASCADE;")
  end
end

# added this migration after installing library and renaming to Repo.Local
defmodule MyApp.Repo.Local.Migrations.AlwaysEnableHandoffNotifications do
  use Ecto.Migration

  def up do
    execute("""
    ALTER TABLE handoffs ENABLE ALWAYS TRIGGER handoff_inserted;
    """)
  end
end

Showing Posts 1 to 2

josevalim

josevalim

Creator of Elixir

Could this potentially be related? https://stackoverflow.com/a/51482130

msimonborg

msimonborg OP

I think you are right. I saw that before and tried to follow the discussion, but I did not follow it deep enough. I got to a point where someone had suggested they had a patch for it, and I guess I made an optimistic assumption based on the age of the issue in years and version numbers and the suggestion of a patch that it must be solved and I was just doing something wrong. Looking farther along, apparently the patch was never merged and the thread died off. Someone made this suggestion in the discussion:

If there’s nothing in the short-term, can we get a caveat in the
documentation for notify/logical
replication explaining this shortcoming?

I think maybe the same would be useful for Postgrex.Notifications’s module docs? I know it would have saved me a lot of time going down this road. I’d be happy to make the PR :slightly_smiling_face:

I do have other ideas for how to achieve my goal, this just seemed like a clean solution. If I have something I think is working I’ll post it here so others can find it. Thanks for your help!

— All posts loaded —

Where Next? Top

Trending in Questions Top

RSP87
I’m working on a project that simulates the bumbl example in the programming phoenix book. It acts almost like an email client. We have a...
New
nseaSeb
Hello, I know there is an approach for handling lists that allows for optimized traversal, but I can’t recall the specific method (somet...
New
RemyXRenard
I’m seeing that a list inside a Kino.DataTable will be interpreted as a charlist, even if the Kino.configure() is set to charlists: :as_l...
New
velrest
So my question is quite simple and i have found no conclusive answer on forum, google or AI. Should we use :erlang.float for Integer to ...
New
brecabral
Documentation While reading the Scoped Routes section, I noticed that the documentation currently refers to a problem without explainin...
New
samoloth
Hi, I’ve just set up an application with ash_authentication. There is only magic link strategy for now, so there is no confirmation add o...
New
FlyingNoodle
If a change or preparation module uses Ash.Changeset.get_argument/2 or Ash.Query.get_argument/2 (or any of the other get_argument functio...
New

Other Trending Topics Top

mudasobwa
I am happy to introduce the very α version of the new programming language compiled to BEAM. Welcome Cure. It has literally three kille...
New
marciok
Hi there! We created Gust: A task orchestrator inspired by Airflow. For those who have never heard about Aiflow, it’s a Python-based wor...
New
jimsynz
Beam Bots (or just BB for short) is a framework for building fault-tolerant robotics applications in Elixir using familiar OTP patterns. ...
New
Dmk
Xamal is a deployment tool for Elixir apps that deploys native releases to bare metal servers over SSH. It’s a port of GitHub - basecamp/...
New
netoum
Corex is an accessible, unstyled UI component library for Phoenix that integrates Zag.js state machines using Vanilla JavaScript and Live...
New
webofbits
With AI doing more of the implementation work, I’ve been wondering how much coding I should deliberately keep doing myself. My main conc...
#ai
New

We're in Beta

About us Mission Statement

Options

Thread Display Mode




Thread Preview

Skip Thread Previews