Streaming replication protocol

[jeff-davis]

This idea isn't fully formed yet, but it would be nice if rust-postgres could support the streaming replication protocol: http://www.postgresql.org/docs/devel/static/protocol-replication.html

It doesn't make a lot of sense with the physical replication (though that could be supported, too). But for logical replication it could be quite interesting.

[sfackler]

cc @posix4e

[posix4e]

We actually have a project called elephant pump that attempts to provide a bridge to writing logical decoder in rust. We intend on working with sfackler to port our code to his library once we get a bit further

[jmealo]

Any updates on a logical decoding/streaming replication client for Rust?

[posix4e]

@jmealo besides your work at http://github.com/posix4e/jsoncdc :)

[eadanfahey]

Any updates on this?
I'm trying to create a logical replication slot, using the wal2json plugin, from Rust:

extern crate postgres;
use postgres::{Connection, TlsMode};

fn main() {
    let url = "postgresql://postgres:postgres@localhost:5432/postgres?replication=database";
    let rep_conn = Connection::connect(url, TlsMode::None).unwrap();
    rep_conn.execute("CREATE_REPLICATION_SLOT test_slot LOGICAL wal2json EXPORT_SNAPSHOT", &[]).unwrap();
}

This produces an error:

{ severity: "ERROR", parsed_severity: Some(Error), code: SqlState("08P01"), message: "extended query protocol not supported in a replication connection", detail: None, hint: None, position: None, where_: None, schema: None, table: None, column: None, datatype: None, constraint: None, file: Some("postgres.c"), line: Some(4339), routine: Some("forbidden_in_wal_sender") }

[jmealo]

@eadanfahey https://www.postgresql.org/docs/current/static/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY

I hope this helps. We need to support the simple query protocol to send the logical replication commands.

You could sniff the protocol from pg_recvlogical or checkout Lapidus to see what's required.

[ruslantalpa]

@sfackler is there still interest in this?
From my experiments, it looks like the state is this:
First config.rs needs support for the replication parameter (easy)

When this works, PG can enter into "replication" mode after the connection.
After this, the queries need to be sent using simple_query so that they are sent directly and not as "prepared statements" which triggers the error @eadanfahey posted.

Once the START_REPLICATION statement is sent, pg starts streaming back the "events".
Now the question becomes how does one capture those events. I tried looking at poll_message function from connection.rs but it looks like it's returning only the Async messages while the replication events come back to this part of the code as BackendMessage::Normal with request_complete=false and the code assumes (from my understanding) that this is a reply to a previously issued query and it starts looking for a sender of the query to send the reply back by doing self.responses.pop_front() and then later response.sender.start_send(messages) (at a lower level in psotgres-protocol they look like Some(CopyData(CopyDataBody { storage: b"..." })))

I could make a PR with the config.rs part but i am not sure how to handle making those replication messages available to the lib user. My rough thinking is to make a new addition to pub enum Message, similar to NoticeResponse(NoticeResponseBody), and then in the poll_read function, somehow detect (based on connection parameters, request_complete and that there is no sender expecting a repsonse) that this is a replication frame and return it as an Async message? Or should this be handled at a lower level in backend.rs or something in between (that part that turns CopyData message into BackendMessage::Normal, donno where that is yet )

thank you

[sfackler]

Definitely interested! Other than adding the replication parameter, I think what we'd need to do is just add a copy_both method for use with START_REPLICATION. I'd imagine it'd have some kind of sink + stream interface with some extra methods to terminate the copy.

[ruslantalpa]

@sfackler sorry, i am not that familiar with the with the layout of your lib/or pg protocol (i just cloned your repo, did a bunch of trace! to try to get the replication started)
What do you mean by copy_both method (is this a rust thing?) I was thinking of modeling the interface similar to how notifications work.

Just point me in the right direction and if i can i'll try and make this happen.

[sfackler]

START_REPLICATION switches the connection into bidirectional copy mode, sort of like how COPY ... TO STDOUT and COPY ... TO STDIN switch the connection into unidirectional copy mode. There are specific methods to handle those modes: https://docs.rs/tokio-postgres/0.5.5/tokio_postgres/struct.Client.html#method.copy_in. Support for replication would involve adding a similar copy_both method.

[jeff-davis]

I have made a lot of progress on my PR #696. Right now it's failing tests, but the biggest reason is because the image doesn't support logical replication.

Can I get some high-level feedback on the PR? I'd like to know whether (a) it's something you want, and there is just more work to do and you don't have the bandwidth to give it a review right now; or (b) you don't really think #696 is the right approach.

[jakajancar]

Re (a), just as a user, I would love this.

Right now I use pg_create_logical_replication_slot() and poll pg_logical_slot_get_changes() which is messy and high latency.

[honne23]

I have made a lot of progress on my PR #696. Right now it's failing tests, but the biggest reason is because the image doesn't support logical replication.

Can I get some high-level feedback on the PR? I'd like to know whether (a) it's something you want, and there is just more work to do and you don't have the bandwidth to give it a review right now; or (b) you don't really think #696 is the right approach.

Would love to contribute as the use-case is coming up quite frequently; might be worth integrating with https://www.datadoghq.com/blog/engineering/introducing-glommio/ for each wal-sender which is a rebuild of Seastar C++.

Could be super speedy

[honne23]

even perhaps https://github.com/tokio-rs/io-uring