feat: faster replication by ygxio · Pull Request #1166 · pgdogdev/pgdog

ygxio · 2026-07-08T19:09:17Z

fixes #1036

ygxio · 2026-07-08T19:09:59Z

I am still working on the design for it, just a peak if the direction seems well @levkk

levkk · 2026-07-08T19:14:58Z

Cool! Would be good to see some benchmark numbers, just to get a sense of how much faster this is.

ygxio · 2026-07-08T19:16:14Z

Do you have any such setup so that i can benchmark it?

codecov · 2026-07-08T19:19:44Z

Codecov Report

❌ Patch coverage is 83.28446% with 57 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...backend/replication/logical/subscriber/pipeline.rs	84.06%	47 Missing ⚠️
...c/backend/replication/logical/subscriber/stream.rs	80.00%	9 Missing ⚠️
pgdog/src/backend/replication/logical/error.rs	0.00%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

levkk · 2026-07-08T19:23:28Z

Do you have any such setup so that i can benchmark it?

This might work:

benches/bench.sh

Specifically, see resharding folder.

ygxio · 2026-07-08T19:38:48Z

PGDOG_BIN=/tmp/pgdog-new bash benches/resharding/replication/run.sh --baseline main

ygxio · 2026-07-08T19:59:27Z

For BENCH_SCALE = 3000, DEST_LATENCY_MS=10 we have more performance boosts

ygxio · 2026-07-08T20:07:18Z

RUNS = 2
WARMUP = 1
BENCH_SCALE = 5000
DEST_LATENCY_MS = 10
USE_TOXI = 1

@levkk results seems good but also too good at a point

levkk · 2026-07-08T20:11:56Z

Async should be a big improvement, so not that surprising. This is very cool. Let me know when you're ready for a review, I'll tag Kiryl.

ygxio · 2026-07-13T11:30:08Z

@levkk ready for review

meskill · 2026-07-15T22:32:04Z

+}
+
+#[cfg(test)]
+mod test {


Do you think we should add more unit tests here? I think we should have some for testing the whole flow - prepare -> execute -> await. And also some tests for possible errors

Sure @meskill would add more tests

We still miss some important tests:

error path - something that will trigger an error response, validate that we get the error properly and we won't block the execution for future calls

missed rows calculation - just some calls to generate missed rows and make sure we get proper stats for it

ygxio · 2026-07-16T11:37:32Z

@meskill i have added a few more could you take a look?

meskill · 2026-07-17T15:42:06Z

Thanks @ygxio I'm still checking the code more deeply, but anyway looking great so far.

Putting my bench results to confirm the improvements:

Against main

With defaults:

With toxi and latency for source/destination for 1ms:

Separately

And the next without comparison with main (compare against prev run) since since it is executing too long to get the result.

With latency 10ms:

With latency 100ms:

ygxio · 2026-07-17T15:54:05Z

i see from 1 to 10 to 100 ms of latency the jump has been pretty high, need to see the cause maybe

meskill · 2026-07-19T17:59:50Z

+    // One entry per outstanding DML op, in send order: `is_direct`. Popped on
+    // each CommandComplete.
+    direct: VecDeque<bool>,
+    // In-transaction prepare sync points: (remaining ParseComplete acks, waiter).
+    parse_ack_sync_points: VecDeque<(usize, oneshot::Sender<()>)>,
+    // Commit / out-of-transaction prepare sync points, resolved on ReadyForQuery.
+    ready_for_query_sync_points: VecDeque<oneshot::Sender<()>>,
+    // Waiter that resolves once every DML ack has been read. At most one is
+    // outstanding: `commit()` issues drain_acks sequentially, one per connection.
+    drain_waiter: Option<oneshot::Sender<()>>,


The multiple separate VecDeque like this works, but I see this more error-prone and harder to maintain. The issue is this multiple queues partially maps to the underlaying FIFO queue executed by the backend. Due to current design of the replication it works just fine, but in case we add something new to a processing or try to extend async handling that's could become a break point.

I'd suggest to use a single queue of enums to control the state of the processing and the expected responses, to make sure we get everything we expect and we don't process entries in the wrong order for some reason.
I think the ProtocolState from pgdog/src/backend/protocol/state.rs does use the single queue for something like this.

meskill · 2026-07-19T19:03:26Z

+impl Listener {
+    async fn run(mut self) {
+        loop {
+            select! {


I can't prove it but I believe here could be a theoretical issue due to disproportion of read/writes events - the usual execute command from recv expected to generate 2 server.read events. Since the select! is fair by default the buffer for server.read could grow fast that could lead to tcp stream blocking.

I'm thinking if we need to add biased to select! and move the read first, so we'll read the messages first if any to avoid this.

meskill · 2026-07-19T19:06:56Z

+        // latched on any connection means whole transaction is aborted.
+        for conn in &self.connections {
+            if let Some(err) = conn.take_error() {
+                return Err(err);


could you please check how does effect on one connection affect other connections and streams - are they cancelled in any way?
I think they stop after dropping the channel, but I want to be sure we are not doing useless work here.

meskill · 2026-07-19T19:30:21Z

+    // Rows a direct-to-shard DML expected to touch but didn't (0 rows affected).
+    missed: MissedRows,
+    // Test-only: total ParseComplete acks consumed by parse-ack sync points.
+    #[cfg(test)]


can we try to avoid this is the main code?

let's better test that everything works as expected after some commands than to rely on some implementation details. If we have prepare statements - we can validate that this statement work after parsing and that listener doesn't block for example.

meskill · 2026-07-19T19:35:17Z

+}
+
+#[cfg(test)]
+mod test {


We still miss some important tests:

error path - something that will trigger an error response, validate that we get the error properly and we won't block the execution for future calls

missed rows calculation - just some calls to generate missed rows and make sure we get proper stats for it

meskill · 2026-07-19T19:37:05Z

+            // If in transaction we send flush and wait for the acknowledgements
+            // since we donot want to commit the open transaction.
+            messages.push(Flush.into());
+            SyncPointKind::ParseAcks(parses.len())


let's add some validation that this should not be 0, since we may overflow on this later

meskill · 2026-07-19T19:38:25Z

-                    c => return Err(Error::RelationOutOfSync(c)),
-                }
-            }
+            server.prepare(parses, in_txn).await?;


also can speed up this a very little if we use try_join to run prepare in parallel

faster replication v1

e1972b6

ygxio changed the title ~~faster replication v1~~ faster replication Jul 8, 2026

ygxio added 4 commits July 13, 2026 16:27

cosmetic changes

5c7f1c4

minor correction

d93587a

making clippy happy

9d5f770

making fmt happy

0611283

ygxio changed the title ~~faster replication~~ feat: faster replication Jul 13, 2026

levkk requested a review from meskill July 13, 2026 17:42

meskill reviewed Jul 15, 2026

View reviewed changes

a few tests

d76e47c

meskill requested changes Jul 19, 2026

View reviewed changes

single queue v1

cd6367a

Uh oh!

Conversation

ygxio commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ygxio commented Jul 8, 2026

Uh oh!

levkk commented Jul 8, 2026

Uh oh!

ygxio commented Jul 8, 2026

Uh oh!

codecov Bot commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

levkk commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ygxio commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ygxio commented Jul 8, 2026

Uh oh!

ygxio commented Jul 8, 2026

Uh oh!

levkk commented Jul 8, 2026

Uh oh!

ygxio commented Jul 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ygxio commented Jul 16, 2026

Uh oh!

meskill commented Jul 17, 2026

Against main

Separately

Uh oh!

ygxio commented Jul 17, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ygxio commented Jul 8, 2026 •

edited

Loading

codecov Bot commented Jul 8, 2026 •

edited

Loading

levkk commented Jul 8, 2026 •

edited

Loading

ygxio commented Jul 8, 2026 •

edited

Loading