fix(connectors): defer postgres source progress until ack - #3957
fix(connectors): defer postgres source progress until ack#3957rohankumardubey wants to merge 3 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3957 +/- ##
=============================================
- Coverage 83.93% 57.69% -26.24%
Complexity 1358 1358
=============================================
Files 1213 1212 -1
Lines 168441 139746 -28695
Branches 135999 107313 -28686
=============================================
- Hits 141373 80622 -60751
- Misses 23389 55625 +32236
+ Partials 3679 3499 -180
🚀 New features to boost your workflow:
|
|
/ready |
|
/request-review @hubcio |
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
a few things without diff lines to hang them on:
.claude/skills/connector-source/SKILL.mdstill teaches the pre-ack pattern this pr removes (the "matches poll_tables" snippet writes cursors during poll, "always return state in every ProducedMessages" no longer holds for empty polls, and state-serialization failure is now a hard poll error, lib.rs:288-296) and doesn't mentionon_batch_resultat all - needs an update to the new contract.mark_or_delete_processed_rowsmaps db errors toError::InvalidRecord(lib.rs:732, :751) - misleading now that these errors surface through the ack path;Error::Connectionlikeadvance_replication_slotuses would fit better.- README.md:15 still promises offset tracking "avoid duplicates" - at-least-once redelivery means duplicates are possible; and README.md:57 documents the
poll_intervaldefault as1swhile the code fallback is10s(lib.rs:180).
| return Ok(()); | ||
| }; | ||
|
|
||
| for operation in pending.operations { |
There was a problem hiding this comment.
any error in this loop permanently stops the connector - the sdk treats Err from on_batch_result as fatal, and the staged delete/mark ops are dropped while the state file already advanced past those rows. transient pg errors (deadlock, failover) should retry here; note with_retry can't wrap these calls directly (wrong error type), it has to go around the inner .execute()s.
| }) | ||
| } | ||
|
|
||
| async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> { |
There was a problem hiding this comment.
after a nack the next poll re-reads the same rows, so while iggy is down every batch nacks - and the sdk stops the connector for good at 5 consecutive nacks. with the shipped 1s poll interval a short outage is enough. before, the counter got reset by empty-poll acks, but only because progress was committed pre-send (the bug being fixed). needs a decision: exempt send-failure nacks from the stop counter, or document the manual-restart behavior loudly.
| continue; | ||
| } | ||
| }; | ||
| let lsn: String = row.try_get("lsn").map_err(|e| { |
There was a problem hiding this comment.
a row failing here fails the whole poll forever - right call for at-least-once (can't skip a row and advance the slot past it), but poll errors are invisible: no stats bump, status stays Running, and the stuck slot pins restart_lsn so pg wal grows without bound. worth a readme note about monitoring slot lag.
| if !processed_ids.is_empty() { | ||
| self.mark_or_delete_processed_rows(pool, table, pk_column, &processed_ids) | ||
| .await?; | ||
| operations.push(PendingOperation::ProcessRows { |
There was a problem hiding this comment.
the delete/mark now runs a full send+ack after the select, so a row updated in between gets deleted (or marked processed) and its new version is never delivered when tracking_column is an updated-at column. AND {tracking} <= '{max_offset}' in the where clause would close it; at minimum a readme caveat.
| ### Delete After Read | ||
|
|
||
| Deletes rows from the source table after successful processing: | ||
| Deletes rows from the source table only after Iggy acknowledges the batch: |
There was a problem hiding this comment.
worth documenting the window: state is persisted before the delete runs, so a crash/shutdown/timeout in between leaves rows delivered but never deleted, and the advanced offset means they are never picked up again.
| .connectors_runtime_mut() | ||
| .expect("connectors runtime") | ||
| // Keep a failed send bounded instead of waiting indefinitely for Iggy to return. | ||
| .set_iggy_connection_options("reconnection_retries=0"); |
There was a problem hiding this comment.
reconnection_retries=0 stays set for the verification phase too (build_envs reruns on every start), so the post-restart polls get no reconnect slack. fine today, just tightens the timing budget for no reason.
| state.processed_rows += total_processed; | ||
| let pending = if total_processed > 0 { | ||
| candidate_state.processed_rows += total_processed; | ||
| for (table, offset) in state_updates { |
There was a problem hiding this comment.
state_updates and this merge loop are leftovers from the old locked-state design - candidate_state is a local clone now, offsets can go straight into it in the table loop. drops the stale lock-era comments too.
| } | ||
|
|
||
| #[derive(Debug)] | ||
| enum PendingOperation { |
There was a problem hiding this comment.
primary_key_column/slot_name are derivable from config at ack time - small slot_name()/pk_column() helpers (like payload_format()) would dedupe the three existing fallback chains and drop these two fields. optional.
| Error::InvalidRecord | ||
| })?; | ||
| let rows = sqlx::query( | ||
| "SELECT lsn::text AS lsn, xid, data FROM pg_logical_slot_peek_changes($1, NULL, $2)", |
There was a problem hiding this comment.
xid is selected but never read - drop it.
| use std::time::Duration; | ||
| use tokio::time::sleep; | ||
|
|
||
| const API_KEY: &str = "test-api-key"; |
There was a problem hiding this comment.
API_KEY/SOURCE_KEY duplicate postgres_source_cdc.rs - SOURCE_KEY fits in postgres/mod.rs next to the other shared consts. source_errors/wait_for_source_errors could also share one fetch helper returning Option.
github-actions
Bot
added
S-waiting-on-author
2 participants
Which issue does this PR address?
Closes #3635
Rationale
The PostgreSQL source advanced tracking offsets, deleted or marked rows, and consumed CDC changes before Iggy confirmed delivery. A failed send could therefore permanently skip source records.
What changed?
PostgreSQL polling now stages cursor updates and row operations until the runtime reports a successful batch acknowledgment. NACK discards the staged work, while ACK commits the state and performs the pending delete or mark operations.
CDC now peeks logical-slot changes and advances the slot only after acknowledgment. A deterministic regression test stops Iggy during delivery and verifies that the PostgreSQL rows are redelivered after restart.
Local Execution
cargo fmt --all -- --checkcargo clippy -p iggy_connector_postgres_source -p integration --all-features --all-targets -- -D warningsgit diff --check