fix(connectors): defer postgres source progress until ack - #3957
fix(connectors): defer postgres source progress until ack#3957rohankumardubey wants to merge 11 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 85.00% 84.96% -0.04%
Complexity 1402 1402
============================================
Files 1225 1225
Lines 180283 180436 +153
Branches 146587 146740 +153
============================================
+ Hits 153248 153309 +61
- Misses 22993 23052 +59
- Partials 4042 4075 +33
🚀 New features to boost your workflow:
|
|
/ready |
|
/request-review @hubcio |
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).
|
@rohankumardubey please write |
|
/ready |
|
looks like CI is failing. |
|
The failure came from a merge-resolution type mismatch in the TLS connection-string branch. Thanks @hubcio |
|
/ready |
hubcio
left a comment
There was a problem hiding this comment.
the staging model is right and closes the old send-failure hole. two things block: 55006 isn't transient, so any slot collision at ack time stops the source for good, and an empty cdc peek never advances the slot, so an idle database retains wal without bound. the tracking <= boundary guard also regresses custom_query, null and numeric tracking users, and the new cdc test races the connector on the slot.
outside this diff, two pre-existing bugs in build_polling_query worth their own issue: $now is replaced before $now_unix so $now_unix never expands, and $offset is substituted raw into custom_query from row data.
| let pool = self.get_pool()?; | ||
| with_retry( | ||
| || { | ||
| sqlx::query("SELECT pg_replication_slot_advance($1, $2::pg_lsn)") |
There was a problem hiding this comment.
55006 (slot active for another pid) isn't in is_transient_error, so one collision on this advance stops the source for good with zero retries. add it to the transient list.
| debug!("CDC: Fetched {} change records", messages.len()); | ||
| } | ||
| Ok(messages) | ||
| let pending = if let Some(lsn) = last_lsn { |
There was a problem hiding this comment.
an empty peek never advances the slot (get_changes confirmed the reader position on every poll), so an idle database in a busy cluster now pins restart_lsn and retains wal without bound. read pg_current_wal_flush_lsn() before the peek and advance to it when no rows come back.
| .bind(DEFAULT_SLOT) | ||
| .fetch_all(pool) | ||
| .await | ||
| .expect("CDC replication slot should be readable"); |
There was a problem hiding this comment.
this peeks the connector's live slot without a 55006 retry while the connector polls it every 50ms, so either side can lose the nowait acquire and the test flakes. retry on 55006 like drop_replication_slot_retrying.
| format!("DELETE FROM {quoted_table} WHERE {quoted_pk} IN ({ids_list})"); | ||
| let delete_query = format!( | ||
| "DELETE FROM {quoted_table} WHERE {quoted_pk} IN ({ids_list}) \ | ||
| AND {quoted_tracking} <= {tracking_boundary}" |
There was a problem hiding this comment.
max_offset is the last row's value, not the max, so with a custom_query that isn't ordered by tracking this guard skips rows and they get redelivered every poll. apply the boundary only on the built query path, or match exact (pk, tracking) pairs.
| self.mark_or_delete_processed_rows(pool, table, pk_column, &processed_ids) | ||
| .await?; | ||
| if self.should_process_rows() && !processed_ids.is_empty() { | ||
| let max_offset = max_offset.clone().ok_or_else(|| { |
There was a problem hiding this comment.
a custom query that selects the pk but not the tracking column now fails every poll here (when primary_key_column differs from tracking_column), and the source keeps reporting running. make max_offset optional and skip the boundary when it's None.
| ) | ||
| .await | ||
| .map_err(|e| { | ||
| error!("Failed to delete processed rows: {e}"); |
There was a problem hiding this comment.
also at lines 674, 752.
with_retry already logs the failure and the sdk logs the returned error again, so this is the third line for the same event. drop it, the error payload carries the context.
| ### Mark as Processed | ||
|
|
||
| Updates a boolean column instead of deleting: | ||
| Updates a boolean column after Iggy acknowledges the batch instead of deleting: |
There was a problem hiding this comment.
worth one sentence here: a row whose tracking value moves past the batch boundary between poll and ack is left unmarked and comes back next poll.
| - Keep `State` small - rewritten every batch. No unbounded vecs. | ||
|
|
||
| The SDK allows one in-flight batch. Five consecutive NACKs stop the source and | ||
| require a manual restart. Returning `Err` from `on_batch_result` is fatal, so |
There was a problem hiding this comment.
the callback runs inside the sdk's 30s batch-result window and this tells authors to retry inside it without saying so. name the budget.
|
|
||
| async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> { | ||
| let pending = self.pending_batch.lock().await.take(); | ||
| match result { |
There was a problem hiding this comment.
match plus let Some ... else collapses to one line: let (SourceBatchResult::Ack, Some(pending)) = (result, self.pending_batch.lock().await.take()) else { return Ok(()) };. same shape fits TEMPLATE.md.
| #[test] | ||
| fn given_nack_when_batch_is_staged_should_keep_committed_state() { | ||
| let src = PostgresSource::new(1, test_config(), None); | ||
| let runtime = tokio::runtime::Runtime::new().expect("failed to create test runtime"); |
There was a problem hiding this comment.
the module now mixes #[tokio::test] (line 2662) with hand-built runtimes. pick one, #[tokio::test] is shorter.
|
@hubcio Agreed, I’ll track staged-operation persistence and replay, along with the two pre-existing |
|
@rohankumardubey sure thing. if you plan to do that, please fill-in the TODOs in code so that it won't be forgotten (in relevant places). |
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