hubcio commented on code in PR #3957:
URL: https://github.com/apache/iggy/pull/3957#discussion_r3901995506
##########
core/integration/tests/connectors/postgres/postgres_source_cdc.rs:
##########
@@ -75,6 +74,17 @@ async fn poll_cdc_records(
received
}
+async fn slot_contains_change(pool: &sqlx::PgPool, expected_value: &str) ->
bool {
+ let changes = sqlx::query_scalar::<_, String>(
+ "SELECT data FROM pg_logical_slot_peek_changes($1, NULL, NULL)",
+ )
+ .bind(DEFAULT_SLOT)
+ .fetch_all(pool)
+ .await
+ .expect("CDC replication slot should be readable");
Review 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`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -468,31 +529,35 @@ impl PostgresSource {
}
}
- // Update state with minimal lock time
- if !messages.is_empty() {
- let mut state = self.state.lock().await;
- state.processed_rows += messages.len() as u64;
- }
-
if self.verbose {
info!("CDC: Fetched {} change records", messages.len());
} else {
debug!("CDC: Fetched {} change records", messages.len());
}
- Ok(messages)
+ let pending = if let Some(lsn) = last_lsn {
Review 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.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +613,87 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- 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(|| {
+ Error::InvalidRecordValue(format!(
+ "tracking column '{tracking_column}' is missing from
rows read from '{table}'"
+ ))
+ })?;
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ max_offset,
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
- }
- state.last_poll_time = Utc::now();
- }
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ })
+ } else {
+ None
+ };
- Ok(messages)
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || {
+ sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
Review 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.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -610,26 +708,32 @@ impl PostgresSource {
.join(", ");
if self.config.delete_after_read.unwrap_or(false) {
- let delete_query =
- 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}"
Review 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.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +613,87 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- 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(|| {
+ Error::InvalidRecordValue(format!(
+ "tracking column '{tracking_column}' is missing from
rows read from '{table}'"
+ ))
+ })?;
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ max_offset,
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
- }
- state.last_poll_time = Utc::now();
- }
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ })
+ } else {
+ None
+ };
- Ok(messages)
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || {
+ sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ error!("Failed to advance replication slot '{slot_name}' to {lsn}:
{e}");
+ Error::Connection(format!(
+ "failed to advance replication slot '{slot_name}' to {lsn}:
{e}"
+ ))
+ })?;
+ Ok(())
}
async fn mark_or_delete_processed_rows(
&self,
pool: &Pool<Postgres>,
table: &str,
- pk_column: &str,
ids: &[String],
+ max_offset: &str,
) -> Result<(), Error> {
if ids.is_empty() {
return Ok(());
}
let quoted_table = quote_qualified_identifier(table)?;
- let quoted_pk = quote_identifier(pk_column)?;
+ let quoted_pk = quote_identifier(self.primary_key_column())?;
+ let quoted_tracking = quote_identifier(self.tracking_column())?;
+ let tracking_boundary = format_offset_value(max_offset);
Review Comment:
numeric tracking columns go through `f64` before they become this boundary,
so above 2^53 the value rounds - up deletes rows that were never delivered,
down leaves the max row redelivered forever. keep the numeric text verbatim.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -610,26 +708,32 @@ impl PostgresSource {
.join(", ");
if self.config.delete_after_read.unwrap_or(false) {
- let delete_query =
- 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}"
+ );
if self.verbose {
info!("Deleting {} processed rows from '{table}'", ids.len());
} else {
debug!("Deleting {} processed rows from '{table}'", ids.len());
}
- sqlx::query(sqlx::AssertSqlSafe(delete_query))
- .execute(pool)
- .await
- .map_err(|e| {
- error!("Failed to delete processed rows: {e}");
- Error::InvalidRecord
- })?;
+ with_retry(
+ ||
sqlx::query(sqlx::AssertSqlSafe(delete_query.as_str())).execute(pool),
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ error!("Failed to delete processed rows: {e}");
+ Error::Connection(format!("failed to delete processed rows:
{e}"))
+ })?;
} else if let Some(processed_col) = &self.config.processed_column {
let quoted_processed = quote_identifier(processed_col)?;
let update_query = format!(
- "UPDATE {quoted_table} SET {quoted_processed} = TRUE WHERE
{quoted_pk} IN ({ids_list})"
+ "UPDATE {quoted_table} SET {quoted_processed} = TRUE \
+ WHERE {quoted_pk} IN ({ids_list}) AND {quoted_tracking} <=
{tracking_boundary}"
Review Comment:
also at line 713.
rows with a null tracking value are delivered on the first poll but `NULL <=
boundary` is never true, so they're never deleted or marked. add `OR tracking
IS NULL`, or match exact `(pk, tracking)` pairs instead.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +281,55 @@ impl Source for PostgresSource {
PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
};
- let persisted_state = self.serialize_state(&state);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
+ let pending = self.pending_batch.lock().await.take();
+ match result {
+ SourceBatchResult::Ack => {}
+ SourceBatchResult::Nack => return Ok(()),
+ }
+
+ let Some(pending) = pending else {
+ return Ok(());
+ };
+
+ for operation in pending.operations {
+ match operation {
+ PendingOperation::ProcessRows {
+ table,
+ ids,
+ max_offset,
+ } => {
+ self.mark_or_delete_processed_rows(self.get_pool()?,
&table, &ids, &max_offset)
+ .await?;
+ }
+ PendingOperation::AdvanceReplicationSlot { lsn } => {
+ self.advance_replication_slot(&lsn).await?;
+ }
+ }
+ }
+
+ *self.state.lock().await = pending.state;
Review Comment:
the runtime persists the advanced cursor before this runs, so a stop or
crash mid-batch leaves delivered rows undeleted and hidden behind the new
cursor forever. follow-up: persist the staged operations with the state and
replay them on `open()`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +613,87 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- 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(|| {
Review 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`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +281,55 @@ impl Source for PostgresSource {
PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
};
- let persisted_state = self.serialize_state(&state);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
Review Comment:
this runs inside the sdk's 30s batch-result timeout and the pool uses sqlx's
default 30s `acquire_timeout`, so a postgres blip at ack time stops the source
even though the ack was applied. set a short `acquire_timeout` so the retries
fit well under 30s.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +281,55 @@ impl Source for PostgresSource {
PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
};
- let persisted_state = self.serialize_state(&state);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
+ let pending = self.pending_batch.lock().await.take();
+ match result {
+ SourceBatchResult::Ack => {}
+ SourceBatchResult::Nack => return Ok(()),
+ }
+
+ let Some(pending) = pending else {
+ return Ok(());
+ };
+
+ for operation in pending.operations {
+ match operation {
+ PendingOperation::ProcessRows {
+ table,
+ ids,
+ max_offset,
+ } => {
+ self.mark_or_delete_processed_rows(self.get_pool()?,
&table, &ids, &max_offset)
Review Comment:
any error here stops the source for good while the cursor on disk has
already moved, so with two tables the first delete lands and the second table's
rows are stranded behind the new cursor. run the operations best-effort and
commit `pending.state` regardless.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +613,87 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- 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(|| {
+ Error::InvalidRecordValue(format!(
+ "tracking column '{tracking_column}' is missing from
rows read from '{table}'"
+ ))
+ })?;
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ max_offset,
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
- }
- state.last_poll_time = Utc::now();
- }
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ })
+ } else {
+ None
+ };
- Ok(messages)
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || {
+ sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ error!("Failed to advance replication slot '{slot_name}' to {lsn}:
{e}");
+ Error::Connection(format!(
Review Comment:
`pg_replication_slot_advance` raises `22023` when the target is below
`confirmed_flush` (two connectors on the shared default slot, or a manual
`get_changes`), and that stops the source too. treat it as already satisfied.
##########
core/connectors/runtime/src/stream.rs:
##########
@@ -24,6 +24,15 @@ use crate::error::RuntimeError;
const TOKEN_FILE_PREFIX: &str = "file:";
+fn append_query_parameters(connection_string: &str, parameters: &str) ->
String {
+ let separator = if connection_string.contains('?') {
Review Comment:
this scans the credentials too, so a `?` in the password or token flips the
separator to `&` and the connection string fails to parse at boot. check
`config.address` for `?` instead.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -422,38 +482,39 @@ impl PostgresSource {
// can still exceed it), so this isn't a hard per-call cap - but it
// stops the backlog from growing unbounded across many transactions
// the way NULL (no limit at all) did.
- let rows =
- sqlx::query("SELECT lsn, xid, data FROM
pg_logical_slot_get_changes($1, NULL, $2)")
- .bind(slot_name)
- .bind(batch_size)
- .fetch_all(pool)
- .await
- .map_err(|e| {
- error!("Failed to fetch CDC changes: {e}");
- Error::InvalidRecord
- })?;
+ let rows = sqlx::query(
+ "SELECT lsn::text AS lsn, data FROM
pg_logical_slot_peek_changes($1, NULL, $2)",
Review Comment:
this is the one database call left outside `with_retry`, and a connection
blip here fails the whole poll as `InvalidRecord`. wrap it like the others.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -2524,6 +2659,95 @@ mod tests {
});
}
+ #[tokio::test]
+ async fn
given_transient_database_errors_when_retrying_should_eventually_succeed() {
+ let attempts = AtomicU32::new(0);
+
+ let result = with_retry(
+ || async {
+ if attempts.fetch_add(1, Ordering::Relaxed) < 2 {
+ Err(sqlx::Error::PoolTimedOut)
+ } else {
+ Ok(())
+ }
+ },
+ 3,
+ 0,
+ )
+ .await;
+
+ assert!(result.is_ok());
+ assert_eq!(attempts.load(Ordering::Relaxed), 3);
+ }
+
+ #[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");
+ runtime.block_on(async {
+ let mut candidate_state = src.state.lock().await.clone();
+ candidate_state
+ .tracking_offsets
+ .insert("users".to_string(), "3".to_string());
+ candidate_state.processed_rows = 3;
+ *src.pending_batch.lock().await = Some(PendingBatch {
+ state: candidate_state,
+ operations: vec![
+ PendingOperation::ProcessRows {
+ table: "users".to_string(),
+ ids: vec!["3".to_string()],
+ max_offset: "3".to_string(),
+ },
+ PendingOperation::AdvanceReplicationSlot {
+ lsn: "0/16D32A0".to_string(),
+ },
+ ],
+ });
+
+ src.on_batch_result(SourceBatchResult::Nack)
+ .await
+ .expect("NACK should discard the candidate state");
+
+ {
+ let state = src.state.lock().await;
+ assert!(state.tracking_offsets.is_empty());
+ assert_eq!(state.processed_rows, 0);
+ }
+ assert!(src.pending_batch.lock().await.is_none());
+ });
+ }
+
+ #[test]
+ fn given_ack_when_batch_is_staged_should_commit_candidate_state() {
+ let src = PostgresSource::new(1, test_config(), None);
+ let runtime = tokio::runtime::Runtime::new().expect("failed to create
test runtime");
+ runtime.block_on(async {
+ let mut candidate_state = src.state.lock().await.clone();
+ candidate_state
+ .tracking_offsets
+ .insert("users".to_string(), "3".to_string());
+ candidate_state.processed_rows = 3;
+ *src.pending_batch.lock().await = Some(PendingBatch {
+ state: candidate_state,
+ operations: Vec::new(),
Review Comment:
the ack test stages no operations, so the failing-op path is untested: an
error at line 320 returns before the state commit and the staged batch is gone.
add a case with `pool: None` and one operation.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -219,7 +226,7 @@ primary_key_column = "id"
### Mark as Processed
-Updates a boolean column instead of deleting:
+Updates a boolean column after Iggy acknowledges the batch instead of deleting:
Review 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.
##########
.claude/skills/connector-source/SKILL.md:
##########
@@ -45,25 +45,31 @@ The macro shares the source as `Arc<T>` across the FFI
callback and forwarding l
### Lock discipline
-Never hold the state `Mutex` across upstream I/O. Canonical pattern (matches
`sources/postgres_source/src/lib.rs::poll_tables`):
+Never hold the state `Mutex` across upstream I/O. Build a candidate from
committed
+state, then stage it until the runtime reports the batch result:
```rust
-let cursor = { self.state.lock().await.cursor.clone() }; // brief read
-let rows = client.query(&sql, &[&cursor]).await?; // no lock held
-let persisted = { // brief write
- let mut state = self.state.lock().await;
- state.cursor = Some(new_cursor);
- ConnectorState::serialize(&*state, CONNECTOR_NAME, self.id)
-};
+let mut candidate = self.state.lock().await.clone();
+let rows = client.query(&sql, &[&candidate.cursor]).await?;
+candidate.cursor = Some(new_cursor);
+let persisted = ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id)
+ .ok_or_else(|| Error::Serialization("failed to serialize source
state".into()))?;
+*self.pending.lock().await = Some(candidate);
```
### State persistence
-- `ConnectorState` is `Vec<u8>` via MessagePack (`rmp_serde`). Use
`ConnectorState::serialize(&state, NAME, id)` +
`ConnectorState::deserialize::<State>(NAME, id)`. Both return `Option<T>` and
log on failure (non-fatal).
-- Runtime saves to `{state_path}/source_{key}.state` only after a successful
Iggy send. Between `poll()` returning and the runtime persisting the save, a
crash leaves the same cursor for the next poll - downstream must tolerate
at-least-once.
-- **Always return state in every `ProducedMessages`**, including empty polls.
Empty results still need to advance watermarks (timestamp sources) or affirm
"nothing new."
+- `ConnectorState` is `Vec<u8>` via MessagePack (`rmp_serde`). Use
`ConnectorState::serialize(&state, NAME, id)` +
`ConnectorState::deserialize::<State>(NAME, id)`.
+- `poll()` must not commit cursors or destructive work. Return messages with
candidate state and keep the corresponding work staged.
+- The runtime sends the batch, persists its candidate state, then calls
`on_batch_result(Ack)`. Commit staged in-memory state and external delete/mark
operations only on ACK. A NACK discards the candidate so the same data can be
polled again.
+- Return `state: None` for an empty poll when no watermark changed. If an
empty poll advances a watermark, stage and return the new state through the
same ACK handshake.
+- Treat candidate-state serialization failure as a poll error. Do not send
messages without the state needed to resume them safely.
- 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
Review 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.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -610,26 +708,32 @@ impl PostgresSource {
.join(", ");
if self.config.delete_after_read.unwrap_or(false) {
- let delete_query =
- 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}"
+ );
if self.verbose {
info!("Deleting {} processed rows from '{table}'", ids.len());
} else {
debug!("Deleting {} processed rows from '{table}'", ids.len());
}
- sqlx::query(sqlx::AssertSqlSafe(delete_query))
- .execute(pool)
- .await
- .map_err(|e| {
- error!("Failed to delete processed rows: {e}");
- Error::InvalidRecord
- })?;
+ with_retry(
+ ||
sqlx::query(sqlx::AssertSqlSafe(delete_query.as_str())).execute(pool),
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ error!("Failed to delete processed rows: {e}");
Review 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.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -2524,6 +2659,95 @@ mod tests {
});
}
+ #[tokio::test]
+ async fn
given_transient_database_errors_when_retrying_should_eventually_succeed() {
+ let attempts = AtomicU32::new(0);
+
+ let result = with_retry(
+ || async {
+ if attempts.fetch_add(1, Ordering::Relaxed) < 2 {
+ Err(sqlx::Error::PoolTimedOut)
+ } else {
+ Ok(())
+ }
+ },
+ 3,
+ 0,
+ )
+ .await;
+
+ assert!(result.is_ok());
+ assert_eq!(attempts.load(Ordering::Relaxed), 3);
+ }
+
+ #[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");
Review Comment:
the module now mixes `#[tokio::test]` (line 2662) with hand-built runtimes.
pick one, `#[tokio::test]` is shorter.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +281,55 @@ impl Source for PostgresSource {
PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
};
- let persisted_state = self.serialize_state(&state);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
+ let pending = self.pending_batch.lock().await.take();
+ match result {
Review 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
