From 19b8d33d8b9fc15049a4b78405e2877dc06cd235 Mon Sep 17 00:00:00 2001 From: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> Date: Wed, 8 Jul 2026 15:40:58 +0545 Subject: [PATCH] fix(postgres-cdc): gate acks on durability; fix shutdown Previously, etl could mark the initial snapshot as complete before the snapshot rows were included in a Feldera checkpoint. This meant a restart could skip the snapshot even though Feldera resumed from a checkpoint that didn't contain those rows. Make the final snapshot ack wait until all snapshot rows are durable. Streaming acks now wait for step completion, or for a checkpoint when fault tolerance is enabled. Regular streaming batches can return Accepted after `streaming_ack_hold_ms`, which lets etl continue reading WAL without advancing the replication slot. Also signal etl shutdown before dropping pending acks. Otherwise, a clean stop could persist an errored table state and prevent the connector from starting again. Upgrade etl to 8762c0fc, which contains the shutdown fix, and add tests for snapshot retries, streaming acks and replication slot advancement. Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> --- Cargo.lock | 216 ++-- Cargo.toml | 6 +- .../src/integrated/postgres/cdc_input.rs | 1039 +++++++++++++-- .../adapters/src/integrated/postgres/test.rs | 1141 ++++++++++++++++- .../feldera-types/src/transport/postgres.rs | 55 + .../pipeline-manager/src/db/types/program.rs | 1 + .../docs/connectors/sources/postgresql-cdc.md | 58 +- openapi.json | 17 + 8 files changed, 2289 insertions(+), 244 deletions(-) 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: "