diff --git a/Cargo.lock b/Cargo.lock index 0902ca7dfcc..1e414f6f72a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1607,7 +1607,7 @@ dependencies = [ "http 1.3.1", "http-body 1.0.1", "http-body-util", - "md-5", + "md-5 0.10.6", "pin-project-lite", "sha1 0.10.6", "sha2 0.10.9", @@ -3633,7 +3633,7 @@ dependencies = [ "hex", "itertools 0.14.0", "log", - "md-5", + "md-5 0.10.6", "memchr", "num-traits", "rand 0.9.4", @@ -5011,27 +5011,26 @@ dependencies = [ [[package]] name = "etcetera" -version = "0.10.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26c7b13d0780cb82722fd59f6f57f925e143427e4a75313a6c77243bf5326ae6" +checksum = "de48cc4d1c1d97a20fd819def54b890cadde72ed3ad0c614822a0a433361be96" dependencies = [ "cfg-if", - "home", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] name = "etl" version = "0.1.0" -source = "git+https://github.com/supabase/etl?rev=b2aab62d#b2aab62db1830df5b72d1d09ad5c053ee8e44db2" +source = "git+https://github.com/supabase/etl?rev=8762c0fc#8762c0fc7501d206a13c800dfa8d8e8bbbb2311f" dependencies = [ - "byteorder", "bytes", "chrono", "etl-config", "etl-postgres", "fail", "futures", + "memchr", "metrics 0.24.3", "pg_escape", "pin-project-lite", @@ -5040,7 +5039,7 @@ dependencies = [ "rustls", "serde", "serde_json", - "sqlx 0.9.0-alpha.1", + "sqlx 0.9.0", "sysinfo 0.38.4", "tokio", "tokio-postgres 0.7.11", @@ -5052,12 +5051,12 @@ dependencies = [ [[package]] name = "etl-config" version = "0.1.0" -source = "git+https://github.com/supabase/etl?rev=b2aab62d#b2aab62db1830df5b72d1d09ad5c053ee8e44db2" +source = "git+https://github.com/supabase/etl?rev=8762c0fc#8762c0fc7501d206a13c800dfa8d8e8bbbb2311f" dependencies = [ "config", "secrecy", "serde", - "sqlx 0.9.0-alpha.1", + "sqlx 0.9.0", "thiserror 2.0.18", "tokio-postgres 0.7.11", "url", @@ -5066,17 +5065,19 @@ dependencies = [ [[package]] name = "etl-postgres" version = "0.1.0" -source = "git+https://github.com/supabase/etl?rev=b2aab62d#b2aab62db1830df5b72d1d09ad5c053ee8e44db2" +source = "git+https://github.com/supabase/etl?rev=8762c0fc#8762c0fc7501d206a13c800dfa8d8e8bbbb2311f" dependencies = [ "aws-lc-rs", + "byteorder", "bytes", + "chrono", "const-oid 0.9.6", "etl-config", "futures", "pg_escape", "rustls", "serde_json", - "sqlx 0.9.0-alpha.1", + "sqlx 0.9.0", "thiserror 2.0.18", "tokio", "tokio-postgres 0.7.11", @@ -5489,7 +5490,7 @@ dependencies = [ "itertools 0.14.0", "lexical-core", "like", - "md-5", + "md-5 0.10.6", "num", "num-traits", "once_cell", @@ -5674,9 +5675,9 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8" [[package]] name = "flume" -version = "0.11.1" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", @@ -5760,9 +5761,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", "futures-sink", @@ -5770,15 +5771,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" [[package]] name = "futures-executor" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" dependencies = [ "futures-core", "futures-task", @@ -5798,9 +5799,9 @@ dependencies = [ [[package]] name = "futures-io" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" [[package]] name = "futures-lite" @@ -5817,9 +5818,9 @@ dependencies = [ [[package]] name = "futures-macro" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", @@ -5828,15 +5829,15 @@ dependencies = [ [[package]] name = "futures-sink" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" [[package]] name = "futures-task" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" [[package]] name = "futures-timer" @@ -5846,9 +5847,9 @@ checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24" [[package]] name = "futures-util" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-channel", "futures-core", @@ -5858,7 +5859,6 @@ dependencies = [ "futures-task", "memchr", "pin-project-lite", - "pin-utils", "slab", ] @@ -6252,6 +6252,15 @@ dependencies = [ "hashbrown 0.15.3", ] +[[package]] +name = "hashlink" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824e001ac4f3012dd16a264bec811403a67ca9deb6c102fc5049b32c4574b35f" +dependencies = [ + "hashbrown 0.16.1", +] + [[package]] name = "hdrhist" version = "0.5.0" @@ -6310,6 +6319,15 @@ dependencies = [ "hmac 0.12.1", ] +[[package]] +name = "hkdf" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" +dependencies = [ + "hmac 0.13.0", +] + [[package]] name = "hmac" version = "0.12.1" @@ -7549,6 +7567,16 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.2", +] + [[package]] name = "mea" version = "0.6.4" @@ -7560,9 +7588,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.8.1" +version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "memmap2" @@ -8061,7 +8089,7 @@ dependencies = [ "humantime", "hyper 1.6.0", "itertools 0.14.0", - "md-5", + "md-5 0.10.6", "parking_lot 0.12.4", "percent-encoding", "quick-xml 0.38.3", @@ -8101,7 +8129,7 @@ dependencies = [ "humantime", "hyper 1.6.0", "itertools 0.14.0", - "md-5", + "md-5 0.10.6", "parking_lot 0.12.4", "percent-encoding", "quick-xml 0.39.4", @@ -8187,7 +8215,7 @@ dependencies = [ "http-body 1.0.1", "jiff", "log", - "md-5", + "md-5 0.10.6", "mea", "percent-encoding", "quick-xml 0.38.3", @@ -8290,7 +8318,7 @@ dependencies = [ "crc32c", "http 1.3.1", "log", - "md-5", + "md-5 0.10.6", "opendal-core", "quick-xml 0.38.3", "reqsign-aws-v4", @@ -9099,7 +9127,7 @@ dependencies = [ "bytes", "fallible-iterator", "hmac 0.12.1", - "md-5", + "md-5 0.10.6", "memchr", "rand 0.8.6", "sha2 0.10.9", @@ -9117,7 +9145,7 @@ dependencies = [ "bytes", "fallible-iterator", "hmac 0.12.1", - "md-5", + "md-5 0.10.6", "memchr", "rand 0.9.4", "sha2 0.10.9", @@ -11858,14 +11886,14 @@ dependencies = [ [[package]] name = "sqlx" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "decccfa5f2f3eac95eb68085cfe69a0172fa9711666c3a634cfc806d4fb74a47" +checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71" dependencies = [ - "sqlx-core 0.9.0-alpha.1", - "sqlx-macros 0.9.0-alpha.1", + "sqlx-core 0.9.0", + "sqlx-macros 0.9.0", "sqlx-mysql", - "sqlx-postgres 0.9.0-alpha.1", + "sqlx-postgres 0.9.0", "sqlx-sqlite", ] @@ -11906,9 +11934,9 @@ dependencies = [ [[package]] name = "sqlx-core" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86854e8c6aba0dafcf1c04b4836b0b7fa3a20c560e3554567afefe1258fa4e60" +checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" dependencies = [ "base64 0.22.1", "bytes", @@ -11922,7 +11950,7 @@ dependencies = [ "futures-io", "futures-util", "hashbrown 0.16.1", - "hashlink 0.10.0", + "hashlink 0.11.1", "indexmap 2.13.0", "log", "memchr", @@ -11937,7 +11965,7 @@ dependencies = [ "tokio-stream", "tracing", "url", - "webpki-roots 0.26.11", + "webpki-roots 1.0.0", ] [[package]] @@ -11955,14 +11983,14 @@ dependencies = [ [[package]] name = "sqlx-macros" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7aab9442ed1568e3aed6c368737226ee4e0e8d1deb0e0887fa6bf15282ace44" +checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf" dependencies = [ "proc-macro2", "quote", - "sqlx-core 0.9.0-alpha.1", - "sqlx-macros-core 0.9.0-alpha.1", + "sqlx-core 0.9.0", + "sqlx-macros-core 0.9.0", "syn 2.0.117", ] @@ -11991,9 +12019,9 @@ dependencies = [ [[package]] name = "sqlx-macros-core" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "34eb4976b8f02ac57ee98d4ce40cd1aad7ab31d9792977bc3171f787ba6ba2fb" +checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3" dependencies = [ "cfg-if", "dotenvy", @@ -12005,9 +12033,9 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", - "sqlx-core 0.9.0-alpha.1", + "sqlx-core 0.9.0", "sqlx-mysql", - "sqlx-postgres 0.9.0-alpha.1", + "sqlx-postgres 0.9.0", "sqlx-sqlite", "syn 2.0.117", "thiserror 2.0.18", @@ -12017,43 +12045,28 @@ dependencies = [ [[package]] name = "sqlx-mysql" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6fef16f3d52a3710a672b48175b713e86476e2df85576a753c8b37ad11a483c0" +checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27" dependencies = [ - "atoi", - "base64 0.22.1", "bitflags 2.10.0", "byteorder", "bytes", "crc", - "digest 0.10.7", + "digest 0.11.2", "dotenvy", "either", - "futures-channel", "futures-core", - "futures-io", "futures-util", "generic-array", - "hex", - "hkdf", - "hmac 0.12.1", - "itoa", "log", - "md-5", - "memchr", "percent-encoding", - "rand 0.8.6", - "rsa", "serde", - "sha1 0.10.6", - "sha2 0.10.9", - "smallvec", - "sqlx-core 0.9.0-alpha.1", - "stringprep", + "sha1 0.11.0", + "sha2 0.11.0", + "sqlx-core 0.9.0", "thiserror 2.0.18", "tracing", - "whoami", ] [[package]] @@ -12073,12 +12086,12 @@ dependencies = [ "futures-core", "futures-util", "hex", - "hkdf", + "hkdf 0.12.4", "hmac 0.12.1", "home", "itoa", "log", - "md-5", + "md-5 0.10.6", "memchr", "once_cell", "rand 0.8.6", @@ -12090,14 +12103,14 @@ dependencies = [ "stringprep", "thiserror 2.0.18", "tracing", - "whoami", + "whoami 1.6.0", ] [[package]] name = "sqlx-postgres" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f053cf36ecb2793a9d9bb02d01bbad1ef66481d5db6ff5ab2dfb7b070cc0d13c" +checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", "base64 0.22.1", @@ -12105,38 +12118,38 @@ dependencies = [ "byteorder", "crc", "dotenvy", - "etcetera 0.10.0", + "etcetera 0.11.0", "futures-channel", "futures-core", "futures-util", "hex", - "hkdf", - "hmac 0.12.1", - "home", + "hkdf 0.13.0", + "hmac 0.13.0", "itoa", "log", - "md-5", + "md-5 0.11.0", "memchr", - "rand 0.8.6", + "rand 0.10.1", "serde", "serde_json", - "sha2 0.10.9", + "sha2 0.11.0", "smallvec", - "sqlx-core 0.9.0-alpha.1", + "sqlx-core 0.9.0", "stringprep", "thiserror 2.0.18", "tracing", - "whoami", + "whoami 2.1.2", ] [[package]] name = "sqlx-sqlite" -version = "0.9.0-alpha.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe2cd6cee87120b1e1dd31356b5589911995c777707e49f2750eec7c7fe43eef" +checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" dependencies = [ "atoi", "flume", + "form_urlencoded", "futures-channel", "futures-core", "futures-executor", @@ -12146,8 +12159,7 @@ dependencies = [ "log", "percent-encoding", "serde", - "serde_urlencoded", - "sqlx-core 0.9.0-alpha.1", + "sqlx-core 0.9.0", "thiserror 2.0.18", "tracing", "url", @@ -12817,7 +12829,7 @@ dependencies = [ "socket2 0.5.10", "tokio", "tokio-util", - "whoami", + "whoami 1.6.0", ] [[package]] @@ -12843,7 +12855,7 @@ dependencies = [ "socket2 0.5.10", "tokio", "tokio-util", - "whoami", + "whoami 1.6.0", ] [[package]] @@ -13960,6 +13972,12 @@ dependencies = [ "web-sys", ] +[[package]] +name = "whoami" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998767ef88740d1f5b0682a9c53c24431453923962269c2db68ee43788c5a40d" + [[package]] name = "winapi" version = "0.3.9" diff --git a/Cargo.toml b/Cargo.toml index c43e29aee4b..f5806cf81de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -318,9 +318,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 = "b2aab62d" } -etl-config = { git = "https://github.com/supabase/etl", rev = "b2aab62d" } -etl-postgres = { git = "https://github.com/supabase/etl", rev = "b2aab62d" } +etl = { git = "https://github.com/supabase/etl", rev = "8762c0fc" } +etl-config = { git = "https://github.com/supabase/etl", rev = "8762c0fc" } +etl-postgres = { git = "https://github.com/supabase/etl", rev = "8762c0fc" } [workspace.metadata.release] release = false diff --git a/crates/adapters/src/integrated/postgres/cdc_input.rs b/crates/adapters/src/integrated/postgres/cdc_input.rs index ca0fa09b91b..a3efd62a23c 100644 --- a/crates/adapters/src/integrated/postgres/cdc_input.rs +++ b/crates/adapters/src/integrated/postgres/cdc_input.rs @@ -1,3 +1,73 @@ +//! PostgreSQL change-data-capture input connector. +//! +//! [`PostgresCdcInputInner::worker_task_inner`] constructs and runs an `etl` [`Pipeline`] that +//! snapshots a PostgreSQL publication and then follows its logical replication stream. `etl` +//! persists table-copy phases and replication progress in the source database through its +//! [`PostgresStore`]. The connector forwards only the configured `source_table` to Feldera, even +//! when the publication contains other tables. +//! +//! # Data flow +//! +//! ```text +//! PostgreSQL snapshot and WAL +//! | +//! v +//! etl Pipeline + PostgresStore +//! | +//! v +//! FelderaDestination (rows/events -> JSON inserts and deletes) +//! | +//! v +//! CdcInputQueue (parsed buffers + deferred acknowledgments) +//! | +//! v +//! PostgresCdcInputReader -> Feldera circuit step +//! ``` +//! +//! [`FelderaDestination`] converts snapshot rows and replication events into Feldera input +//! records. It splits large writes into bounded buffers and pushes them into [`CdcInputQueue`]. +//! When the controller requests input, [`PostgresCdcInputReader`] flushes up to the connector's +//! batch limit. Every deferred acknowledgment encountered in that flush is assigned to the same +//! circuit step, which has flushed all the data covered by each acknowledgment. +//! +//! # Acknowledgments and durability +//! +//! etl waits for each destination acknowledgment before it can continue. With completion +//! tracking, snapshot batches are accepted while their data accumulates in the queue. +//! +//! etl then sends an empty write that terminates a snapshot, which is queued behind that data +//! as a durability barrier; etl finishes the copy only after the barrier becomes durable. +//! +//! A streaming write's [`DeferredAck`] is attached to its final queue entry (each streaming write may +//! produce multiple queue entries). Once that entry is flushed, the completion watcher ties the +//! acknowledgment to the resulting circuit step. It reports `Durable` after the step completes in +//! fast mode, or after a checkpoint containing the step completes when fault tolerance is enabled. +//! This prevents the PostgreSQL replication slot from advancing beyond Feldera's completion frontier. +//! +//! A streaming acknowledgment marked `MayDefer` that waits longer than +//! `streaming_ack_hold_ms` may be reported as `Accepted`. This lets etl continue reading WAL +//! without advancing the replication slot; a later durable acknowledgment confirms the accepted +//! prefix. etl marks terminal writes as `RequireDurable`; those acknowledgments never time out and +//! wait for the completion frontier. While an acknowledgment remains queued, the reader requests +//! bounded steps so it does not wait for the controller's normal buffer timeout. +//! +//! # Lifecycle and errors +//! +//! Pausing the Feldera pipeline stops the destination from accepting new etl batches; already +//! queued records can still drain. [`TableErrorMonitor`] turns non-retriable per-table etl errors, +//! such as unsupported source schema changes, into connector errors instead of allowing the input +//! to stall silently. Terminating or dropping the connector shuts down the etl pipeline and its +//! completion watcher. +//! +//! etl normally drains pending streaming writes during shutdown. If the completion watcher or +//! destination side disappears first, etl observes the dropped acknowledgment as a destination +//! error and may persist it in the table state. The same state can be left by a failed run in +//! which etl has enough time to observe the closed acknowledgment before the process exits. Since +//! Feldera never reported that write as [`DestinationWriteStatus::Durable`], the saved PostgreSQL +//! replication position does not include it and the write must be replayed. On startup, +//! `discard_shutdown_errors` therefore defaults to `true` and rolls back only this dropped-ack +//! error. Other persisted table errors remain intact unless `discard_table_errors` is enabled. + use crate::transport::{ InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint, NonFtInputReaderCommand, }; @@ -5,24 +75,21 @@ use crate::{ControllerError, InputConsumer, InputReader, PipelineState, RecordFo use anyhow::{Result as AnyResult, anyhow}; use chrono::Utc; use dbsp::circuit::tokio::TOKIO; -use etl::concurrency::ShutdownTx; use etl::config::{ BatchConfig, InvalidatedSlotBehavior, MemoryBackpressureConfig, PgConnectionConfig, PipelineConfig, TableSyncCopyConfig, TcpKeepaliveConfig, }; -use etl::destination::Destination; -use etl::destination::async_result::{ - DropTableForCopyResult, WriteEventsResult, WriteTableRowsResult, +use etl::data::{ArrayCell, Cell, OldTableRow, TableRow, UpdatedTableRow}; +use etl::destination::{ + Destination, DestinationWriteStatus, DropTableForCopyResult, WriteEventsDurability, + WriteEventsResult, WriteTableRowsResult, }; use etl::error::{ErrorKind, EtlResult}; use etl::etl_error; -use etl::pipeline::Pipeline; -use etl::state::{TableRetryPolicy, TableState}; -use etl::store::both::postgres::PostgresStore; -use etl::store::state::StateStore; -use etl::types::{ - ArrayCell, Cell, Event, OldTableRow, ReplicatedTableSchema, TableRow, UpdatedTableRow, -}; +use etl::event::Event; +use etl::pipeline::{Pipeline, ShutdownTx}; +use etl::schema::ReplicatedTableSchema; +use etl::store::{PostgresStore, StateStore, TableRetryPolicy, TableState}; use feldera_adapterlib::catalog::{DeCollectionStream, InputCollectionHandle}; use feldera_adapterlib::format::ParseError; use feldera_adapterlib::transport::{Resume, Watermark}; @@ -32,19 +99,189 @@ use feldera_types::format::json::JsonFlavor; use feldera_types::transport::postgres::{PostgresCdcReaderConfig, PostgresTlsConfig}; use serde_json::{Value, json}; use std::collections::BTreeSet; +use std::future::pending; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; +use std::time::{Duration, Instant}; use tokio::select; use tokio::sync::mpsc; use tokio::sync::watch::{Receiver, Sender, channel}; +use tokio::time::{Instant as TokioInstant, sleep_until}; use tracing::{debug, error, info, warn}; use url::Url; use xxhash_rust::xxh3::xxh3_64; use super::tls::make_etl_tls_config; -/// Deferred async result senders waiting for step completion. -type DeferredSenders = Vec>; +const DROPPED_DESTINATION_ACK_ERROR: &str = + "[DestinationError] Async result channel closed before sending"; +const MAX_ERROR_ROLLBACKS_PER_TABLE: usize = 32; + +/// An etl write acknowledgment deferred until Feldera has processed the data it covers. +/// The ack is tagged by write type because snapshot and stream acks follow different rules. +enum DeferredAck { + /// etl sends an empty table-copy write once we've accepted the snapshot batches. + /// We need to keep it around to mark that the snapshot has now become durable. + /// Answering `Accepted` on this empty table-copy write is illegal. + SnapshotCopyBarrier(WriteTableRowsResult), + /// Each `write_events` call may produce multiple smaller batches; we attach + /// this ack to the last one. `write_events` fires both during a table's + /// post-copy catchup and in steady-state streaming. + /// + /// We answer `Durable` as soon as the completion frontier passes it. A + /// `MayDefer` write may instead be answered `Accepted` after + /// `streaming_ack_hold_ms`; a `RequireDurable` write waits for the frontier. + Stream { + result: WriteEventsResult, + durability: WriteEventsDurability, + }, + #[cfg(test)] + Test { + kind: TestAckKind, + tx: std::sync::mpsc::Sender, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[cfg(test)] +enum TestAckKind { + SnapshotCopyBarrier, + StreamMayDefer, + StreamRequireDurable, +} + +impl DeferredAck { + /// Report the ack as `Durable`, letting etl advance the replication slot + /// or finish the table copy. + fn complete(self) { + match self { + Self::SnapshotCopyBarrier(result) | Self::Stream { result, .. } => { + result.send(Ok(DestinationWriteStatus::Durable)) + } + #[cfg(test)] + Self::Test { tx, .. } => { + let _ = tx.send(TestAckStatus::Durable); + } + } + } + + /// Report a stream ack as `Accepted`: Feldera has accepted the rows for processing, + /// but they are not durable yet, so etl may keep reading WAL without moving the slot. + fn accept(self) { + match self { + Self::Stream { + result, + durability: WriteEventsDurability::MayDefer, + } => result.send(Ok(DestinationWriteStatus::Accepted)), + Self::Stream { + durability: WriteEventsDurability::RequireDurable, + .. + } => unreachable!("a required-durability stream ack is never accepted"), + Self::SnapshotCopyBarrier(_) => { + unreachable!("the snapshot copy barrier is never accepted") + } + #[cfg(test)] + Self::Test { + kind: TestAckKind::StreamMayDefer, + tx, + } => { + let _ = tx.send(TestAckStatus::Accepted); + } + #[cfg(test)] + Self::Test { .. } => { + unreachable!("a non-deferrable ack is never accepted") + } + } + } + + fn is_stream(&self) -> bool { + match self { + Self::SnapshotCopyBarrier(_) => false, + Self::Stream { .. } => true, + #[cfg(test)] + Self::Test { kind, .. } => matches!( + kind, + TestAckKind::StreamMayDefer | TestAckKind::StreamRequireDurable + ), + } + } + + /// Whether etl permits this stream write to complete as `Accepted`. + fn may_defer(&self) -> bool { + match self { + Self::SnapshotCopyBarrier(_) => false, + Self::Stream { durability, .. } => *durability == WriteEventsDurability::MayDefer, + #[cfg(test)] + Self::Test { kind, .. } => *kind == TestAckKind::StreamMayDefer, + } + } +} + +/// Deferred etl acks stored as auxiliary data on a queue entry. +/// Each ack answers one etl destination call. +type DeferredSenders = Vec; + +/// FIFO input queue for parsed CDC data. Each entry carries [`DeferredSenders`] +/// as auxiliary data; see [`DeferredAck`] for what each variant answers and when. +/// +/// One etl write may be parsed into several queue entries but has at most one +/// ack, so most entries carry none; the ack sits on the write's final entry. +/// +/// ```text +/// etl write([e1, e2, e3, e4]) +/// | +/// +--> queue entry: [e1, e2] acks: [] +/// +--> queue entry: [e3] acks: [] +/// `--> queue entry: [e4] acks: [ack] +/// +/// FIFO flush order: entry 1 -> entry 2 -> entry 3 + ack +/// ``` +/// +/// That positioning is what lets [`PostgresCdcInputReader::request`] hand each +/// ack to a step that flushed everything it covers. A bounded flush may cross +/// several ack boundaries; all of those acks share the same step completion. +type CdcInputQueue = InputQueue; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[cfg(test)] +enum TestAckStatus { + Accepted, + Durable, +} + +struct PendingAcks { + /// `total_completed_steps` observed right after the queue flush. The + /// flushed data lands in the next step, so these acks are done once + /// `total_completed_steps >= step_at_flush + 1` (or the corresponding + /// checkpoint frontier reaches that step in strict mode). + step_at_flush: u64, + /// When the acks were assigned to a step. The stream ack hold deadline + /// counts from here. + held_since: Instant, + senders: DeferredSenders, +} + +/// Two ways an ack reaches the completion watcher. +/// +/// A stream ack that times out as `Accepted` has no further data of its own +/// to confirm it durable; only a later write can do that. A `write_events` +/// call whose events all get filtered out (e.g. a change to another table in +/// the publication, or an update we cannot reconstruct) is our chance to do +/// so, but it queues no Feldera rows and so never goes through the input +/// queue: it never gets a `step_at_flush` to anchor it. This variant gives it +/// a path to the watcher without one. +enum CompletionMessage { + /// A write that queued data. Resolved against the `step_at_flush` on + /// [`PendingAcks`]. + Queued(PendingAcks), + /// A stream write that produced no Feldera rows: there is no queue entry, + /// so no step to wait on. But it is still an ack we can answer, and + /// answering it `Durable` means the completion frontier has caught up with + /// everything before it, including any earlier stream ack that timed out + /// as `Accepted`, so it doubles as confirmation for that data too. + NoRowStream(DeferredAck), +} /// Integrated input connector that reads from Postgres via logical replication (CDC). pub struct PostgresCdcInputEndpoint { @@ -165,13 +402,22 @@ impl InputReader for PostgresCdcInputReader { match command.as_nonft().unwrap() { NonFtInputReaderCommand::Queue => { - // Flush queue to circuit, collecting timestamps for watermarks. + // Flush the queue to the circuit, collecting timestamps for + // watermarks and every ack encountered by this bounded flush. + // Each ack sits on the final entry of the data it covers, so + // all collected acks can share this step's completion frontier. let (buffer_size, _hasher, flushed) = self.inner.queue.flush_with_aux(); - let watermarks: Vec = flushed - .iter() - .map(|(ts, _)| Watermark::new(*ts, None)) - .collect(); + let mut watermarks = Vec::new(); + let mut senders = Vec::new(); + for (ts, mut flushed_senders) in flushed { + watermarks.push(Watermark::new(ts, None)); + senders.append(&mut flushed_senders); + } + + self.inner + .queued_acks + .fetch_sub(senders.len(), Ordering::Release); // Build resume metadata so Feldera can checkpoint our position. // The actual resume state is managed by etl's PostgresStore; @@ -190,34 +436,50 @@ impl InputReader for PostgresCdcInputReader { .consumer .extended(buffer_size, Some(resume), watermarks); - // Take any deferred senders that write_events stored. - let senders: DeferredSenders = - std::mem::take(&mut *self.inner.pending_senders.lock().unwrap()); - if !senders.is_empty() { if let Some(tx) = self.inner.completion_task_tx.as_ref() { - // Snapshot total_completed_steps AFTER flush. The - // data will land in the next step (>completed), so - // this value is the correct lower bound for both - // fast mode (fire when completed_steps > this) and - // strict mode (fire when checkpointed_steps > this, - // per the `total_checkpointed_steps >= n` semantics). + // The data we just flushed lands in the next step, so + // remember how many steps have completed right now. + // The watcher answers these acks once the frontier + // passes this value: after the step itself in fast + // mode, after its checkpoint in strict mode. let step_at_flush = self .inner .step_completion_rx .as_ref() .map(|rx| rx.borrow().total_completed_steps) .unwrap_or(0); - let _ = tx.send((step_at_flush, senders)); + let _ = tx.send(CompletionMessage::Queued(PendingAcks { + step_at_flush, + held_since: Instant::now(), + senders, + })); } else { - // No completion tracking — fire immediately. + // No completion tracking: complete immediately. for sender in senders { - sender.send(Ok(())); + sender.complete(); } } } + + // etl blocks on each write's ack before it can proceed, so an + // ack stuck in the queue stalls replication. That can happen: + // a flush may stop at `max_batch_size` before reaching the + // ack's entry. If the leftover tail is smaller than + // `min_batch_size_records`, the controller may wait until the + // buffer timeout before scheduling another step, stalling + // replication in the meantime. So while any ack is still queued, + // keep requesting a step. + if self.inner.queued_acks.load(Ordering::Acquire) > 0 { + self.inner.consumer.request_step(); + } + } + NonFtInputReaderCommand::Transition(state) => { + if state == PipelineState::Terminated { + self.inner.shutdown_etl_pipeline(); + } + let _ = self.sender.send_replace(state); } - NonFtInputReaderCommand::Transition(state) => drop(self.sender.send_replace(state)), } } @@ -236,24 +498,26 @@ struct PostgresCdcInputInner { endpoint_name: String, config: PostgresCdcReaderConfig, consumer: Box, - queue: Arc, + queue: Arc, /// Deterministic pipeline ID used for replication slot naming and resume. pipeline_id: u64, - /// Deferred async result senders from `write_events`, waiting to be paired - /// with a step number during the next `Queue` command. - pending_senders: Arc>, - /// Watch receiver for step completion — used to snapshot `step_at_flush` - /// in the Queue handler. Always tracks `total_completed_steps`. + /// How many `DeferredAck`s are sitting in `queue`, waiting to be flushed. + /// While nonzero, the reader keeps requesting bounded steps; otherwise an + /// ack behind a tail smaller than `min_batch_size_records` could sit there + /// forever with etl blocked on it. + queued_acks: Arc, + /// Watch receiver for step completion, used to capture `step_at_flush` in + /// the Queue handler. Always tracks `total_completed_steps`. step_completion_rx: Option>, - /// Watcher source for the background task. Taken once by `worker_task_inner`. + /// Watcher source for the background task. Taken once by `worker_task_inner`. /// `Strict` when fault tolerance is enabled (gates slot on checkpoint); /// `Fast` otherwise (gates slot on step completion). watcher_rx: Mutex>, - /// Sender for passing (step_at_flush, senders) to the background task. - /// Created once at construction time if completion tracking is available. - completion_task_tx: Option>, + /// Sender for passing pending acks to the background task. + /// Created at construction time if completion tracking is available. + completion_task_tx: Option>, /// Receiver half, taken once by worker_task_inner to spawn the background task. - completion_task_rx: Mutex>>, + completion_task_rx: Mutex>>, /// etl shutdown handle for the currently running pipeline. /// Used to stop etl workers when Feldera terminates the connector. etl_shutdown_tx: Mutex>, @@ -290,7 +554,7 @@ impl PostgresCdcInputInner { consumer, queue, pipeline_id, - pending_senders: Arc::new(Mutex::new(Vec::new())), + queued_acks: Arc::new(AtomicUsize::new(0)), step_completion_rx, watcher_rx: Mutex::new(watcher_rx), completion_task_tx, @@ -352,14 +616,16 @@ impl PostgresCdcInputInner { max_table_sync_workers: PipelineConfig::DEFAULT_MAX_TABLE_SYNC_WORKERS, max_copy_connections_per_table: PipelineConfig::DEFAULT_MAX_COPY_CONNECTIONS_PER_TABLE, memory_refresh_interval_ms: PipelineConfig::DEFAULT_MEMORY_REFRESH_INTERVAL_MS, + replication_lag_refresh_interval_ms: + PipelineConfig::DEFAULT_REPLICATION_LAG_REFRESH_INTERVAL_MS, memory_backpressure: Some(MemoryBackpressureConfig::default()), table_sync_copy: TableSyncCopyConfig::IncludeAllTables, invalidated_slot_behavior: InvalidatedSlotBehavior::default(), + run_source_migrations: true, }; - // Use PostgresStore to persist table replication phases across restarts. - // This allows etl to resume from the replication slot position instead of - // re-snapshotting the entire table on restart. + // Persist table phases and slot progress across restarts. Once a table + // copy completes, this avoids repeating it on an ordinary restart. let store = match PostgresStore::new(self.pipeline_id, pg_conn).await { Ok(store) => store, Err(e) => { @@ -372,11 +638,17 @@ impl PostgresCdcInputInner { } }; - let pending_senders = if self.step_completion_rx.is_some() { - Some(Arc::clone(&self.pending_senders)) + let discard_errors_result = if self.config.discard_table_errors { + self.discard_table_errors(&store).await + } else if self.config.discard_shutdown_errors { + self.discard_shutdown_errors(&store).await } else { - None + Ok(()) }; + if let Err(e) = discard_errors_result { + let _ = init_status_sender.send(Err(e)); + return; + } let destination = FelderaDestination { input_stream: Arc::new(Mutex::new(input_stream)), @@ -384,7 +656,8 @@ impl PostgresCdcInputInner { source_table: self.config.source_table.clone(), endpoint_name: self.endpoint_name.clone(), feldera_required_columns, - pending_senders, + completion_task_tx: self.completion_task_tx.clone(), + queued_acks: Arc::clone(&self.queued_acks), pipeline_state_rx: receiver.clone(), }; @@ -426,6 +699,7 @@ impl PostgresCdcInputInner { watcher, rx, self.endpoint_name.clone(), + Duration::from_millis(self.config.streaming_ack_hold_ms), ))), _ => None, }; @@ -436,8 +710,7 @@ impl PostgresCdcInputInner { // forever while the input silently stalls; the watcher reports such an // error so the controller fails the endpoint instead. let mut receiver_clone = receiver.clone(); - let pipeline_wait = pipeline.wait(); - tokio::pin!(pipeline_wait); + let mut pipeline_wait = Box::pin(pipeline.wait()); let (pipeline_result, report_error) = select! { result = &mut pipeline_wait => (result, true), _ = receiver_clone.wait_for(|state| state == &PipelineState::Terminated) => { @@ -485,6 +758,152 @@ impl PostgresCdcInputInner { let _ = shutdown_tx.shutdown(); } } + + /// Roll back persisted errors caused by a destination acknowledgment being + /// dropped by a previous connector run. + async fn discard_shutdown_errors(&self, store: &PostgresStore) -> Result<(), ControllerError> { + self.discard_matching_table_errors(store, false).await + } + + /// Roll back every persisted `Errored` table state so etl retries those + /// tables on this run (the `discard_table_errors` config option). + async fn discard_table_errors(&self, store: &PostgresStore) -> Result<(), ControllerError> { + self.discard_matching_table_errors(store, true).await + } + + /// Roll back matching persisted `Errored` table states. + /// + /// Errored states can stack, so keep rolling back until something else + /// surfaces. If a rollback fails, reset the table to `Init` instead, + /// which re-copies it from scratch. + async fn discard_matching_table_errors( + &self, + store: &PostgresStore, + discard_all: bool, + ) -> Result<(), ControllerError> { + store.load_table_states().await.map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to load etl table states before discarding errors: {e}"), + ) + })?; + + let states = store.get_table_states().await.map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to read etl table states before discarding errors: {e}"), + ) + })?; + + let errored_table_ids: Vec<_> = states + .iter() + .filter_map(|(table_id, state)| { + should_discard_table_error(state, discard_all).then_some(*table_id) + }) + .collect(); + + if errored_table_ids.is_empty() { + return Ok(()); + } + + warn!( + "postgres_cdc {}: discarding {} persisted etl {} error(s) before startup", + &self.endpoint_name, + errored_table_ids.len(), + if discard_all { "table" } else { "shutdown" }, + ); + + for table_id in errored_table_ids { + let mut discarded = 0usize; + loop { + let Some(state) = store.get_table_state(table_id).await.map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to read etl table state for table {table_id}: {e}"), + ) + })? + else { + break; + }; + if !should_discard_table_error(&state, discard_all) { + break; + } + + if discarded == MAX_ERROR_ROLLBACKS_PER_TABLE { + warn!( + "postgres_cdc {}: table {table_id} still has a matching etl error after \ + {MAX_ERROR_ROLLBACKS_PER_TABLE} rollbacks; resetting table state to init", + &self.endpoint_name + ); + store + .update_table_state(table_id, TableState::Init) + .await + .map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!( + "failed to reset etl table state for table {table_id} after \ + reaching the rollback limit: {e}" + ), + ) + })?; + break; + } + + match store.rollback_table_state(table_id).await { + Ok(restored_state) => { + discarded += 1; + info!( + "postgres_cdc {}: discarded etl table error for table {table_id}, \ + restored previous state {restored_state}", + &self.endpoint_name + ); + } + Err(e) => { + warn!( + "postgres_cdc {}: failed to roll back etl table error for table \ + {table_id}: {e}; resetting table state to init", + &self.endpoint_name + ); + store + .update_table_state(table_id, TableState::Init) + .await + .map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!( + "failed to reset etl table state for table {table_id} \ + after discarding error: {e}" + ), + ) + })?; + discarded += 1; + } + } + } + + debug!( + "postgres_cdc {}: discarded {discarded} etl table error state(s) for table {table_id}", + &self.endpoint_name + ); + } + + Ok(()) + } +} + +fn should_discard_table_error(state: &TableState, discard_all: bool) -> bool { + match state { + TableState::Errored { reason, .. } => { + discard_all || reason == DROPPED_DESTINATION_ACK_ERROR + } + _ => false, + } } impl Drop for PostgresCdcInputInner { @@ -569,16 +988,19 @@ impl TableErrorMonitor { #[derive(Clone)] struct FelderaDestination { input_stream: Arc>>, - queue: Arc, + queue: Arc, source_table: String, endpoint_name: String, /// Canonical names of the non-nullable Feldera columns. Each must be present /// (by name) in the target Postgres table schema etl passes with each target /// batch/event. Nullable and extra columns need not match. feldera_required_columns: Vec, - /// Deferred async result senders. If `Some`, write_events stores senders here - /// instead of firing them immediately. The Queue handler picks them up. - pending_senders: Option>>, + /// Sends deferred acks to the completion watcher. When absent, writes are + /// acked immediately. + completion_task_tx: Option>, + /// How many `DeferredAck`s are sitting in the queue, waiting to be + /// flushed. See [`PostgresCdcInputInner::queued_acks`]. + queued_acks: Arc, /// Pipeline state receiver used to stop accepting new etl batches while the /// Feldera pipeline is paused. pipeline_state_rx: Receiver, @@ -612,7 +1034,7 @@ impl Destination for FelderaDestination { &self, replicated_table_schema: &ReplicatedTableSchema, table_rows: Vec, - async_result: WriteTableRowsResult<()>, + async_result: WriteTableRowsResult, ) -> EtlResult<()> { self.wait_unpaused().await?; @@ -620,7 +1042,7 @@ impl Destination for FelderaDestination { let column_names = match self.column_names_for_target_schema(replicated_table_schema)? { Some(columns) => columns, None => { - async_result.send(Ok(())); + async_result.send(Ok(DestinationWriteStatus::Durable)); return Ok(()); } }; @@ -628,6 +1050,7 @@ impl Destination for FelderaDestination { let mut stream = self.input_stream.lock().unwrap(); let mut bytes = 0; let mut errors = Vec::new(); + let mut queued_data = false; let timestamp = Utc::now(); for row in &table_rows { @@ -647,30 +1070,42 @@ impl Destination for FelderaDestination { bytes += json_str.len(); if bytes >= 2 * 1024 * 1024 { - self.queue.push((stream.take_all(), errors), timestamp); + self.queue.push_with_aux( + (stream.take_all(), errors), + timestamp, + DeferredSenders::new(), + ); + queued_data = true; bytes = 0; errors = Vec::new(); } } if bytes > 0 || !errors.is_empty() { - self.queue.push((stream.take_all(), errors), timestamp); + self.queue.push_with_aux( + (stream.take_all(), errors), + timestamp, + DeferredSenders::new(), + ); + queued_data = true; } - async_result.send(Ok(())); + self.ack_snapshot_copy(queued_data, async_result); Ok(()) } async fn write_events( &self, events: Vec, - async_result: WriteEventsResult<()>, + durability: WriteEventsDurability, + async_result: WriteEventsResult, ) -> EtlResult<()> { self.wait_unpaused().await?; let mut stream = self.input_stream.lock().unwrap(); let mut bytes = 0; let mut errors = Vec::new(); + let mut queue_entries = Vec::new(); let timestamp = Utc::now(); for event in &events { @@ -779,21 +1214,40 @@ impl Destination for FelderaDestination { } if bytes >= 2 * 1024 * 1024 { - self.queue.push((stream.take_all(), errors), timestamp); + queue_entries.push((stream.take_all(), errors)); bytes = 0; errors = Vec::new(); } } if bytes > 0 || !errors.is_empty() { - self.queue.push((stream.take_all(), errors), timestamp); + queue_entries.push((stream.take_all(), errors)); } - // Defer or fire the async result. - if let Some(ref pending) = self.pending_senders { - pending.lock().unwrap().push(async_result); + let ack = DeferredAck::Stream { + result: async_result, + durability, + }; + if let Some(last_entry) = queue_entries.pop() { + for entry in queue_entries { + self.queue + .push_with_aux(entry, timestamp, DeferredSenders::new()); + } + if self.completion_task_tx.is_some() { + // Increment before pushing so the count cannot underflow if + // the reader flushes and decrements before this lands. + self.queued_acks.fetch_add(1, Ordering::Release); + self.queue.push_with_aux(last_entry, timestamp, vec![ack]); + // etl waits for this ack before sending more events. Keep + // taking steps until the queue entry carrying it is flushed. + self.queue.consumer.request_step(); + } else { + self.queue + .push_with_aux(last_entry, timestamp, DeferredSenders::new()); + ack.complete(); + } } else { - async_result.send(Ok(())); + self.ack_no_row_stream_write(ack); } Ok(()) @@ -801,6 +1255,60 @@ impl Destination for FelderaDestination { } impl FelderaDestination { + /// Answer a table-copy write. + /// + /// Without completion tracking there is nothing to wait for, so every + /// write is `Durable` immediately. + /// + /// With tracking, a batch that queued rows is answered `Accepted` right + /// away: Feldera owns the rows, and etl keeps copying while they wait for + /// the completion frontier. + /// + /// The one empty write etl sends at the end of the copy (an empty table + /// sends only this) is the table's durability barrier: answering it + /// `Durable` tells etl the whole snapshot is safe. So instead of answering + /// now, queue it behind all the snapshot data and let the completion + /// watcher answer once its step passes the frontier. + fn ack_snapshot_copy(&self, queued_data: bool, async_result: WriteTableRowsResult) { + if self.completion_task_tx.is_none() { + async_result.send(Ok(DestinationWriteStatus::Durable)); + } else if queued_data { + async_result.send(Ok(DestinationWriteStatus::Accepted)); + } else { + // At most one barrier can be live at a time: only the configured + // source table's copy reaches this branch. Every other publication + // table fails the column match in `write_table_rows` and is acked + // Durable immediately. + // + // Increment before pushing so the count cannot underflow if the + // reader flushes and decrements before this increment lands. + self.queued_acks.fetch_add(1, Ordering::Release); + self.queue.push_with_aux( + (None, Vec::new()), + Utc::now(), + vec![DeferredAck::SnapshotCopyBarrier(async_result)], + ); + } + } + + /// Answer a stream write that produced no Feldera rows. + /// + /// The completion watcher answers `Durable` if the frontier already covers + /// all earlier `Accepted` data. Otherwise a `MayDefer` write is `Accepted`, + /// while a `RequireDurable` write waits for that frontier. + fn ack_no_row_stream_write(&self, ack: DeferredAck) { + if !ack.is_stream() { + unreachable!("snapshot copy barriers are handled by ack_snapshot_copy") + } else if let Some(tx) = self.completion_task_tx.as_ref() { + // A no-row write is etl's chance to hear that data it previously + // got an Accepted for has since become durable. + let _ = tx.send(CompletionMessage::NoRowStream(ack)); + } else { + // Without completion tracking, data writes are already Durable. + ack.complete(); + } + } + /// Wait until the Feldera pipeline is running before accepting a new etl /// batch. async fn wait_unpaused(&self) -> EtlResult<()> { @@ -948,6 +1456,7 @@ fn cell_to_json(cell: &Cell) -> Value { } Cell::Date(d) => json!(d.to_string()), Cell::Time(t) => json!(t.to_string()), + Cell::TimeTz(t) => json!(t.to_string()), Cell::Timestamp(ts) => json!(ts.format("%Y-%m-%dT%H:%M:%S%.f").to_string()), Cell::TimestampTz(ts) => json!(ts.to_rfc3339()), Cell::Uuid(u) => json!(u.to_string()), @@ -1021,6 +1530,16 @@ fn array_cell_to_json(arr: &ArrayCell) -> Value { .collect(); Value::Array(vals) } + ArrayCell::TimeTz(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(t) => json!(t.to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } ArrayCell::Timestamp(v) => { let vals: Vec = v .iter() @@ -1077,7 +1596,7 @@ fn array_cell_to_json(arr: &ArrayCell) -> Value { /// Typed watch receiver used by the completion watcher background task. /// /// `Fast` waits for step completion (`total_completed_steps`); used when fault -/// tolerance is not enabled. `Strict` waits for checkpoint completion; used +/// tolerance is not enabled. `Strict` waits for checkpoint completion; used /// when fault tolerance is enabled so the replication slot only advances past /// the last durable checkpoint, preserving at-least-once correctness for /// stateful circuits after a crash. @@ -1102,68 +1621,205 @@ impl WatcherReceiver { } } -/// Background task that fires deferred ETL async result senders when the -/// completion frontier passes the step recorded at Queue time. +/// Background task that answers deferred acks. /// -/// Each entry is `(step_at_flush, senders)` where `step_at_flush` is the -/// value of `total_completed_steps` at the time the data was flushed to the -/// circuit. The data lands in the next step, so we fire when the frontier -/// strictly exceeds `step_at_flush`. +/// Each [`PendingAcks`] remembers how many steps had completed when its data +/// was flushed. The data lands in the next step, so its acks are `Durable` once +/// the frontier moves strictly past `step_at_flush`. `MayDefer` stream acks that +/// wait longer than the hold deadline are answered `Accepted` instead, and a +/// later write confirms them once the frontier catches up. `RequireDurable` +/// acks always wait for the frontier. async fn completion_watcher_task( mut watcher: WatcherReceiver, - mut pending_rx: mpsc::UnboundedReceiver<(u64, DeferredSenders)>, + mut pending_rx: mpsc::UnboundedReceiver, endpoint_name: String, + streaming_ack_hold: Duration, ) { - let mut waiting: Vec<(u64, DeferredSenders)> = Vec::new(); + let mut waiting: Vec = Vec::new(); + // The latest step whose stream ack timed out as Accepted. etl applies + // events in one ordered stream, so a Durable answer for any later step + // also covers this one; forget it once that happens. + let mut accepted_stream_step = None; loop { + let next_stream_deadline = earliest_stream_deadline(&waiting, streaming_ack_hold); tokio::select! { result = watcher.changed() => { if result.is_err() { break; // Sender dropped (pipeline shutting down) } - let f = watcher.frontier(); - fire_completed(&mut waiting, f); + let frontier = watcher.frontier(); + if let Some(completed_step) = fire_completed(&mut waiting, frontier) + && accepted_stream_step.is_some_and(|step| step <= completed_step) + { + accepted_stream_step = None; + } } - maybe_entry = pending_rx.recv() => { - match maybe_entry { - Some((step_at_flush, senders)) => { - let f = watcher.frontier(); - if f > step_at_flush { - // Already past the threshold — fire immediately. - for sender in senders { - sender.send(Ok(())); + maybe_message = pending_rx.recv() => { + match maybe_message { + Some(CompletionMessage::Queued(entry)) => { + let frontier = watcher.frontier(); + if frontier > entry.step_at_flush { + // Already past the threshold: complete immediately. + let completes_stream = entry.senders.iter().any(DeferredAck::is_stream); + for sender in entry.senders { + sender.complete(); + } + if completes_stream + && accepted_stream_step + .is_some_and(|step| step <= entry.step_at_flush) + { + accepted_stream_step = None; } } else { - waiting.push((step_at_flush, senders)); + waiting.push(entry); } } + Some(CompletionMessage::NoRowStream(ack)) => { + complete_no_row_stream_ack( + ack, + watcher.frontier(), + &mut accepted_stream_step, + &mut waiting, + ); + } None => break, // Channel closed } } + _ = async { + match next_stream_deadline { + Some(deadline) => sleep_until(TokioInstant::from_std(deadline)).await, + None => pending::<()>().await, + } + } => { + debug!( + "postgres_cdc {endpoint_name}: stream ack hold expired, \ + accepting non-durable stream acks" + ); + if let Some(step) = accept_expired_stream_acks( + &mut waiting, + Instant::now(), + streaming_ack_hold, + ) { + accepted_stream_step = Some( + accepted_stream_step.map_or(step, |previous: u64| previous.max(step)), + ); + } + } } } - // On shutdown, remaining senders are dropped. AsyncResult's Drop impl - // sends an error to the ETL side, causing it to shut down gracefully. + // Dropping the remaining senders here is safe: the connector signals etl + // shutdown before aborting this task, so etl's copy loop sees the + // shutdown before it sees the dropped ack, and its table-sync error + // handler does not persist errors that surface after a shutdown request. debug!( "postgres_cdc {endpoint_name}: completion watcher exiting with {} pending entries", waiting.len() ); } -/// Fires deferred senders whose data has been fully processed. -fn fire_completed(waiting: &mut Vec<(u64, DeferredSenders)>, completed_steps: u64) { - waiting.retain_mut(|(step_at_flush, senders)| { - if completed_steps > *step_at_flush { - for sender in senders.drain(..) { - sender.send(Ok(())); +/// Answers `Durable` for every waiting ack whose step the frontier has +/// passed. Returns the latest step whose stream ack became `Durable`, so the +/// caller can clear `accepted_stream_step` when that Durable covers it. +fn fire_completed(waiting: &mut Vec, frontier: u64) -> Option { + let mut completed_stream_step = None; + waiting.retain_mut(|entry| { + if frontier > entry.step_at_flush { + for sender in entry.senders.drain(..) { + if sender.is_stream() { + completed_stream_step = Some( + completed_stream_step.map_or(entry.step_at_flush, |step: u64| { + step.max(entry.step_at_flush) + }), + ); + } + sender.complete(); } false } else { true } }); + completed_stream_step +} + +/// Answer a no-row stream write against the latest accepted prefix. +/// +/// If the prefix is already complete, answer `Durable`. Otherwise answer a +/// `MayDefer` write as `Accepted`, or retain a `RequireDurable` write until the +/// frontier passes the accepted step. +fn complete_no_row_stream_ack( + ack: DeferredAck, + frontier: u64, + accepted_stream_step: &mut Option, + waiting: &mut Vec, +) { + debug_assert!(ack.is_stream()); + let Some(accepted_step) = *accepted_stream_step else { + ack.complete(); + return; + }; + + if frontier > accepted_step { + *accepted_stream_step = None; + ack.complete(); + } else if ack.may_defer() { + ack.accept(); + } else { + waiting.push(PendingAcks { + step_at_flush: accepted_step, + held_since: Instant::now(), + senders: vec![ack], + }); + } +} + +/// The next moment a waiting stream ack's hold expires, if any. +fn earliest_stream_deadline( + waiting: &[PendingAcks], + streaming_ack_hold: Duration, +) -> Option { + waiting + .iter() + .filter(|entry| entry.senders.iter().any(DeferredAck::may_defer)) + .map(|entry| entry.held_since + streaming_ack_hold) + .min() +} + +/// Answers `Accepted` for deferrable stream acks whose hold has expired. +/// Snapshot barriers and required-durability stream acks stay queued. +/// Returns the latest step whose stream ack was accepted. +fn accept_expired_stream_acks( + waiting: &mut Vec, + now: Instant, + streaming_ack_hold: Duration, +) -> Option { + let mut accepted_stream_step = None; + waiting.retain_mut(|entry| { + if now >= entry.held_since + streaming_ack_hold { + let senders = std::mem::take(&mut entry.senders); + entry.senders = senders + .into_iter() + .filter_map(|sender| { + if sender.may_defer() { + accepted_stream_step = Some( + accepted_stream_step.map_or(entry.step_at_flush, |step: u64| { + step.max(entry.step_at_flush) + }), + ); + sender.accept(); + None + } else { + Some(sender) + } + }) + .collect(); + } + + !entry.senders.is_empty() + }); + accepted_stream_step } async fn abort_completion_watcher(handle: &mut Option>) { @@ -1244,10 +1900,197 @@ fn parse_pg_uri( mod tests { use super::*; use chrono::{NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc}; - use etl::types::PgNumeric; + use etl::data::PgNumeric; use serde_json::json; use std::str::FromStr; + fn errored_table_state(reason: &str) -> TableState { + serde_json::from_value(json!({ + "type": "errored", + "reason": reason, + "solution": null, + "retry_policy": { "type": "manual_retry" }, + })) + .unwrap() + } + + #[test] + fn discard_shutdown_errors_only_matches_dropped_destination_acks() { + let shutdown_error = errored_table_state(DROPPED_DESTINATION_ACK_ERROR); + let replication_error = errored_table_state("source schema changed"); + + assert!(should_discard_table_error(&shutdown_error, false)); + assert!(!should_discard_table_error(&replication_error, false)); + assert!(should_discard_table_error(&shutdown_error, true)); + assert!(should_discard_table_error(&replication_error, true)); + } + + fn test_ack(kind: TestAckKind, tx: std::sync::mpsc::Sender) -> DeferredAck { + DeferredAck::Test { kind, tx } + } + + fn test_required_stream_ack(tx: std::sync::mpsc::Sender) -> DeferredAck { + DeferredAck::Test { + kind: TestAckKind::StreamRequireDurable, + tx, + } + } + + #[test] + fn watcher_deadline_accepts_only_deferrable_stream_acks() { + let (tx, rx) = std::sync::mpsc::channel(); + let now = Instant::now(); + let hold = Duration::from_secs(2); + let mut waiting = vec![PendingAcks { + step_at_flush: 7, + held_since: now - Duration::from_secs(3), + senders: vec![ + test_ack(TestAckKind::StreamMayDefer, tx.clone()), + test_required_stream_ack(tx.clone()), + test_ack(TestAckKind::SnapshotCopyBarrier, tx), + ], + }]; + + let accepted_step = accept_expired_stream_acks(&mut waiting, now, hold); + + assert_eq!(accepted_step, Some(7)); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Accepted); + assert!(rx.try_recv().is_err()); + assert_eq!(waiting.len(), 1); + assert_eq!(waiting[0].senders.len(), 2); + assert!(waiting[0].senders.iter().all(|ack| !ack.may_defer())); + + assert_eq!(fire_completed(&mut waiting, 8), Some(7)); + + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + assert!(waiting.is_empty()); + } + + #[test] + fn watcher_frontier_completes_all_acks_sharing_a_step() { + let (tx, rx) = std::sync::mpsc::channel(); + let now = Instant::now(); + let mut waiting = vec![PendingAcks { + step_at_flush: 7, + held_since: now, + senders: vec![ + test_ack(TestAckKind::StreamMayDefer, tx.clone()), + test_ack(TestAckKind::SnapshotCopyBarrier, tx), + ], + }]; + + assert_eq!(fire_completed(&mut waiting, 7), None); + assert!(rx.try_recv().is_err()); + assert_eq!(waiting.len(), 1); + assert_eq!(waiting[0].senders.len(), 2); + + assert_eq!(fire_completed(&mut waiting, 8), Some(7)); + + let mut statuses = vec![rx.recv().unwrap(), rx.recv().unwrap()]; + statuses.sort_by_key(|status| match status { + TestAckStatus::Accepted => 0, + TestAckStatus::Durable => 1, + }); + assert_eq!( + statuses, + vec![TestAckStatus::Durable, TestAckStatus::Durable] + ); + assert!(waiting.is_empty()); + } + + #[test] + fn no_row_stream_ack_makes_checkpointed_accepted_data_durable() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut accepted_stream_step = Some(7); + let mut waiting = Vec::new(); + + complete_no_row_stream_ack( + test_ack(TestAckKind::StreamMayDefer, tx.clone()), + 7, + &mut accepted_stream_step, + &mut waiting, + ); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Accepted); + assert_eq!(accepted_stream_step, Some(7)); + assert!(waiting.is_empty()); + + complete_no_row_stream_ack( + test_ack(TestAckKind::StreamMayDefer, tx), + 8, + &mut accepted_stream_step, + &mut waiting, + ); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + assert_eq!(accepted_stream_step, None); + assert!(waiting.is_empty()); + } + + #[test] + fn no_row_stream_ack_is_durable_without_accepted_data() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut accepted_stream_step = None; + let mut waiting = Vec::new(); + + complete_no_row_stream_ack( + test_ack(TestAckKind::StreamMayDefer, tx), + 0, + &mut accepted_stream_step, + &mut waiting, + ); + + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + assert_eq!(accepted_stream_step, None); + assert!(waiting.is_empty()); + } + + #[test] + fn required_no_row_stream_ack_waits_for_accepted_prefix() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut accepted_stream_step = Some(7); + let mut waiting = Vec::new(); + + complete_no_row_stream_ack( + test_required_stream_ack(tx), + 7, + &mut accepted_stream_step, + &mut waiting, + ); + + assert!(rx.try_recv().is_err()); + assert_eq!(accepted_stream_step, Some(7)); + assert_eq!(waiting.len(), 1); + assert!(!waiting[0].senders[0].may_defer()); + + assert_eq!(fire_completed(&mut waiting, 8), Some(7)); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + assert!(waiting.is_empty()); + } + + #[test] + fn snapshot_copy_barrier_completes_durable_only() { + // The copy barrier must never be reported as merely accepted: etl fails + // a table copy whose terminal barrier does not confirm durability. + let (tx, rx) = std::sync::mpsc::channel(); + DeferredAck::Test { + kind: TestAckKind::SnapshotCopyBarrier, + tx: tx.clone(), + } + .complete(); + assert_eq!(rx.recv().unwrap(), TestAckStatus::Durable); + } + + #[test] + #[should_panic(expected = "never accepted")] + fn snapshot_copy_barrier_never_accepts() { + let (tx, _rx) = std::sync::mpsc::channel(); + DeferredAck::Test { + kind: TestAckKind::SnapshotCopyBarrier, + tx, + } + .accept(); + } + // ----------------------------------------------------------------------- // cell_to_json unit tests // ----------------------------------------------------------------------- diff --git a/crates/adapters/src/integrated/postgres/test.rs b/crates/adapters/src/integrated/postgres/test.rs index 533cf65759c..7810085a94f 100644 --- a/crates/adapters/src/integrated/postgres/test.rs +++ b/crates/adapters/src/integrated/postgres/test.rs @@ -2400,11 +2400,12 @@ mod cdc_tests { /// Helper: creates a table, publication, and sets REPLICA IDENTITY FULL. /// Returns a connected client for further DML operations. - /// On drop, cleans up the publication and table. + /// On drop, cleans up the publication and its test tables. struct CdcTestTable { client: postgres::Client, table_name: String, publication_name: String, + extra_tables: Vec, url: String, } @@ -2461,6 +2462,7 @@ mod cdc_tests { client, table_name: table_name.to_string(), publication_name: publication_name.to_string(), + extra_tables: Vec::new(), url: url.to_string(), } } @@ -2520,6 +2522,7 @@ mod cdc_tests { client, table_name: table_name.to_string(), publication_name: publication_name.to_string(), + extra_tables: Vec::new(), url: url.to_string(), } } @@ -2538,6 +2541,11 @@ mod cdc_tests { &format!("DROP PUBLICATION IF EXISTS {}", self.publication_name), &[], ); + for table_name in &self.extra_tables { + let _ = self + .client + .execute(&format!("DROP TABLE IF EXISTS {table_name}"), &[]); + } let _ = self .client .execute(&format!("DROP TABLE IF EXISTS {}", self.table_name), &[]); @@ -2545,6 +2553,28 @@ mod cdc_tests { } impl CdcTestTable { + fn add_empty_table_to_publication(&mut self, table_name: &str) { + let _ = self + .client + .execute(&format!("DROP TABLE IF EXISTS {table_name}"), &[]); + self.client + .execute( + &format!("CREATE TABLE {table_name} (id INTEGER PRIMARY KEY)"), + &[], + ) + .expect("failed to create auxiliary CDC test table"); + self.client + .execute( + &format!( + "ALTER PUBLICATION {} ADD TABLE {table_name}", + self.publication_name + ), + &[], + ) + .expect("failed to add auxiliary table to publication"); + self.extra_tables.push(table_name.to_string()); + } + /// Drop the replication slots etl created for this pipeline. /// /// etl names slots after the *pipeline ID* (e.g. @@ -2606,6 +2636,48 @@ mod cdc_tests { std::thread::sleep(std::time::Duration::from_millis(250)); } } + + /// Wait until every replication slot this pipeline owns is inactive (no + /// attached backend), so a restart can re-acquire the same slot. + /// + /// Postgres releases a slot asynchronously after the owning pipeline + /// stops, so a fixed sleep races that release: the next run can fail + /// with "replication slot is active for PID ...". Polling `active_pid` + /// removes the race. + fn wait_replication_slots_inactive(&mut self) { + let connector_url = cdc_connector_url(&self.url); + let source_table = format!("public.{}", self.table_name); + let pipeline_id = crate::integrated::postgres::cdc_input::pipeline_id( + &connector_url, + &self.publication_name, + &source_table, + ) + .to_string(); + let owns_slot = |name: &str| name.split('_').any(|token| token == pipeline_id); + + // ~30 s budget (300 * 100 ms); release normally takes well under 1 s. + for _attempt in 0..300 { + let active: Vec = self + .client + .query( + "SELECT slot_name FROM pg_replication_slots WHERE active_pid IS NOT NULL", + &[], + ) + .unwrap_or_default() + .into_iter() + .map(|r| r.get::<_, String>("slot_name")) + .filter(|name| owns_slot(name)) + .collect(); + + if active.is_empty() { + return; + } + + std::thread::sleep(std::time::Duration::from_millis(100)); + } + + panic!("timeout waiting for this pipeline's replication slots to become inactive"); + } } fn cdc_connector_url(url: &str) -> String { @@ -2956,17 +3028,83 @@ mod cdc_tests { (controller, err_receiver) } - /// Like `cdc_simple_test_circuit` but with fault tolerance enabled, - /// file storage at `storage_dir`, and a 3600-second checkpoint interval so - /// that no automatic checkpoint fires during the test. With fault - /// tolerance enabled the connector uses strict mode: the replication slot - /// only advances after a durable checkpoint. + /// Like `cdc_simple_test_circuit` but with fault tolerance enabled, file + /// storage at `storage_dir`, and a 3600-second checkpoint interval so no + /// automatic checkpoint fires during the test. Under fault tolerance the + /// connector runs in strict mode: the replication slot only advances after + /// a durable checkpoint. fn cdc_ft_test_circuit( url: &str, publication: &str, source_table: &str, storage_dir: &Path, output_path: &Path, + ) -> (Controller, crossbeam::channel::Receiver) { + cdc_ft_test_circuit_with_options( + url, + publication, + source_table, + storage_dir, + output_path, + CdcFtTestOptions::default(), + ) + } + + #[derive(Clone, Copy)] + struct CdcFtTestOptions { + streaming_ack_hold_ms: u64, + discard_shutdown_errors: bool, + discard_table_errors: bool, + max_batch_size: Option, + min_batch_size_records: u64, + max_buffering_delay_usecs: u64, + send_snapshot: bool, + } + + impl Default for CdcFtTestOptions { + fn default() -> Self { + Self { + streaming_ack_hold_ms: + feldera_types::transport::postgres::default_streaming_ack_hold_ms(), + discard_shutdown_errors: + feldera_types::transport::postgres::default_discard_shutdown_errors(), + discard_table_errors: false, + max_batch_size: None, + min_batch_size_records: 0, + max_buffering_delay_usecs: 0, + send_snapshot: false, + } + } + } + + fn cdc_ft_test_circuit_with_streaming_ack_hold( + url: &str, + publication: &str, + source_table: &str, + storage_dir: &Path, + output_path: &Path, + streaming_ack_hold_ms: u64, + ) -> (Controller, crossbeam::channel::Receiver) { + cdc_ft_test_circuit_with_options( + url, + publication, + source_table, + storage_dir, + output_path, + CdcFtTestOptions { + streaming_ack_hold_ms, + ..Default::default() + }, + ) + } + + fn cdc_ft_test_circuit_with_options( + url: &str, + publication: &str, + source_table: &str, + storage_dir: &Path, + output_path: &Path, + options: CdcFtTestOptions, ) -> (Controller, crossbeam::channel::Receiver) { let url = cdc_connector_url(url); let schema = TestStruct::schema(); @@ -2976,15 +3114,21 @@ mod cdc_tests { "storage_config": { "path": storage_dir }, "storage": true, "fault_tolerance": { "model": "at_least_once", "checkpoint_interval_secs": 3600 }, + "min_batch_size_records": options.min_batch_size_records, + "max_buffering_delay_usecs": options.max_buffering_delay_usecs, "inputs": { "cdc_in": { "stream": "test_input1", + "max_batch_size": options.max_batch_size, "transport": { "name": "postgres_cdc_input", "config": { "uri": url, "publication": publication, "source_table": source_table, + "streaming_ack_hold_ms": options.streaming_ack_hold_ms, + "discard_shutdown_errors": options.discard_shutdown_errors, + "discard_table_errors": options.discard_table_errors, }, }, }, @@ -2992,6 +3136,7 @@ mod cdc_tests { "outputs": { "test_output1": { "stream": "test_output1", + "send_snapshot": options.send_snapshot, "transport": { "name": "file_output", "config": { "path": output_path }, @@ -3012,6 +3157,7 @@ mod cdc_tests { let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| { let mut catalog = Catalog::new(); let (input, hinput) = circuit.add_input_zset::(); + input.set_persistent_mir_id("input"); let input_schema = serde_json::to_string(&Relation::new( "test_input1".into(), schema.clone(), @@ -3031,7 +3177,8 @@ mod cdc_tests { hinput, &input_schema, ); - catalog.register_materialized_output_zset::<_, TestStruct>( + catalog.register_materialized_output_zset_persistent::<_, TestStruct>( + Some("test_output1"), input, &output_schema, ); @@ -3092,7 +3239,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_basic_insert() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_basic_insert"); @@ -3174,7 +3320,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_pause_unpause() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_pause_unpause"); @@ -3216,7 +3361,7 @@ mod cdc_tests { .is_input_endpoint_paused("cdc_in") .unwrap_or(false) }, - 10_000, + 30_000, ) .expect("timeout: CDC input endpoint did not report paused"); @@ -3332,7 +3477,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_all_data_types() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_all_types"); @@ -3477,7 +3621,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege, REPLICA IDENTITY FULL. #[test] #[serial] - #[ignore] fn test_cdc_update_delete() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_upd_del"); @@ -3596,7 +3739,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_compatible_schema_changes() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_compatible_schema"); @@ -3688,7 +3830,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_drop_primary_key_column_rejected() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_drop_pk_col"); @@ -3752,7 +3893,6 @@ mod cdc_tests { /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] fn test_cdc_restart_resumes_from_slot() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_restart"); @@ -3806,11 +3946,34 @@ mod cdc_tests { ) .expect("timeout: first run did not receive streamed row"); + // Wait until etl has finished catchup, then force one more steady-state + // transaction if needed so the apply worker advances `sync_done` to + // `ready` before we stop. Stopping in `sync_done` is legal, but a + // restart can replay a row event whose relation message was already + // skipped in the previous run. + let oid = table_oid(&mut table, &format!("public.{table_name}")); + wait(|| table_sync_done_or_ready(&mut table, oid), 60_000) + .expect("timeout: table did not reach sync_done/ready in run 1"); + if !table_ready(&mut table, oid) { + table.execute(&format!( + "INSERT INTO {table_name} VALUES (20, true, 7, 'ready_anchor')" + )); + wait( + || { + let rows = read_output_json(&output_path_1); + has_insert_id(&rows, 20) || !err_receiver_1.is_empty() + }, + 60_000, + ) + .expect("timeout: ready anchor row did not reach output"); + wait(|| table_ready(&mut table, oid), 60_000) + .expect("timeout: table did not reach ready in run 1"); + } + // Stop the first pipeline. controller_1.stop().unwrap(); - // Small delay to let the replication slot become inactive. - std::thread::sleep(std::time::Duration::from_secs(2)); + table.wait_replication_slots_inactive(); // --- Second pipeline run (restart) --- let output_file_2 = NamedTempFile::new().unwrap(); @@ -3880,17 +4043,13 @@ mod cdc_tests { // Fault-tolerance / strict-mode tests // ------------------------------------------------------------------- - /// With fault tolerance enabled the replication slot LSN is not - /// advanced until a Feldera checkpoint completes. When the pipeline is - /// stopped before any checkpoint occurs and then restarted with the same - /// connection identity (same slot), Postgres replays all events from the - /// original slot position — guaranteeing at-least-once delivery. + /// An uncheckpointed snapshot is retried after a clean stop without + /// poisoning etl's persisted table state. /// /// Requires: wal_level=logical, user with REPLICATION privilege. #[test] #[serial] - #[ignore] - fn test_cdc_ft_mode_holds_slot() { + fn test_cdc_ft_uncheckpointed_snapshot_retries_cleanly() { let url = postgres_url(); let table_name = unique_pg_name("cdc_test_strict_hold"); let publication = unique_pg_name("cdc_pub_strict_hold"); @@ -3926,11 +4085,22 @@ mod cdc_tests { errs_1.try_recv() ); - // Stop without a checkpoint — slot LSN is still at the snapshot position. + // Stop while the terminal copy barrier is still waiting for a checkpoint. ctrl_1.stop().unwrap(); - std::thread::sleep(std::time::Duration::from_secs(1)); + table.wait_replication_slots_inactive(); + + // A clean stop must not poison etl's stored table state. With no + // checkpoint, etl's terminal table-copy durability barrier is still + // waiting at stop; if the connector drops it as an error, etl would + // persist an `errored` state that it never retries across restarts, so + // run 2 would fail (`TableErrorMonitor`) or stall. + let oid = table_oid(&mut table, &format!("public.{table_name}")); + assert!( + !errored_state_persisted(&mut table, oid), + "etl persisted an errored table state after a clean stop" + ); - // --- Run 2: fresh output, same slot (same url/publication/source_table) --- + // --- Run 2: fresh Feldera state, same etl copy state --- let out_2 = NamedTempFile::new().unwrap(); let storage_2 = tempfile::tempdir().unwrap(); let (ctrl_2, errs_2) = cdc_ft_test_circuit( @@ -3968,17 +4138,928 @@ mod cdc_tests { }) .collect(); - // The slot was held back in run 1 — rows 1 and 2 must be redelivered. + // The copy never became durable in run 1, so rows 1 and 2 must be + // delivered by a retry rather than skipped as copy-complete. assert!( ids.contains(&1), - "row 1 must be redelivered (slot held); ids={ids:?}" + "row 1 must be redelivered by the snapshot retry; ids={ids:?}" ); assert!( ids.contains(&2), - "row 2 must be redelivered (slot held); ids={ids:?}" + "row 2 must be redelivered by the snapshot retry; ids={ids:?}" ); assert!(ids.contains(&3), "row 3 (new) must appear; ids={ids:?}"); ctrl_2.stop().unwrap(); } + + /// Accepted streaming acks should let later WAL batches reach the circuit + /// before the next checkpoint. Once a checkpoint covers those batches, a + /// no-row write must cumulatively advance the replication slot. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_ft_streaming_ack_latency_and_slot_advance() { + const STREAMING_ACK_HOLD_MS: u64 = 5_000; + const STREAMING_ACK_OBSERVATION_MS: u64 = STREAMING_ACK_HOLD_MS + STREAMING_ACK_HOLD_MS / 2; + + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_ft_stream_accept"); + let publication = unique_pg_name("cdc_pub_ft_stream_accept"); + let auxiliary_table = unique_pg_name("cdc_test_ft_stream_aux"); + let source_table = format!("public.{table_name}"); + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + table.add_empty_table_to_publication(&auxiliary_table); + let source_table_oid = table_oid(&mut table, &source_table); + let auxiliary_table_oid = table_oid(&mut table, &format!("public.{auxiliary_table}")); + + let storage = tempfile::tempdir().unwrap(); + let output = NamedTempFile::new().unwrap(); + let (controller, errors) = cdc_ft_test_circuit_with_streaming_ack_hold( + &url, + &publication, + &source_table, + storage.path(), + output.path(), + STREAMING_ACK_HOLD_MS, + ); + controller.start(); + + checkpoint_initial_sync_and_wait_ready( + &controller, + &errors, + &mut table, + source_table_oid, + output.path(), + &table_name, + 900_001, + ); + + let mut auxiliary_ready = false; + for _ in 0..12 { + if table_sync_done_or_ready(&mut table, auxiliary_table_oid) { + auxiliary_ready = true; + break; + } + controller + .checkpoint() + .expect("auxiliary table checkpoint failed"); + let _ = wait( + || table_sync_done_or_ready(&mut table, auxiliary_table_oid), + 5_000, + ); + } + assert!( + auxiliary_ready || table_sync_done_or_ready(&mut table, auxiliary_table_oid), + "auxiliary publication table did not finish bootstrap" + ); + + wait( + || confirmed_flush_lsn(&mut table, &source_table).is_some(), + 60_000, + ) + .expect("timeout: apply slot did not appear"); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'first')" + )); + let first_row_lsn = current_wal_lsn(&mut table); + wait( + || has_insert_id(&read_output_json(output.path()), 1) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: first streamed row did not reach output"); + + // The first row's stream ack is accepted after the hold deadline. On + // the old durable-only path the second insert below would remain stuck + // behind row 1 until a checkpoint. + std::thread::sleep(std::time::Duration::from_millis( + STREAMING_ACK_OBSERVATION_MS, + )); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 42, 'second')" + )); + wait( + || has_insert_id(&read_output_json(output.path()), 2) || !errors.is_empty(), + 30_000, + ) + .expect("timeout: second streamed row was gated by checkpoint"); + assert!( + errors.is_empty(), + "unexpected errors while streaming: {:?}", + errors.try_recv() + ); + + assert!( + confirmed_flush_lsn_is_before(&mut table, &source_table, &first_row_lsn), + "Accepted advanced the replication slot past the first streamed row" + ); + + // Let row 2 time out as Accepted, then checkpoint it. The checkpoint + // cannot update an acknowledgment that ETL has already consumed. + std::thread::sleep(std::time::Duration::from_millis( + STREAMING_ACK_OBSERVATION_MS, + )); + controller.checkpoint().expect("stream checkpoint failed"); + assert!( + confirmed_flush_lsn_is_before(&mut table, &source_table, &first_row_lsn), + "checkpoint unexpectedly advanced an already-Accepted write" + ); + + // This produces a write_events call but no Feldera rows. Its cumulative + // Durable result must cover the now-checkpointed target-table rows. + table.execute(&format!("INSERT INTO {auxiliary_table} VALUES (1)")); + wait( + || !confirmed_flush_lsn_is_before(&mut table, &source_table, &first_row_lsn), + 30_000, + ) + .expect("apply slot did not advance after the checkpoint and no-row write"); + + controller.stop().unwrap(); + } + + /// A steady-state stream ack queued behind a sub-minimum buffered tail must + /// still flush so replication can keep advancing. + /// + /// The ack rides the final data entry of an etl write. If the buffered tail + /// is below `min_batch_size_records` and the buffering delay is huge, the + /// connector must request bounded steps so that final entry is not stranded. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_ft_stream_ack_flushes_below_min_batch() { + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_ft_stream_ack_min_batch"); + let publication = unique_pg_name("cdc_pub_ft_stream_ack_min_batch"); + let source_table = format!("public.{table_name}"); + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + let source_table_oid = table_oid(&mut table, &source_table); + + let storage = tempfile::tempdir().unwrap(); + let output = NamedTempFile::new().unwrap(); + let (controller, errors) = cdc_ft_test_circuit_with_options( + &url, + &publication, + &source_table, + storage.path(), + output.path(), + CdcFtTestOptions { + min_batch_size_records: 1_000_000, + max_buffering_delay_usecs: 3_600_000_000, + ..Default::default() + }, + ); + controller.start(); + + checkpoint_initial_sync_and_wait_ready( + &controller, + &errors, + &mut table, + source_table_oid, + output.path(), + &table_name, + 900_004, + ); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'first')" + )); + wait( + || has_insert_id(&read_output_json(output.path()), 1) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: streamed row did not flush below min batch"); + assert!( + errors.is_empty(), + "unexpected errors while flushing stream ack: {:?}", + errors.try_recv() + ); + + controller.stop().unwrap(); + } + + /// Accepted-but-not-durable streaming progress must replay after restart. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_ft_accepted_streaming_events_replay_after_restart() { + const STREAMING_ACK_HOLD_MS: u64 = 500; + const STREAMING_ACK_OBSERVATION_MS: u64 = STREAMING_ACK_HOLD_MS * 2; + + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_ft_stream_replay"); + let publication = unique_pg_name("cdc_pub_ft_stream_replay"); + let source_table = format!("public.{table_name}"); + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + let source_table_oid = table_oid(&mut table, &source_table); + + let storage = tempfile::tempdir().unwrap(); + let out_1 = NamedTempFile::new().unwrap(); + let (ctrl_1, errs_1) = cdc_ft_test_circuit_with_streaming_ack_hold( + &url, + &publication, + &source_table, + storage.path(), + out_1.path(), + STREAMING_ACK_HOLD_MS, + ); + ctrl_1.start(); + + checkpoint_initial_sync_and_wait_ready( + &ctrl_1, + &errs_1, + &mut table, + source_table_oid, + out_1.path(), + &table_name, + 900_002, + ); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'first')" + )); + wait( + || has_insert_id(&read_output_json(out_1.path()), 1) || !errs_1.is_empty(), + 60_000, + ) + .expect("timeout: run 1 did not receive row 1"); + std::thread::sleep(std::time::Duration::from_millis( + STREAMING_ACK_OBSERVATION_MS, + )); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 42, 'second')" + )); + wait( + || has_insert_id(&read_output_json(out_1.path()), 2) || !errs_1.is_empty(), + 30_000, + ) + .expect("timeout: run 1 did not receive row 2 after accepted ack"); + std::thread::sleep(std::time::Duration::from_millis( + STREAMING_ACK_OBSERVATION_MS, + )); + assert!( + errs_1.is_empty(), + "unexpected errors in run 1: {:?}", + errs_1.try_recv() + ); + + ctrl_1.stop().unwrap(); + table.wait_replication_slots_inactive(); + + let out_2 = NamedTempFile::new().unwrap(); + let (ctrl_2, errs_2) = cdc_ft_test_circuit_with_streaming_ack_hold( + &url, + &publication, + &source_table, + storage.path(), + out_2.path(), + STREAMING_ACK_HOLD_MS, + ); + ctrl_2.start(); + + wait( + || { + let rows = read_output_json(out_2.path()); + (has_insert_id(&rows, 1) && has_insert_id(&rows, 2)) || !errs_2.is_empty() + }, + 60_000, + ) + .expect("timeout: accepted rows were not redelivered after restart"); + assert!( + errs_2.is_empty(), + "unexpected errors in run 2: {:?}", + errs_2.try_recv() + ); + + ctrl_2.stop().unwrap(); + } + + /// `discard_table_errors` rolls back persisted etl table errors before + /// startup, so a table poisoned by a previous run can be retried without + /// manual SQL. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_discard_table_errors_on_startup() { + run_cdc_discard_error_on_startup( + "cdc_test_discard_errors", + "cdc_pub_discard_errors", + "forced test table error", + CdcFtTestOptions { + discard_table_errors: true, + ..Default::default() + }, + ); + } + + /// `discard_shutdown_errors` defaults to true and rolls back a persisted + /// dropped-ack error from a previous failed run. Since the write was never + /// reported durable, replication can resume from the previous state. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_discard_shutdown_errors_on_startup() { + run_cdc_discard_error_on_startup( + "cdc_test_discard_shutdown", + "cdc_pub_discard_shutdown", + "[DestinationError] Async result channel closed before sending", + CdcFtTestOptions::default(), + ); + } + + fn run_cdc_discard_error_on_startup( + table_prefix: &str, + publication_prefix: &str, + error_reason: &str, + restart_options: CdcFtTestOptions, + ) { + let url = postgres_url(); + let table_name = unique_pg_name(table_prefix); + let publication = unique_pg_name(publication_prefix); + let source_table = format!("public.{table_name}"); + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'snapshot')" + )); + let source_table_oid = table_oid(&mut table, &source_table); + + let storage = tempfile::tempdir().unwrap(); + let out_1 = NamedTempFile::new().unwrap(); + let (ctrl_1, errs_1) = cdc_ft_test_circuit( + &url, + &publication, + &source_table, + storage.path(), + out_1.path(), + ); + ctrl_1.start(); + + wait( + || has_insert_id(&read_output_json(out_1.path()), 1) || !errs_1.is_empty(), + 60_000, + ) + .expect("timeout: snapshot row did not reach output"); + assert!( + errs_1.is_empty(), + "unexpected errors before checkpoint: {:?}", + errs_1.try_recv() + ); + + checkpoint_initial_sync_and_wait_ready( + &ctrl_1, + &errs_1, + &mut table, + source_table_oid, + out_1.path(), + &table_name, + 900_003, + ); + ctrl_1.stop().unwrap(); + table.wait_replication_slots_inactive(); + + force_current_table_error(&mut table, &source_table, source_table_oid, error_reason); + assert!( + errored_state_persisted(&mut table, source_table_oid), + "test setup failed to persist forced etl table error" + ); + + let out_2 = NamedTempFile::new().unwrap(); + let (ctrl_2, errs_2) = cdc_ft_test_circuit_with_options( + &url, + &publication, + &source_table, + storage.path(), + out_2.path(), + restart_options, + ); + ctrl_2.start(); + + wait( + || !errored_state_persisted(&mut table, source_table_oid) || !errs_2.is_empty(), + 60_000, + ) + .expect("timeout: startup did not clear persisted etl table error"); + assert!( + errs_2.is_empty(), + "unexpected errors after discarding table error: {:?}", + errs_2.try_recv() + ); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 42, 'after-error')" + )); + wait( + || has_insert_id(&read_output_json(out_2.path()), 2) || !errs_2.is_empty(), + 60_000, + ) + .expect("timeout: restarted connector did not stream after discarding table error"); + assert!( + errs_2.is_empty(), + "unexpected errors after restart: {:?}", + errs_2.try_recv() + ); + + ctrl_2.stop().unwrap(); + } + + /// Nonterminal snapshot-catchup batches may be accepted after the stream + /// acknowledgment timeout, but the terminal batch must wait for a Feldera + /// checkpoint before etl completes catchup. + /// + /// etl can use an acknowledged catchup stream batch to persist + /// `sync_done`/`ready` for the table. If the connector reports that batch + /// as `Accepted` (not durable) before Feldera checkpoints it, a crash can + /// make etl skip re-copying while Feldera resumes from a checkpoint that + /// predates the catchup rows. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_ft_terminal_catchup_ack_waits_for_checkpoint() { + const STREAMING_ACK_HOLD_MS: u64 = 250; + const STREAMING_ACK_OBSERVATION_MS: u64 = STREAMING_ACK_HOLD_MS * 2; + const NEGATIVE_OBSERVATION_MS: u128 = (STREAMING_ACK_HOLD_MS as u128) * 40; + + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_ft_catchup_accept"); + let publication = unique_pg_name("cdc_pub_ft_catchup_accept"); + let source_table = format!("public.{table_name}"); + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'snapshot')" + )); + let source_table_oid = table_oid(&mut table, &source_table); + + let storage = tempfile::tempdir().unwrap(); + let output = NamedTempFile::new().unwrap(); + let (controller, errors) = cdc_ft_test_circuit_with_streaming_ack_hold( + &url, + &publication, + &source_table, + storage.path(), + output.path(), + STREAMING_ACK_HOLD_MS, + ); + controller.start(); + + wait( + || has_insert_id(&read_output_json(output.path()), 1) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: snapshot row did not reach output"); + assert!( + errors.is_empty(), + "unexpected errors before catchup insert: {:?}", + errors.try_recv() + ); + + let catchup_id = 1_000_001; + table.execute(&format!( + "INSERT INTO {table_name} VALUES ({catchup_id}, false, 42, 'catchup')" + )); + + // This checkpoint covers the accepted snapshot copy rows and releases + // etl's terminal table-copy durability barrier. The catchup row above + // was inserted before etl could persist copy completion, so etl must + // process it during catchup after this point. + controller.checkpoint().expect("snapshot checkpoint failed"); + + wait( + || has_insert_id(&read_output_json(output.path()), catchup_id) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: catchup row did not reach output"); + assert!( + errors.is_empty(), + "unexpected errors after catchup row: {:?}", + errors.try_recv() + ); + + std::thread::sleep(std::time::Duration::from_millis( + STREAMING_ACK_OBSERVATION_MS, + )); + let anchor_id = 1_000_002; + table.execute(&format!( + "INSERT INTO {table_name} VALUES ({anchor_id}, true, 43, 'catchup-anchor')" + )); + + // The first nonterminal catchup batch is MayDefer, so its timeout lets + // etl send the terminal batch. Receiving the terminal batch is safe; + // etl cannot complete catchup until Feldera durably acknowledges it. + wait( + || has_insert_id(&read_output_json(output.path()), anchor_id) || !errors.is_empty(), + NEGATIVE_OBSERVATION_MS, + ) + .expect("etl did not send the terminal catchup batch after the nonterminal timeout"); + assert!( + errors.is_empty(), + "unexpected errors before catchup checkpoint: {:?}", + errors.try_recv() + ); + + let advanced_before_checkpoint = wait( + || table_sync_done_or_ready(&mut table, source_table_oid) || !errors.is_empty(), + NEGATIVE_OBSERVATION_MS, + ) + .is_ok(); + assert!( + errors.is_empty(), + "unexpected errors before catchup checkpoint: {:?}", + errors.try_recv() + ); + // Bounded-window check: this can catch early advancement, not prove absence. + assert!( + !advanced_before_checkpoint, + "etl reached sync_done/ready before the terminal catchup batch was checkpointed" + ); + + controller.checkpoint().expect("checkpoint failed"); + wait( + || table_sync_done_or_ready(&mut table, source_table_oid) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: etl did not complete catchup after checkpoint"); + assert!( + errors.is_empty(), + "unexpected errors after catchup checkpoint: {:?}", + errors.try_recv() + ); + + controller.stop().unwrap(); + } + + /// Resolve the OID of `source_table` (e.g. `"public.foo"`) as an `i32`. + fn table_oid(table: &mut CdcTestTable, source_table: &str) -> i32 { + table + .client + .query_one("SELECT to_regclass($1)::oid::int4", &[&source_table]) + .expect("failed to resolve source table oid") + .get(0) + } + + /// True once etl has persisted a copy-complete replication state + /// (`finished_copy`, `sync_done`, or `ready`) for `source_table_oid`. + fn copy_state_persisted(table: &mut CdcTestTable, source_table_oid: i32) -> bool { + table + .client + .query_one( + "SELECT EXISTS ( + SELECT 1 + FROM etl.replication_state + WHERE table_id::int4 = $1 + AND state IN ( + 'finished_copy'::etl.table_state, + 'sync_done'::etl.table_state, + 'ready'::etl.table_state + ) + AND is_current = true + )", + &[&source_table_oid], + ) + .map(|row| row.get::<_, bool>(0)) + .unwrap_or(false) + } + + /// True once etl has persisted `ready` for `source_table_oid`. + fn table_ready(table: &mut CdcTestTable, source_table_oid: i32) -> bool { + table + .client + .query_one( + "SELECT EXISTS ( + SELECT 1 + FROM etl.replication_state + WHERE table_id::int4 = $1 + AND state = 'ready'::etl.table_state + AND is_current = true + )", + &[&source_table_oid], + ) + .map(|row| row.get::<_, bool>(0)) + .unwrap_or(false) + } + + fn table_sync_done_or_ready(table: &mut CdcTestTable, source_table_oid: i32) -> bool { + table + .client + .query_one( + "SELECT EXISTS ( + SELECT 1 + FROM etl.replication_state + WHERE table_id::int4 = $1 + AND state IN ('sync_done'::etl.table_state, 'ready'::etl.table_state) + AND is_current = true + )", + &[&source_table_oid], + ) + .map(|row| row.get::<_, bool>(0)) + .unwrap_or(false) + } + + fn checkpoint_initial_sync_and_wait_ready( + controller: &Controller, + errors: &crossbeam::channel::Receiver, + table: &mut CdcTestTable, + source_table_oid: i32, + output_path: &Path, + table_name: &str, + ready_anchor_id: i64, + ) { + let mut copy_complete = false; + for _ in 0..12 { + controller.checkpoint().expect("initial checkpoint failed"); + if wait( + || copy_state_persisted(table, source_table_oid) || !errors.is_empty(), + 5_000, + ) + .is_ok() + { + assert!( + errors.is_empty(), + "unexpected errors during initial sync: {:?}", + errors.try_recv() + ); + if copy_state_persisted(table, source_table_oid) { + copy_complete = true; + break; + } + } + } + assert!( + copy_complete, + "timeout: table did not finish initial sync after checkpoints" + ); + + if table_ready(table, source_table_oid) { + return; + } + + table.execute(&format!( + "INSERT INTO {table_name} VALUES ({ready_anchor_id}, true, 7, 'ready_anchor')" + )); + wait( + || has_insert_id(&read_output_json(output_path), ready_anchor_id) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: ready anchor row did not reach output"); + assert!( + errors.is_empty(), + "unexpected errors while waiting for ready anchor: {:?}", + errors.try_recv() + ); + + controller + .checkpoint() + .expect("ready anchor checkpoint failed"); + wait( + || table_ready(table, source_table_oid) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: table did not reach ready after initial checkpoint"); + assert!( + errors.is_empty(), + "unexpected errors while waiting for ready: {:?}", + errors.try_recv() + ); + } + + fn apply_slot_name(table: &CdcTestTable, source_table: &str) -> String { + let connector_url = cdc_connector_url(&table.url); + let pipeline_id = crate::integrated::postgres::cdc_input::pipeline_id( + &connector_url, + &table.publication_name, + source_table, + ); + + format!("supabase_etl_apply_{pipeline_id}") + } + + fn confirmed_flush_lsn(table: &mut CdcTestTable, source_table: &str) -> Option { + let slot_name = apply_slot_name(table, source_table); + table + .client + .query_opt( + "SELECT confirmed_flush_lsn::text FROM pg_replication_slots WHERE slot_name = $1", + &[&slot_name], + ) + .expect("failed to read replication slot") + .map(|row| row.get(0)) + } + + fn current_wal_lsn(table: &mut CdcTestTable) -> String { + table + .client + .query_one("SELECT pg_current_wal_lsn()::text", &[]) + .expect("failed to read current WAL LSN") + .get(0) + } + + fn confirmed_flush_lsn_is_before( + table: &mut CdcTestTable, + source_table: &str, + lsn: &str, + ) -> bool { + let slot_name = apply_slot_name(table, source_table); + table + .client + .query_one( + "SELECT confirmed_flush_lsn < $2::text::pg_lsn + FROM pg_replication_slots + WHERE slot_name = $1", + &[&slot_name, &lsn], + ) + .expect("failed to compare replication slot LSN") + .get(0) + } + + fn force_current_table_error( + table: &mut CdcTestTable, + source_table: &str, + source_table_oid: i32, + reason: &str, + ) { + let connector_url = cdc_connector_url(&table.url); + let pipeline_id = crate::integrated::postgres::cdc_input::pipeline_id( + &connector_url, + &table.publication_name, + source_table, + ) as i64; + let source_table_oid = source_table_oid as u32; + let metadata = json!({ + "type": "errored", + "reason": reason, + "solution": null, + "retry_policy": { "type": "manual_retry" }, + }); + + table + .client + .execute( + "WITH mark_old AS ( + UPDATE etl.replication_state + SET is_current = false, updated_at = now() + WHERE pipeline_id = $1 + AND table_id = $2::oid + AND is_current = true + RETURNING id + ) + INSERT INTO etl.replication_state + (pipeline_id, table_id, state, metadata, prev, is_current) + VALUES + ($1, $2::oid, 'errored'::etl.table_state, $3::jsonb, + (SELECT id FROM mark_old), true)", + &[&pipeline_id, &source_table_oid, &metadata], + ) + .expect("failed to force current etl table error"); + } + + /// True if etl has persisted a current `errored` replication state for + /// `source_table_oid`. + fn errored_state_persisted(table: &mut CdcTestTable, source_table_oid: i32) -> bool { + table + .client + .query_one( + "SELECT EXISTS ( + SELECT 1 + FROM etl.replication_state + WHERE table_id::int4 = $1 + AND state = 'errored'::etl.table_state + AND is_current = true + )", + &[&source_table_oid], + ) + .map(|row| row.get::<_, bool>(0)) + .unwrap_or(false) + } + + /// The terminal snapshot barrier must stay behind every copied row across + /// multiple controller queue flushes. + /// + /// The large rows span several `InputQueue` entries. After the initial + /// data-triggered step, the high minimum batch size suppresses further + /// automatic steps, so the barrier drives the remaining bounded flushes. + /// After copy-complete is persisted, the test restarts from the latest + /// checkpoint and snapshots the output. A prefix-only result means + /// copy-complete turned durable while a queue tail was still uncheckpointed. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_ft_snapshot_barrier_covers_buffered_tail() { + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_ft_snap_tail"); + let publication = unique_pg_name("cdc_pub_ft_snap_tail"); + const SNAPSHOT_ROWS: usize = 12; + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} (id, b, i, s) \ + SELECT id, true, id * 10, repeat('x', 1024 * 1024) \ + FROM generate_series(1, {SNAPSHOT_ROWS}) AS id" + )); + + let source_table = format!("public.{table_name}"); + let source_table_oid = table_oid(&mut table, &source_table); + + let storage = tempfile::tempdir().unwrap(); + let output_1 = NamedTempFile::new().unwrap(); + let (controller_1, errors_1) = cdc_ft_test_circuit_with_options( + &url, + &publication, + &source_table, + storage.path(), + output_1.path(), + CdcFtTestOptions { + max_batch_size: Some(1), + min_batch_size_records: 1_000_000, + max_buffering_delay_usecs: 3_600_000_000, + ..Default::default() + }, + ); + controller_1.start(); + + wait( + || { + let inserts = count_inserts(&read_output_json(output_1.path())); + inserts > 0 || !errors_1.is_empty() + }, + 60_000, + ) + .expect("timeout: terminal snapshot barrier did not start flushing the snapshot"); + assert!( + errors_1.is_empty(), + "unexpected error during snapshot: {:?}", + errors_1.try_recv() + ); + + let persisted_before_checkpoint = + wait(|| copy_state_persisted(&mut table, source_table_oid), 5_000).is_ok(); + assert!( + !persisted_before_checkpoint, + "etl persisted copy-complete before any snapshot checkpoint" + ); + + let mut copy_complete = false; + for _ in 0..SNAPSHOT_ROWS { + controller_1 + .checkpoint() + .expect("snapshot checkpoint failed"); + if wait(|| copy_state_persisted(&mut table, source_table_oid), 5_000).is_ok() { + copy_complete = true; + break; + } + } + assert!(copy_complete, "etl did not persist snapshot copy-complete"); + + controller_1.stop().unwrap(); + table.wait_replication_slots_inactive(); + + let output_2 = NamedTempFile::new().unwrap(); + let (controller_2, errors_2) = cdc_ft_test_circuit_with_options( + &url, + &publication, + &source_table, + storage.path(), + output_2.path(), + CdcFtTestOptions { + max_batch_size: Some(1), + min_batch_size_records: 1_000_000, + max_buffering_delay_usecs: 3_600_000_000, + send_snapshot: true, + ..Default::default() + }, + ); + controller_2.start(); + + let _ = wait( + || { + count_inserts(&read_output_json(output_2.path())) >= SNAPSHOT_ROWS + || !errors_2.is_empty() + }, + 10_000, + ); + assert!( + errors_2.is_empty(), + "unexpected error after restart: {:?}", + errors_2.try_recv() + ); + + let restored_rows = count_inserts(&read_output_json(output_2.path())); + assert_eq!( + restored_rows, SNAPSHOT_ROWS, + "snapshot copy-complete was persisted before every buffered row was checkpointed" + ); + + controller_2.stop().unwrap(); + } } diff --git a/crates/feldera-types/src/transport/postgres.rs b/crates/feldera-types/src/transport/postgres.rs index 16ce811e10b..52b72ccfb0e 100644 --- a/crates/feldera-types/src/transport/postgres.rs +++ b/crates/feldera-types/src/transport/postgres.rs @@ -90,6 +90,41 @@ pub struct PostgresCdcReaderConfig { #[serde(flatten)] #[schema(inline)] pub tls: PostgresTlsConfig, + + /// Maximum time to wait for each non-final CDC batch to become durable before + /// ingestion may continue without advancing the PostgreSQL replication position. + /// The final batch of an initial snapshot or catch-up always waits for durability. + #[serde(default = "default_streaming_ack_hold_ms")] + #[schema(default = default_streaming_ack_hold_ms)] + pub streaming_ack_hold_ms: u64, + + /// True to retry reading the table when the previous connector run stopped + /// before acknowledging pending data. + /// + /// These writes were not reported durable and are safe to reingest. + #[serde(default = "default_discard_shutdown_errors")] + #[schema(default = default_discard_shutdown_errors)] + pub discard_shutdown_errors: bool, + + /// True to retry reading the table with a persisted etl error when the connector starts. + /// + /// Recovery may repeat the initial snapshot. Define a primary key on the + /// input relation so replayed rows do not appear as duplicates. + #[serde(default = "default_discard_table_errors")] + #[schema(default = default_discard_table_errors)] + pub discard_table_errors: bool, +} + +pub const fn default_streaming_ack_hold_ms() -> u64 { + 2_000 +} + +pub const fn default_discard_shutdown_errors() -> bool { + true +} + +pub const fn default_discard_table_errors() -> bool { + false } impl PostgresCdcReaderConfig { @@ -102,6 +137,10 @@ impl PostgresCdcReaderConfig { return Err("source_table cannot be empty".to_string()); } + if self.streaming_ack_hold_ms == 0 { + return Err("streaming_ack_hold_ms must be greater than zero".to_string()); + } + if self.tls.ssl_client_pem.is_some() || self.tls.ssl_client_location.is_some() || self.tls.ssl_client_key.is_some() @@ -311,6 +350,9 @@ mod tests { publication: "publication".to_string(), source_table: "public.table".to_string(), tls, + streaming_ack_hold_ms: default_streaming_ack_hold_ms(), + discard_shutdown_errors: default_discard_shutdown_errors(), + discard_table_errors: default_discard_table_errors(), } } @@ -345,6 +387,19 @@ mod tests { assert!(config.validate().is_ok()); } + #[test] + fn postgres_cdc_config_discards_shutdown_errors_by_default() { + let config: PostgresCdcReaderConfig = serde_json::from_value(serde_json::json!({ + "uri": "postgres://user:password@localhost:5432/database", + "publication": "publication", + "source_table": "public.table", + })) + .unwrap(); + + assert!(config.discard_shutdown_errors); + assert!(!config.discard_table_errors); + } + #[test] fn postgres_cdc_config_rejects_empty_publication() { let mut config = postgres_cdc_config(PostgresTlsConfig::default()); diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs index 4057794b639..65efeadc4ce 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -728,6 +728,7 @@ pub fn generate_program_info( | TransportConfig::S3Input(_) | TransportConfig::DeltaTableInput(_) | TransportConfig::PostgresInput(_) + | TransportConfig::PostgresCdcInput(_) | TransportConfig::IcebergInput(_) | TransportConfig::Datagen(_) | TransportConfig::Nexmark(_) diff --git a/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md b/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md index 19cf9c38025..ad4ea3b20fa 100644 --- a/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md +++ b/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md @@ -24,13 +24,16 @@ privileges. Use transport name `postgres_cdc_input`. -| Property | Type | Default | Description | -| ----------------- | ------ | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `uri`\* | string | | PostgreSQL connection URL, e.g. `"postgres://postgres:password@localhost:5432/postgres"`. The URL must include a username, host, and database name. The user must have `REPLICATION` privilege. | -| `publication`\* | string | | Name of an existing PostgreSQL publication. The publication must include `source_table`. | -| `source_table`\* | string | | PostgreSQL table to replicate, usually schema-qualified, e.g. `"public.orders"`. | -| `ssl_ca_pem` | string | | CA certificates in PEM format. Setting this enables TLS and takes precedence over `ssl_ca_location`. | -| `ssl_ca_location` | string | | Path to a PEM file containing CA certificates. Used when `ssl_ca_pem` is not set. | +| Property | Type | Default | Description | +| ------------------------- | ------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `uri`\* | string | | PostgreSQL connection URI, e.g. `"postgres://postgres:password@localhost:5432/postgres"`. It must include a username, host, and database name. The user needs `REPLICATION` privilege. | +| `publication`\* | string | | Name of an existing PostgreSQL publication. The publication must include `source_table`. | +| `source_table`\* | string | | PostgreSQL table to replicate, usually schema-qualified, e.g. `"public.orders"`. | +| `ssl_ca_pem` | string | | CA certificates in PEM format. Setting this enables TLS and takes precedence over `ssl_ca_location`. | +| `ssl_ca_location` | string | | Path to a PEM file containing CA certificates. Used when `ssl_ca_pem` is not set. | +| `streaming_ack_hold_ms` | integer | `2000` | Time to wait for a nonterminal CDC batch to become durable before ingestion may continue without advancing its PostgreSQL WAL position. Must be greater than zero. | +| `discard_shutdown_errors` | boolean | `true` | Whether to retry a table when the previous connector run stopped before acknowledging pending data. | +| `discard_table_errors` | boolean | `false` | Whether to retry tables with persisted replication errors at startup. See [Discarding table errors](#discarding-table-errors). | [*]: Required fields @@ -87,6 +90,28 @@ Feldera matches columns by name. - If a required Feldera column is removed from PostgreSQL while the connector is running, the connector reports a fatal error. +We recommend defining a `PRIMARY KEY` on the Feldera input relation because the +connector provides at-least-once delivery. If the initial copy is interrupted, +the connector may read the source snapshot again from the beginning. This does +not create another copy inside PostgreSQL; it sends the same source rows to +Feldera again. If a Feldera checkpoint already contains rows from the earlier +attempt, a table with a primary key retains one row per key. Without a primary +key, the repeated rows can appear as duplicates. + +## Discarding table errors + +While the connector is stopping, its Feldera destination may close before it +sends an acknowledgment for pending data. On the next startup, +`discard_shutdown_errors` defaults to `true`, so the connector retries the +table from its last durable PostgreSQL position. Set the option to `false` to +disable this recovery. + +Other replication errors remain persisted by default. Set +`discard_table_errors` to `true` to discard persisted errors and retry the +affected tables at startup. This recovery may repeat the initial copy, so +define a primary key on the input relation (see [Schema requirements](#schema-requirements)) +to prevent duplicate rows. + ## Example First, create a PostgreSQL table and publication: @@ -168,11 +193,16 @@ CREATE TABLE orders ( ## Resume behavior -The connector stores replication state in PostgreSQL and uses logical -replication slots managed by the connector. Restarting a pipeline with the same -database host, port, database, publication, and source table resumes from the -existing replication state. Changing any of those values creates a different -replication identity and can cause a new snapshot. +The connector derives a stable identity from the source database host, port, +and database name together with `publication` and `source_table`. On an +ordinary restart with the same identity, it resumes from the last durable +position. If the connector stops before the initial copy is complete, it may +repeat that copy. + +Changing any of those identity fields starts a new initial copy. The PostgreSQL +username and password are not part of the identity, so rotating credentials +does not start a new copy. -Rotating the PostgreSQL username or password does not change the replication -identity. +Do not manually reset the connector's replication state. Feldera does not +provide a supported partial-reset procedure, and resetting state can replay +rows already present in the Feldera table. diff --git a/openapi.json b/openapi.json index cb4264f90c0..9995cef9f91 100644 --- a/openapi.json +++ b/openapi.json @@ -11944,6 +11944,16 @@ "source_table" ], "properties": { + "discard_shutdown_errors": { + "type": "boolean", + "description": "True to retry reading the table when the previous connector run stopped\nbefore acknowledging pending data.\n\nThese writes were not reported durable and are safe to reingest.", + "default": true + }, + "discard_table_errors": { + "type": "boolean", + "description": "True to retry reading the table with a persisted etl error when the connector starts.\n\nRecovery may repeat the initial snapshot. Define a primary key on the\ninput relation so replayed rows do not appear as duplicates.", + "default": false + }, "publication": { "type": "string", "description": "Name of the pre-created Postgres publication to replicate from." @@ -11952,6 +11962,13 @@ "type": "string", "description": "Postgres table to replicate (e.g. \"public.orders\").\nMust be included in the publication." }, + "streaming_ack_hold_ms": { + "type": "integer", + "format": "int64", + "description": "Maximum time to wait for each non-final CDC batch to become durable before\ningestion may continue without advancing the PostgreSQL replication position.\nThe final batch of an initial snapshot or catch-up always waits for durability.", + "default": 2000, + "minimum": 0 + }, "uri": { "type": "string", "description": "Postgres connection URI. The user must have REPLICATION privilege.\nSee: "