Pick up from last wal_position

[supabase]

Pick up from last wal_position

[fracek]

From what I understand, at the moment we are using a temporary replication slot, this means that when the client disconnects (e.g. on crash) the replication slot is dropped.

I then tried to look at the pg_replication_slots table using a named replication slot

postgres=# select * from pg_replication_slots;
 slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn
-----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
 abcd      | pgoutput | logical   |  13117 | postgres | f         | t      |       2222 |      |          644 | 0/1696A58   | 0/1696A58
(1 row)

When I insert a record from the example web application we can see the restart_lsn/confirmed_flush_lsn values increment.

postgres=# select * from pg_replication_slots;
 slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
-----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
 abcd      | pgoutput | logical   |  13117 | postgres | f         | t      |       2233 |      |          648 | 0/1698080   | 0/1698080
(1 row)

I then disconnect the realtime application and manually insert a couple of rows in the users table, the *_lsn values don't change since we are not replicating.

postgres=# select * from pg_replication_slots;
 slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
-----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
 abcd      | pgoutput | logical   |  13117 | postgres | f         | f      |            |      |          648 | 0/1698080   | 0/1698080
(1 row)

When I restart the application (using xlog/offset = {"0", "0'}, i.e. what we have now) we start reading the replication log where we left and correctly push the new records to the channels.

postgres=# select * from pg_replication_slots;
 slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn 
-----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
 abcd      | pgoutput | logical   |  13117 | postgres | f         | t      |       2270 |      |          650 | 0/16980B8   | 0/1699530
(1 row)

So I think the only real change required is to use named replication slots and not a temporary one, then postgres will do the right thing automatically.

[kiwicopple]

I imagine with this one, we are going to need a database of some sort. Ideally it isn't another whole install and can just be done within the server itself. Something like SQLite. I know that elixir also has strong support for mnesia, but that's beyond my knowlege of whether this is a good use case

[kiwicopple]

Following my own advice to post on GH:

We may not even need to store state. The master server keeps track of the subscribers progress for purging. It may even be possible to ask the master PG "give me everything from the last time I acknowleged you"

[kiwicopple]

that's amazing news. If that's the case, then I believe we can just update this line right?

realtime/server/lib/realtime/application.ex

Line 32 in 9d24b60

slot: :temporary, # :temporary is also supported if you don't want Postgres keeping track of what you've acknowledged

Looks like it from the comments. And according to the comments we can also leave it at {0, 0} so that the master can decide what to start sending:

realtime/server/lib/realtime/application.ex

Line 33 in 9d24b60

wal_position: {"0", "0"}, # You can provide a different WAL position if desired, or default to allowing Postgres to send you what it thinks you need

So just to confirm, these are the only tasks: (?)

• update slot: :temporary to accept a config var
• add a config var that reads from ENV (slot_name = System.get_env("SLOT_NAME") || :temporary)
• On our servers, add a new ENV var (something like SLOT_NAME=realtime)

[fracek]

Yes, that's exactly how I plan to implement it!

[kiwicopple]

Nice one @fracek 👍

[awalias]

@kiwicopple Do I need to add this env var to KPS once we include this latest realtime version?

[kiwicopple]

Ah yes please @awalias . The env var is SLOT_NAME

[awalias]

KPS is shipped and frontend/worker will ship with next deployment