implement support for streaming replication WIP (for #116) #652

ruslantalpa · 2020-09-05T11:54:14Z

so this is how far I've gotten and i wanted to check if the direction is right (before cleaning the code, writing some docs and a better test).

The questions I have are (since i don't know what i am really doing here :))):

is this the way you think it should be implemented from the interface point of view?
i know you said to implement a Sync and i know the protocol is in "copy_both" state but why is it needed, what would one send in the sync back to PG, besides stop replication?
right now the output is raw messages, is this ok (leave the parsing to the user since i think the format of the body is dependent on the plugin) or should there be additional parsing of the body?
~~in the test you'll see that i do 3 inserts but receive 7 events back from replication, are there some messages that should be skipped and not returned?~~
~~when i do a stop_replication call, pg also logs an error unexpected standby message type "S", after receiving CopyDone any idea where that comes from?~~

ruslantalpa · 2020-09-07T18:06:54Z

does the PR have any hope of getting to a point of being good and merged or should I abandon it :)?

…tion

vitaly-burovoy · 2020-10-26T07:17:54Z

Hello, @ruslantalpa!

Thank you for the PR, it is much better than just "do it for us, please"!

Some thoughts to it…

I. The replication parameter in Config should not be a string since allowed values are only boolean (all its string variants: on, true, 1) for physical and "database" for logical replication:
https://www.postgresql.org/docs/12/protocol-replication.html
I.e. it is a triple-state parameter: none, physical and logical.

II. The LSN (Log Sequence Number) is u64 rather than i64 (see fn standby_status_update(…)) because it is "a byte position" which can not be negative.
https://www.postgresql.org/docs/current/datatype-pg-lsn.html

III. From my understanding[1] there is no "stop replication". The "START_REPLICATION" change the connection mode until disconnection.

IV.

is this the way you think it should be implemented from the interface point of view?

It is up to @sfackler, but I would like to see API which do not allow to make a mistake: it could be better if start_replication consumes an InnerClient and returns a ReplicationStream and a proxy struct which only allows to send "standby_status_update" and close the connection because nothing (except Standby status update/Hot Standby feedback message) should be sent: my experiments show the connection just hangs if I send something even with a simple_query command.

V. May the start_replication create its own "START_REPLICATION SLOT xxx YYYY ..." string where YYYY can be "LOGICAL" or "PHYSICAL" because we can know it from the Config.replication param? But it is up to @sfackler (he mentioned the crate should be as simple as it can).

VI.

right now the output is raw messages, is this ok…?

I would return an object similar to Row (from the RowStream) like XLogRow with attributes from the received message to be used in replies via the standby_status_update.

VII.

in the test you'll see that i do 3 inserts but receive 7 events back from replication

You receive data from a plugin. It can send empty messages. Each time DB detects any successful commit (in WAL) it calls a callback with changed rows (one call per row) and then "commit" callback which may not detect there was no rows for it.

—

Supporting logical replication is an amazing feature for me, I can publish changes on top of your commits (it is better to squash them in a single one and rebase to the current master) but I do not know how to add my commit to this PR.

—
[1]

when i do a stop_replication call, pg also logs an error unexpected standby message type "S", after receiving CopyDone any idea where that comes from?

It seems the protocol allows to stop copying and do something else but I do not think it is a real-world case.
Also the answer to you question is in the PG code ('X' is Terminate command):
https://git.postgresql.org/gitweb/?p=postgresql.git;a=blob;f=src/backend/replication/walsender.c;h=6d959f9ecf8dc056bb4da5a318825b3a755ecc86;hb=29be9983a64c011eac0b9ee29895cce71e15ea77#l1734

/*
 * If we already received a CopyDone from the frontend, the frontend
 * should not send us anything until we've closed our end of the COPY.
 * XXX: In theory, the frontend could already send the next command
 * before receiving the CopyDone, but libpq doesn't currently allow
 * that.
 */
if (streamingDoneReceiving && firstchar != 'X')
    ereport(FATAL,
            (errcode(ERRCODE_PROTOCOL_VIOLATION),
             errmsg("unexpected standby message type \"%c\", after receiving CopyDone",
                    firstchar)));

ruslantalpa · 2020-10-26T09:21:48Z

i edited my original message in this PR to reflect things i "fixed/figured out".
Since then, also, i am using this code for my needs and it seems to work well

sfackler · 2020-10-28T00:06:15Z

tokio-postgres/src/client.rs

+    }
+
+    /// Stoppes the current replication by sending a copy_done message
+    pub async fn stop_replication(&self) -> Result<(), Error> {


This should be handled on the stream type like it is for copy_in.

sfackler · 2020-10-28T00:07:52Z

tokio-postgres/src/client.rs

+    }
+
+    /// Notifies PostgreSQL of the last processed WAL
+    pub async fn standby_status_update(


This is sent as part of the CopyBoth stream, not at the top level: https://www.postgresql.org/docs/13/protocol-replication.html.

jeff-davis

I had just started hacking on this problem today, and then I found your PR. Thanks for working on this!

jeff-davis · 2020-11-13T00:09:20Z

tokio-postgres/src/replication.rs

+}
+
+impl Stream for ReplicationStream {
+    type Item = Result<Bytes, Error>;


I'd suggest an enum here that describes the messages coming back at least as far as the protocol defines them, e.g. XLogData and PrimaryKeepAlive.

At least for now, we don't need to parse the WAL data itself, but we can represent all of the information that the protocol defines.

jeff-davis · 2020-11-13T00:11:31Z

tokio-postgres/src/replication.rs

+    Ok(())
+}
+
+pub async fn standby_status_update(


Did you consider also adding support for:

identify_system

base_backup

hot standby feedback

creating/dropping slots

timeline history

show

(I'm not the crate maintainer so don't take these as blockers. It might be fine to do those in later PRs after the basic support is in.)

jeff-davis · 2020-11-13T00:12:11Z

tokio-postgres/src/client.rs

@@ -3,6 +3,7 @@ use crate::config::{Host, SslMode};
 use crate::connection::{Request, RequestMessages};
 use crate::copy_out::CopyOutStream;
 use crate::query::RowStream;
+use crate::replication::ReplicationStream;


Should we have separate LogicalReplicationStream and PhysicalReplicationStream?

jeff-davis · 2020-11-13T00:17:16Z

tokio-postgres/src/config.rs

@@ -548,6 +564,7 @@ impl fmt::Debug for Config {
            .field("keepalives_idle", &self.keepalives_idle)
            .field("target_session_attrs", &self.target_session_attrs)
            .field("channel_binding", &self.channel_binding)
+            .field("replication", &self.replication)


Instead of making this an ad-hoc parameter, should we make this a different connection type? That would add better compile-time checking that you decide what you are going to use the connection for (normal, physical replication, or logical replication).

jeff-davis · 2020-11-15T23:47:06Z

Just to put some code behind my ideas, I created #696 . It's not meant to replace this PR, but you can have a look.

ruslantalpa · 2022-09-13T05:45:45Z

Closing in favor of #778

implement support for streaming replication WIP1

ae8b799

ruslantalpa added 3 commits October 5, 2020 12:24

Merge remote-tracking branch 'upstream/master' into streaming_replica…

2780b2e

…tion

fmt code and use wal2json in tests

e016a04

replication: send_status_update and cleanup test

24e540b

ruslantalpa force-pushed the streaming_replication branch from 8587850 to 24e540b Compare October 8, 2020 08:44

sfackler reviewed Oct 28, 2020

View reviewed changes

jeff-davis reviewed Nov 13, 2020

View reviewed changes

jeff-davis mentioned this pull request Nov 15, 2020

Replication support (#116) #696

Open

ruslantalpa closed this Sep 13, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

implement support for streaming replication WIP (for #116) #652

implement support for streaming replication WIP (for #116) #652

Uh oh!

ruslantalpa commented Sep 5, 2020 •

edited

Loading

Uh oh!

ruslantalpa commented Sep 7, 2020

Uh oh!

vitaly-burovoy commented Oct 26, 2020

Uh oh!

ruslantalpa commented Oct 26, 2020

Uh oh!

sfackler Oct 28, 2020

Uh oh!

sfackler Oct 28, 2020

Uh oh!

jeff-davis left a comment

Uh oh!

jeff-davis Nov 13, 2020

Uh oh!

jeff-davis Nov 13, 2020

Uh oh!

jeff-davis Nov 13, 2020

Uh oh!

jeff-davis Nov 13, 2020

Uh oh!

jeff-davis commented Nov 15, 2020

Uh oh!

ruslantalpa commented Sep 13, 2022

Uh oh!

Uh oh!

Uh oh!

implement support for streaming replication WIP (for #116) #652

implement support for streaming replication WIP (for #116) #652

Uh oh!

Conversation

ruslantalpa commented Sep 5, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ruslantalpa commented Sep 7, 2020

Uh oh!

vitaly-burovoy commented Oct 26, 2020

Uh oh!

ruslantalpa commented Oct 26, 2020

Uh oh!

sfackler Oct 28, 2020

Choose a reason for hiding this comment

Uh oh!

sfackler Oct 28, 2020

Choose a reason for hiding this comment

Uh oh!

jeff-davis left a comment

Choose a reason for hiding this comment

Uh oh!

jeff-davis Nov 13, 2020

Choose a reason for hiding this comment

Uh oh!

jeff-davis Nov 13, 2020

Choose a reason for hiding this comment

Uh oh!

jeff-davis Nov 13, 2020

Choose a reason for hiding this comment

Uh oh!

jeff-davis Nov 13, 2020

Choose a reason for hiding this comment

Uh oh!

jeff-davis commented Nov 15, 2020

Uh oh!

ruslantalpa commented Sep 13, 2022

Uh oh!

Uh oh!

ruslantalpa commented Sep 5, 2020 •

edited

Loading