pgsync_core/
traits.rs

1//! Boundaries between the engine and database-specific adapters.
2
3use crate::Result;
4use crate::checkpoint::Checkpoint;
5use crate::event::{Op, Relation, SourceEvent};
6use crate::lsn::Lsn;
7use std::future::Future;
8use std::sync::Arc;
9
10/// Standby status update sent to the source.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub struct Feedback {
13    pub written: Lsn,
14    pub flushed: Lsn,
15    pub applied: Lsn,
16    pub reply_requested: bool,
17}
18
19/// An open logical replication stream.
20pub trait EventStream: Send {
21    /// Next event in commit order. Must be cancel-safe: dropping the future must not
22    /// lose or corrupt a partially received message.
23    fn next_event(&mut self) -> impl Future<Output = Result<SourceEvent>> + Send;
24
25    /// Report progress to the server. Only positions that are durable in the target may be
26    /// reported as flushed.
27    fn send_feedback(&mut self, fb: Feedback) -> impl Future<Output = Result<()>> + Send;
28}
29
30/// Produces streams; called again after a transient failure.
31pub trait StreamFactory: Send {
32    type Stream: EventStream;
33    /// Open a stream resuming from `cp.committed_lsn` (the server may start later, at its
34    /// confirmed flush position). Implementations verify source identity and slot
35    /// continuity against `cp` and fail with `ResyncRequired` when continuity cannot be
36    /// proven. Returns the stream and the server's current WAL end.
37    fn connect(
38        &mut self,
39        cp: &Checkpoint,
40    ) -> impl Future<Output = Result<(Self::Stream, Lsn)>> + Send;
41}
42
43/// Applies transactions to a destination and persists checkpoints with them.
44pub trait ReplicationSink: Send {
45    /// Load the job's checkpoint; `None` if the job has never been bootstrapped.
46    fn load_checkpoint(
47        &mut self,
48        job_id: &str,
49    ) -> impl Future<Output = Result<Option<Checkpoint>>> + Send;
50
51    /// Persist a checkpoint outside of any data transaction (autocommit).
52    fn save_checkpoint(&mut self, cp: &Checkpoint) -> impl Future<Output = Result<()>> + Send;
53
54    /// Start a destination transaction.
55    fn begin(&mut self) -> impl Future<Output = Result<()>> + Send;
56
57    /// Relation metadata; may arrive inside or outside a transaction and must be cached.
58    fn relation(&mut self, rel: Arc<Relation>) -> impl Future<Output = Result<()>> + Send;
59
60    /// Apply one change inside the open transaction. Implementations may batch.
61    fn apply(&mut self, op: Op) -> impl Future<Output = Result<()>> + Send;
62
63    /// Flush pending changes, write `cp`, and commit atomically. After `Ok`, the data and
64    /// the checkpoint are durable.
65    fn commit(&mut self, cp: &Checkpoint) -> impl Future<Output = Result<()>> + Send;
66
67    /// Abort the open transaction, if any. Best effort; must not panic.
68    fn rollback(&mut self) -> impl Future<Output = Result<()>> + Send;
69}
70
71/// Produces sinks; called again after a transient failure.
72pub trait SinkFactory: Send {
73    type Sink: ReplicationSink;
74    fn connect(&mut self) -> impl Future<Output = Result<Self::Sink>> + Send;
75}
pgsync_core/traits.rs

pgsync_core/
traits.rs
