feat: faster replication

pgdog/src/backend/replication/logical/error.rs

@@ -144,6 +144,9 @@ pub enum Error {
     #[error("parallel connection error")]
     ParallelConnection,
 
+    #[error("pipelined connection task closed")]
+    PipelineClosed,
+
     #[error("no replicas available for table sync")]
     NoReplicas,
 
@@ -274,6 +277,10 @@ impl Error {
             // either the ParallelConnection wrapper should be removed or
             // the proper error should be propagated
             Self::ParallelConnection => true,
+            // A pipelined shard task closed (write failed, socket died, or the
+            // handle was dropped). The transaction is torn down; retry from a
+            // fresh connection.
+            Self::PipelineClosed => true,
             _ => false,
         }
     }

pgdog/src/backend/replication/logical/subscriber/mod.rs

@@ -2,6 +2,7 @@ pub mod context;
 pub mod copy;
 pub mod omni_ownership;
 pub mod parallel_connection;
+pub mod pipeline;
 pub mod stream;
 
 #[cfg(test)]
@@ -10,4 +11,5 @@ mod tests;
 pub use context::StreamContext;
 pub use copy::CopySubscriber;
 pub use parallel_connection::ParallelConnection;
+pub use pipeline::PipelinedConnection;
 pub use stream::StreamSubscriber;