From 6d33c24cda9a94753d5a487df6709d603bd2e613 Mon Sep 17 00:00:00 2001 From: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> Date: Tue, 7 Jul 2026 20:10:02 +0545 Subject: [PATCH] fix(etl): skip errored state and interrupt flush wait on table-copy shutdown Currently, during table-copy, if the destination drops the AsyncResult while the pipeline is shutting down, the table ends up in an Errored state. But dropping the AsyncResult during shutdown should be safe: on restart, we should just ensure the retry copies the table from the start. Also make the copy loop stop waiting on the flush result once shutdown fires, instead of blocking on a destination that never completes it. Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> --- crates/etl/src/replication/table_sync/copy.rs | 12 +- crates/etl/src/runtime/table_sync/worker.rs | 15 ++ crates/etl/tests/pipeline.rs | 161 +++++++++++++++++- 3 files changed, 185 insertions(+), 3 deletions(-) diff --git a/crates/etl/src/replication/table_sync/copy.rs b/crates/etl/src/replication/table_sync/copy.rs index 67be5c926..88fef8feb 100644 --- a/crates/etl/src/replication/table_sync/copy.rs +++ b/crates/etl/src/replication/table_sync/copy.rs @@ -756,7 +756,17 @@ where destination .write_table_rows(&replicated_table_schema, table_rows, flush_result) .await?; - let write_status = pending_flush_result.await.into_result()?; + let write_status = tokio::select! { + biased; + + _ = shutdown_rx.changed() => { + return Ok(ShutdownResult::Shutdown(progress)); + } + + completed = pending_flush_result => { + completed.into_result()? + } + }; progress.record_batch(batch_size, write_status); diff --git a/crates/etl/src/runtime/table_sync/worker.rs b/crates/etl/src/runtime/table_sync/worker.rs index 23e05da76..451e000e2 100644 --- a/crates/etl/src/runtime/table_sync/worker.rs +++ b/crates/etl/src/runtime/table_sync/worker.rs @@ -400,6 +400,21 @@ where ) -> EtlResult> { error!(table_id = table_id.0, error = %err, "table sync worker failed"); + // A failure that surfaces while shutdown is already in progress is a + // consequence of the shutdown itself (e.g. the destination rejects or + // drops in-flight work), not a replication problem. Do not persist it: + // a stored `Errored` state is never retried across restarts, so the + // table would otherwise stall on every later run. A dropped sender also + // counts as shutdown. + if shutdown_rx.has_changed().unwrap_or(true) { + info!( + table_id = table_id.0, + "table sync worker failed during shutdown, skipping error state persistence" + ); + + return Ok(Some(TableSyncWorkerResult::Shutdown)); + } + // Build a retry policy from the shared classifier. The concrete retry timestamp // is computed in the worker from config so both table sync and apply // worker use the same retry timing settings. diff --git a/crates/etl/tests/pipeline.rs b/crates/etl/tests/pipeline.rs index 717c21879..85a76f652 100644 --- a/crates/etl/tests/pipeline.rs +++ b/crates/etl/tests/pipeline.rs @@ -7,10 +7,11 @@ use etl::{ WriteEventsResult, WriteTableRowsResult, }, error::{ErrorKind, EtlResult}, + etl_error, event::{Event, EventType, InsertEvent}, pipeline::PipelineId, schema::{ColumnSchema, ReplicatedTableSchema, TableId}, - store::{SchemaStore, TableState, TableStateType}, + store::{SchemaStore, StateStore, TableState, TableStateType}, test_utils::{ database::{spawn_source_database, test_table_name}, event::{EventCondition, group_events_by_type_and_table_id}, @@ -41,7 +42,7 @@ use etl_telemetry::tracing::init_test_tracing; use pg_escape::{quote_identifier, quote_literal}; use rand::random; use tokio::{ - sync::{Mutex, Notify}, + sync::{Mutex, Notify, watch}, time::sleep, }; use tokio_postgres::types::{PgLsn, Type}; @@ -230,6 +231,162 @@ async fn pipeline_shutdown_calls_destination_shutdown() { assert!(destination.shutdown_called().await); } +/// How [`StalledCopyDestination`] handles table-copy writes. +#[derive(Clone, Copy)] +enum CopyWriteBehavior { + /// Parks the write, then fails it once the release signal fires. + BlockThenFail, + /// Accepts the write but never completes its [`WriteTableRowsResult`]. + NeverCompleteFlush, +} + +/// Destination that stalls table-copy writes so tests can interleave pipeline +/// shutdown with an in-flight copy. +#[derive(Clone)] +struct StalledCopyDestination { + behavior: CopyWriteBehavior, + write_entered: Arc, + release_rx: watch::Receiver, + held_flush_results: Arc>>, +} + +impl StalledCopyDestination { + /// Creates the destination together with the sender that releases parked + /// writes. + fn new(behavior: CopyWriteBehavior) -> (Self, watch::Sender) { + let (release_tx, release_rx) = watch::channel(false); + + let destination = Self { + behavior, + write_entered: Arc::new(Notify::new()), + release_rx, + held_flush_results: Arc::new(std::sync::Mutex::new(Vec::new())), + }; + + (destination, release_tx) + } + + /// Waits until a table-copy write reached the destination. + async fn wait_for_copy_write(&self) { + self.write_entered.notified().await; + } +} + +impl Destination for StalledCopyDestination { + fn name() -> &'static str { + "stalled_copy" + } + + async fn drop_table_for_copy( + &self, + _replicated_table_schema: &ReplicatedTableSchema, + async_result: DropTableForCopyResult<()>, + ) -> EtlResult<()> { + async_result.send(Ok(())); + + Ok(()) + } + + async fn write_table_rows( + &self, + _replicated_table_schema: &ReplicatedTableSchema, + _table_rows: Vec, + async_result: WriteTableRowsResult, + ) -> EtlResult<()> { + self.write_entered.notify_one(); + + match self.behavior { + CopyWriteBehavior::BlockThenFail => { + let mut release_rx = self.release_rx.clone(); + release_rx + .wait_for(|released| *released) + .await + .expect("release channel should stay open"); + + Err(etl_error!(ErrorKind::DestinationError, "Copy write failed")) + } + CopyWriteBehavior::NeverCompleteFlush => { + // Keep the flush result alive without completing it, so the + // copy loop stays blocked on the pending flush. + self.held_flush_results.lock().unwrap().push(async_result); + + Ok(()) + } + } + } + + async fn write_events( + &self, + _events: Vec, + async_result: WriteEventsResult, + ) -> EtlResult<()> { + async_result.send(Ok(DestinationWriteStatus::Durable)); + + Ok(()) + } +} + +/// Verifies that a table-copy write blocked on shutdown neither hangs the +/// pipeline nor persists an `Errored` table state. +/// +/// Covers both ways a stalled copy write can behave once shutdown starts: +/// [`CopyWriteBehavior::BlockThenFail`] fails the write after shutdown was +/// signaled (must not persist the error), and +/// [`CopyWriteBehavior::NeverCompleteFlush`] never completes the flush at all +/// (must not hang shutdown). +async fn assert_shutdown_during_stalled_copy_leaves_table_retryable(behavior: CopyWriteBehavior) { + init_test_tracing(); + + let mut database = spawn_source_database().await; + let database_schema = setup_test_database_schema(&database, TableSelection::UsersOnly).await; + + insert_users_data(&mut database, &database_schema.users_schema().name, 1..=10).await; + + let store = NotifyingStore::new(); + let (destination, release_tx) = StalledCopyDestination::new(behavior); + + let pipeline_id: PipelineId = random(); + let mut pipeline = create_pipeline( + &database.config, + pipeline_id, + database_schema.publication_name(), + store.clone(), + destination.clone(), + ); + + pipeline.start().await.unwrap(); + + // Park the copy write first, then signal shutdown, so any write failure + // surfaces only while shutdown is already in progress. Releasing is a + // no-op for `NeverCompleteFlush`, which never reads from the channel. + destination.wait_for_copy_write().await; + pipeline.shutdown(); + release_tx.send(true).unwrap(); + + pipeline.wait().await.unwrap(); + + // Neither a shutdown-induced write failure nor an abandoned flush wait + // may persist an errored state; the table must restart the copy on the + // next run. + let table_state = + store.get_table_state(database_schema.users_schema().id).await.unwrap().unwrap(); + assert!(matches!(table_state, TableState::DataSync)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn table_sync_worker_error_during_shutdown_is_not_persisted() { + assert_shutdown_during_stalled_copy_leaves_table_retryable(CopyWriteBehavior::BlockThenFail) + .await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn table_copy_shutdown_interrupts_pending_flush_wait() { + assert_shutdown_during_stalled_copy_leaves_table_retryable( + CopyWriteBehavior::NeverCompleteFlush, + ) + .await; +} + #[tokio::test(flavor = "multi_thread")] async fn pipeline_fails_when_slot_deleted_with_non_init_tables() { init_test_tracing();