pg_flo copy and stream on a single container

[AbhinavKoul]

Hi,
First of all; an amazing project. Looks like I might be able to use it in one of my use cases. Something that brings me to this tool is how it tried to paralyze initial bulk copy in logical replication, giving it an edge over other tools.

I see the recommendations of using multiple boxes to spin up NATS, replicator and worker; but I am looking just to create a single box in AWS that would handle all of this. I am able to create a docker script that does that, with running NATS in the background; then spinning up a worker and letting it run in the background and then finally having the replicator work in the foreground.

I currently have just created a dummy DB to test out how this works, but what I see is that the replicator seems to remain stuck at LSN 0/<some big number>. I am not even trying out new committed transactions in the DB at the moment, but rather just looking to get the base copy in the target DB.
The only log about worker I get is

INF Starting worker component=worker group=test stream=pgflo_test_stream subject=pgflo.test

I would love to know more about this; and if I am missing something obvious in terms of letting it work.

[AbhinavKoul]

I have used the sample DB schema mentioned on the readme for this testing.

-- Using primary key (recommended)
CREATE TABLE users (
  id SERIAL PRIMARY KEY,
  email TEXT,
  name TEXT
);

-- Using unique constraint
CREATE TABLE orders (
  order_id TEXT NOT NULL,
  customer_id TEXT NOT NULL,
  data JSONB,
  CONSTRAINT orders_unique UNIQUE (order_id, customer_id)
);
ALTER TABLE orders REPLICA IDENTITY USING INDEX orders_unique;

-- Using all columns (higher overhead in terms of performance)
CREATE TABLE audit_logs (
  id SERIAL,
  action TEXT,
  data JSONB
);
ALTER TABLE audit_logs REPLICA IDENTITY FULL;

More logs for context.

13:24:02.932 DBG Sent standby status update component=replicator lsn=0/2A006BF0
13:23:48.735 INF Starting worker component=worker group=test stream=pgflo_test_stream subject=pgflo.test
13:23:48.735 INF Starting worker...
13:23:44.592 INF Replication started successfully component=replicator startLSN=0/2A006BF0
13:23:44.592 INF Starting replication component=replicator publication=pg_flo_test_publication startLSN=0/2A006BF0
13:23:44.591 INF Starting parallel copy component=replicator snapshotID=00000004-000005CE-1 startLSN=0/2A006BF0
13:23:44.590 INF Replication slot created successfully component=replicator consistentPoint=0/2A006BF0 slot=pg_flo_test_publication snapshotName=00000003-00000577-1
13:23:44.469 INF Creating replication slot component=replicator slot=pg_flo_test_publication
13:23:44.467 INF Publication created successfully component=replicator publication=pg_flo_test_publication
13:23:44.428 INF Starting PostgreSQL logical replication stream component=replicator database=<db_name> group=test host=<host_url> port=5432 publication=pg_flo_test_publication schema=public tables=[] user=postgres
13:23:44.428 DBG Initialized replication key configuration columns=[] component=replicator key_type=FULL replica_identity=f table=audit_logs
13:23:44.428 DBG Initialized replication key configuration columns=["order_id","customer_id"] component=replicator key_type=UNIQUE replica_identity=i table=orders
13:23:44.428 DBG Initialized replication key configuration columns=["id"] component=replicator key_type="PRIMARY KEY" replica_identity=d table=users
[7] 2025/01/16 13:23:43.359976 [INF] Server is ready
[7] 2025/01/16 13:23:43.359748 [INF] Listening for client connections on 0.0.0.0:4222

[shayonj]

Hey! Great question - if you are using copy and stream mode, and the stage has already reached streaming, then streams are resumable. It's built with the understanding that deployments may come in and roll out existing tasks/containers for new ones. It does some house keeping when a SIGTERM is arrived but can also tolerate container deaths.

pg_flo keeps track of the LSN while streaming to NATS. The same applies for streaming mode as well. You can find more information here: https://github.com/pgflo/pg_flo/blob/main/internal/how-it-works.md

However, if the container itself dies while in the copying stage, there is currently no way to resume (related issue: #24)

In that case, I typically recommend dropping the table in the target database and starting again.

Does that help answer your question? Happy to elaborate further, thanks!

[shayonj]

Update: I wrote an automation to extract tables, as this approach is more prudent in the long run.

It could be worthwhile to add a --exclude-tables feature to pg_flo. Good suggestion.

Regarding container exit after replication completion:

Both the replicator and worker respond to SIGTERM, which you can issue at any time. For the replicator, if the data has fully synced to NATS, you can safely shut it down. This assumes you're using copy-only mode, not copy + stream. For the worker, the timing depends on how close it is to completing synchronization with the target database. Debug lines are emitted for every flush, so you can track progress that way. However, it would be beneficial to have a pg_flo stats command or similar functionality (related issue: #36).

[AbhinavKoul]

@shayonj How can I be sure that it succeeded after sending a SIGTERM to the command? Does it send a successful Exit code? This is required to make sure that passing this signal in the container actually did a graceful shutdown.

[shayonj]

yes it it does end up with success exit code, if not, safe to assume exit process was not successful.

[AbhinavKoul]

Hi @shayonj, any thoughts on this?

[shayonj]

Heya! Thanks for your interest in the project. I am actually out / afk on vacation, but let me try to get back to you asap :).