feldera · blp · Apr 24, 2026 · Apr 5, 2026 · Apr 23, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -268,7 +268,7 @@ tikv-jemallocator = "0.6.0"
 time = { version = "0.3.47", features = ["serde", "serde-well-known"] }
 tokio = "1.50.0"
 tokio-postgres = "0.7"
-tokio-stream = "0.1.15"
+tokio-stream = "0.1.18"
 tokio-util = "0.7.11"
 tracing = "0.1.40"
 tracing-subscriber = "0.3.20"
@@ -286,6 +286,9 @@ zip = "6.0.0"
 zstd = "0.12.0"
 backtrace = "0.3.75"
 parking_lot = "0.12.4"
+etl = { git = "https://github.com/supabase/etl", rev = "05cb11ae" }
+etl-config = { git = "https://github.com/supabase/etl", rev = "05cb11ae" }
+etl-postgres = { git = "https://github.com/supabase/etl", rev = "05cb11ae" }
 
 [workspace.metadata.release]
 release = false

diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs
@@ -3,6 +3,7 @@ use chrono::{DateTime, Utc};
 use dyn_clone::DynClone;
 use feldera_types::adapter_stats::ConnectorHealth;
 use feldera_types::config::FtModel;
+use feldera_types::coordination::Completion;
 use feldera_types::program_schema::Relation;
 use rmpv::{Value as RmpValue, ext::Error as RmpDecodeError};
 use serde::Deserialize;
@@ -766,6 +767,19 @@ pub trait InputConsumer: Send + Sync + DynClone {
     /// so connectors that have no custom metrics need not override it.
     fn set_custom_metrics(&self, _metrics: Arc<dyn ConnectorMetrics>) {}
 
+    /// Returns a watch receiver that tracks completion of pipeline steps.
+    ///
+    /// The receiver yields [`Completion`] values whose `total_completed_steps`
+    /// field indicates how many steps have been fully processed (circuit
+    /// execution + all output connectors).
+    ///
+    /// Input adapters that need to defer acknowledgment until data is durably
+    /// processed (e.g., CDC adapters controlling a replication slot) can use
+    /// this to detect when their data has been consumed.
+    ///
+    /// Return `None` if the consumer does not support completion tracking.
+    fn completion_watcher(&self) -> Option<tokio::sync::watch::Receiver<Completion>>;
+
     /// Endpoint failed.
     ///
     /// Reports that the endpoint failed and that it will not queue any more

diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml
@@ -24,6 +24,7 @@ default = [
     "with-pubsub",
     "with-redis",
     "with-nats",
+    "with-postgres-cdc",
 ]
 with-kafka = ["rdkafka"]
 with-deltalake = ["deltalake", "deltalake-catalog-unity"]
@@ -52,6 +53,7 @@ iceberg-tests-glue = []
 iceberg-tests-rest = []
 fips = ["rustls/fips"]
 bench-mode = []
+with-postgres-cdc = ["etl", "etl-config", "etl-postgres"]
 
 [dependencies]
 feldera-types = { workspace = true }
@@ -201,6 +203,9 @@ zip = { workspace = true }
 smallvec = { workspace = true }
 delta_kernel = { workspace = true }
 flate2 = { workspace = true }
+etl = { workspace = true, optional = true }
+etl-config = { workspace = true, optional = true }
+etl-postgres = { workspace = true, optional = true }
 
 [package.metadata.cargo-machete]
 ignored = ["num-traits"]

diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -7260,6 +7260,10 @@ impl InputConsumer for InputProbe {
         self.transaction_in_progress.store(false, Ordering::Release);
     }
 
+    fn completion_watcher(&self) -> Option<tokio::sync::watch::Receiver<Completion>> {
+        Some(self.controller.status.completion_notifier.subscribe())
+    }
+
     fn error(&self, fatal: bool, error: AnyError, tag: Option<&str>) {
         self.controller.input_transport_error(
             self.endpoint_id,

diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs
@@ -9,6 +9,8 @@ use std::sync::Weak;
 pub mod delta_table;
 mod postgres;
 
+#[cfg(feature = "with-postgres-cdc")]
+use crate::integrated::postgres::PostgresCdcInputEndpoint;
 use crate::integrated::postgres::PostgresInputEndpoint;
 pub use crate::integrated::postgres::PostgresOutputEndpoint;
 
@@ -99,6 +101,12 @@ pub fn create_integrated_input_endpoint(
         TransportConfig::PostgresInput(config) => {
             Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer))
         }
+        #[cfg(feature = "with-postgres-cdc")]
+        TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new(
+            endpoint_name,
+            config,
+            consumer,
+        )),
         transport => {
             return Err(ControllerError::unknown_input_transport(
                 endpoint_name,

diff --git a/crates/adapters/src/integrated/postgres.rs b/crates/adapters/src/integrated/postgres.rs
@@ -5,8 +5,14 @@ mod output_macros;
 mod prepared_statements;
 mod tls;
 
+#[cfg(feature = "with-postgres-cdc")]
+pub(crate) mod cdc_input;
+
 #[cfg(test)]
 mod test;
 
 pub use input::PostgresInputEndpoint;
 pub use output::PostgresOutputEndpoint;
+
+#[cfg(feature = "with-postgres-cdc")]
+pub use cdc_input::PostgresCdcInputEndpoint;