From bfe85711493585212dce81377fd43ffd585e78ed Mon Sep 17 00:00:00 2001 From: Jordan McQueen Date: Sat, 22 Aug 2026 04:21:59 +0900 Subject: [PATCH 1/4] Test dirty restarts with Postgres store --- crates/etl/tests/pipeline_dirty_restart.rs | 51 +++++++++++++++++----- 1 file changed, 40 insertions(+), 11 deletions(-) diff --git a/crates/etl/tests/pipeline_dirty_restart.rs b/crates/etl/tests/pipeline_dirty_restart.rs index d3dd6c2c5..385165a94 100644 --- a/crates/etl/tests/pipeline_dirty_restart.rs +++ b/crates/etl/tests/pipeline_dirty_restart.rs @@ -5,7 +5,7 @@ use etl::{ event::Event, pipeline::PipelineId, schema::{TableId, TableName}, - store::{StateStore, TableStateType}, + store::{PostgresStore, StateStore, TableStateType}, test_utils::{ database::{replication_slot_state, spawn_source_database}, faults::FaultyOp, @@ -131,6 +131,27 @@ async fn wait_for_notification( .map_err(|_| TestCaseError::fail(format!("timed out waiting for {description}"))) } +/// Waits until the persistent state store reports completed table sync. +async fn wait_for_table_sync_complete( + store: &PostgresStore, + table_id: TableId, +) -> Result<(), TestCaseError> { + tokio::time::timeout(DIRTY_RESTART_TIMEOUT, async { + loop { + let state = store.get_table_state(table_id).await.map_err(|error| { + TestCaseError::fail(format!("failed to read table state: {error}")) + })?; + if state.as_ref().is_some_and(|state| state.as_type().has_completed_table_sync()) { + return Ok(()); + } + + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .map_err(|_| TestCaseError::fail("timed out waiting for table sync to complete"))? +} + /// Inserts one autocommit user transaction and waits for its acknowledgement. async fn insert_user_and_wait( database: &mut PgDatabase, @@ -340,10 +361,14 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr let users_schema = database_schema.users_schema(); let table_id = users_schema.id; - let store = NotifyingStore::new(); - let memory_destination = MemoryDestination::new(store.clone()); + let destination_store = NotifyingStore::new(); + let memory_destination = MemoryDestination::new(destination_store); let first_destination = TestDestinationWrapper::wrap(memory_destination.clone()); - let pipeline_id: PipelineId = random(); + let pipeline_id: PipelineId = u64::from(random::()); + let first_store = + PostgresStore::new(pipeline_id, database.config.clone()).await.map_err(|error| { + TestCaseError::fail(format!("failed to create the first Postgres store: {error}")) + })?; let apply_slot_name: String = EtlReplicationSlot::for_apply_worker(pipeline_id).try_into().unwrap(); let sync_slot_name: String = @@ -352,16 +377,15 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr &database.config, pipeline_id, database_schema.publication_name(), - store.clone(), + first_store.clone(), first_destination.clone(), ); - let users_sync_complete = store.notify_on_table_sync_complete(table_id).await; first_pipeline .start() .await .map_err(|error| TestCaseError::fail(format!("pipeline failed to start: {error}")))?; - wait_for_notification(&users_sync_complete, "users table synchronization to complete").await?; + wait_for_table_sync_complete(&first_store, table_id).await?; // Table sync completes before the worker deletes its progress row and // replication slot. Wait for slot removal so the crash below hits a state @@ -416,13 +440,18 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr wait_for_apply_disconnect(database.client.as_ref().unwrap(), &apply_slot_name).await?; drop(first_destination); drop(held_response); + drop(first_store); + let restarted_store = + PostgresStore::new(pipeline_id, database.config.clone()).await.map_err(|error| { + TestCaseError::fail(format!("failed to reopen the Postgres store: {error}")) + })?; let restarted_destination = TestDestinationWrapper::wrap(memory_destination.clone()); let mut restarted_pipeline = create_pipeline( &database.config, pipeline_id, database_schema.publication_name(), - store.clone(), + restarted_store.clone(), restarted_destination.clone(), ); restarted_pipeline.start().await.map_err(|error| { @@ -447,7 +476,7 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr TestCaseError::fail(format!("restarted pipeline shutdown failed: {error}")) })?; - let table_state = store + let table_state = restarted_store .get_table_state(table_id) .await .map_err(|error| TestCaseError::fail(format!("failed to read table state: {error}")))? @@ -475,11 +504,11 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr } #[tokio::test(flavor = "multi_thread")] -async fn dirty_restart_at_randomized_positions_converges_without_recopy() { +async fn postgres_store_dirty_restart_at_randomized_positions_converges_without_recopy() { init_test_tracing(); let strategy = dirty_restart_cases(); - run_expensive_property("dirty restart convergence", &strategy, |case| { + run_expensive_property("Postgres store dirty restart convergence", &strategy, |case| { block_on(run_dirty_restart_case(*case)) }); } From 073183494754e03208b696d91a43421f9369bb79 Mon Sep 17 00:00:00 2001 From: Jordan McQueen Date: Tue, 25 Aug 2026 03:37:04 +0900 Subject: [PATCH 2/4] Share table sync wait test helper --- crates/etl/src/test_utils/mod.rs | 1 + crates/etl/src/test_utils/store.rs | 46 ++++++++++++++++++++++ crates/etl/tests/pipeline_dirty_restart.rs | 26 ++---------- 3 files changed, 51 insertions(+), 22 deletions(-) create mode 100644 crates/etl/src/test_utils/store.rs diff --git a/crates/etl/src/test_utils/mod.rs b/crates/etl/src/test_utils/mod.rs index a091b9a56..2d67c45e0 100644 --- a/crates/etl/src/test_utils/mod.rs +++ b/crates/etl/src/test_utils/mod.rs @@ -20,5 +20,6 @@ pub mod pipeline; pub mod property; pub mod replication_stream; pub mod schema; +pub mod store; pub mod test_destination_wrapper; pub mod test_schema; diff --git a/crates/etl/src/test_utils/store.rs b/crates/etl/src/test_utils/store.rs new file mode 100644 index 000000000..64d716264 --- /dev/null +++ b/crates/etl/src/test_utils/store.rs @@ -0,0 +1,46 @@ +use std::time::Duration; + +use tokio::time::{sleep, timeout}; + +use crate::{ + error::{ErrorKind, EtlResult}, + etl_error, + schema::TableId, + store::StateStore, +}; + +/// Interval between persistent table-state reads while waiting for a condition. +const TABLE_STATE_POLL_INTERVAL: Duration = Duration::from_millis(50); + +/// Waits until a table has completed its initial synchronization. +/// +/// Returns an error when the store read fails or `timeout_duration` elapses. +pub async fn wait_for_table_sync_complete( + store: &S, + table_id: TableId, + timeout_duration: Duration, +) -> EtlResult<()> +where + S: StateStore, +{ + let wait = async { + loop { + let state = store.get_table_state(table_id).await?; + if state.as_ref().is_some_and(|state| state.as_type().has_completed_table_sync()) { + return Ok(()); + } + + sleep(TABLE_STATE_POLL_INTERVAL).await; + } + }; + + timeout(timeout_duration, wait).await.map_err(|_| { + etl_error!( + ErrorKind::Unknown, + "Timed out waiting for table synchronization", + format!( + "Table {table_id} did not complete synchronization within {timeout_duration:?}" + ) + ) + })? +} diff --git a/crates/etl/tests/pipeline_dirty_restart.rs b/crates/etl/tests/pipeline_dirty_restart.rs index 385165a94..2d0daa800 100644 --- a/crates/etl/tests/pipeline_dirty_restart.rs +++ b/crates/etl/tests/pipeline_dirty_restart.rs @@ -15,6 +15,7 @@ use etl::{ notifying_store::NotifyingStore, pipeline::create_pipeline, property::{block_on, run_expensive_property}, + store::wait_for_table_sync_complete, test_destination_wrapper::TestDestinationWrapper, test_schema::{TableSelection, insert_users_data, setup_test_database_schema}, }, @@ -131,27 +132,6 @@ async fn wait_for_notification( .map_err(|_| TestCaseError::fail(format!("timed out waiting for {description}"))) } -/// Waits until the persistent state store reports completed table sync. -async fn wait_for_table_sync_complete( - store: &PostgresStore, - table_id: TableId, -) -> Result<(), TestCaseError> { - tokio::time::timeout(DIRTY_RESTART_TIMEOUT, async { - loop { - let state = store.get_table_state(table_id).await.map_err(|error| { - TestCaseError::fail(format!("failed to read table state: {error}")) - })?; - if state.as_ref().is_some_and(|state| state.as_type().has_completed_table_sync()) { - return Ok(()); - } - - tokio::time::sleep(Duration::from_millis(50)).await; - } - }) - .await - .map_err(|_| TestCaseError::fail("timed out waiting for table sync to complete"))? -} - /// Inserts one autocommit user transaction and waits for its acknowledgement. async fn insert_user_and_wait( database: &mut PgDatabase, @@ -385,7 +365,9 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr .start() .await .map_err(|error| TestCaseError::fail(format!("pipeline failed to start: {error}")))?; - wait_for_table_sync_complete(&first_store, table_id).await?; + wait_for_table_sync_complete(&first_store, table_id, DIRTY_RESTART_TIMEOUT).await.map_err( + |error| TestCaseError::fail(format!("failed to wait for table sync completion: {error}")), + )?; // Table sync completes before the worker deletes its progress row and // replication slot. Wait for slot removal so the crash below hits a state From 923d51055fc80752b105fcb0d98764ac97d35780 Mon Sep 17 00:00:00 2001 From: Jordan McQueen Date: Tue, 25 Aug 2026 04:43:17 +0900 Subject: [PATCH 3/4] test: wait for PostgresStore readiness --- crates/etl/src/test_utils/store.rs | 36 +++++++++++++++++++++- crates/etl/tests/pipeline_dirty_restart.rs | 16 +++++++++- 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/crates/etl/src/test_utils/store.rs b/crates/etl/src/test_utils/store.rs index 64d716264..f67ba27bb 100644 --- a/crates/etl/src/test_utils/store.rs +++ b/crates/etl/src/test_utils/store.rs @@ -6,7 +6,7 @@ use crate::{ error::{ErrorKind, EtlResult}, etl_error, schema::TableId, - store::StateStore, + store::{StateStore, TableStateType}, }; /// Interval between persistent table-state reads while waiting for a condition. @@ -44,3 +44,37 @@ where ) })? } + +/// Waits until a table reaches the expected state type. +/// +/// Returns an error when the store read fails or `timeout_duration` elapses. +pub async fn wait_for_table_state_type( + store: &S, + table_id: TableId, + expected_state: TableStateType, + timeout_duration: Duration, +) -> EtlResult<()> +where + S: StateStore, +{ + let wait = async { + loop { + let state = store.get_table_state(table_id).await?; + if state.as_ref().is_some_and(|state| state.as_type() == expected_state) { + return Ok(()); + } + + sleep(TABLE_STATE_POLL_INTERVAL).await; + } + }; + + timeout(timeout_duration, wait).await.map_err(|_| { + etl_error!( + ErrorKind::Unknown, + "Timed out waiting for table state", + format!( + "Table {table_id} did not reach {expected_state:?} within {timeout_duration:?}" + ) + ) + })? +} diff --git a/crates/etl/tests/pipeline_dirty_restart.rs b/crates/etl/tests/pipeline_dirty_restart.rs index 2d0daa800..4a4dd8b6b 100644 --- a/crates/etl/tests/pipeline_dirty_restart.rs +++ b/crates/etl/tests/pipeline_dirty_restart.rs @@ -15,7 +15,7 @@ use etl::{ notifying_store::NotifyingStore, pipeline::create_pipeline, property::{block_on, run_expensive_property}, - store::wait_for_table_sync_complete, + store::{wait_for_table_state_type, wait_for_table_sync_complete}, test_destination_wrapper::TestDestinationWrapper, test_schema::{TableSelection, insert_users_data, setup_test_database_schema}, }, @@ -451,6 +451,20 @@ async fn run_dirty_restart_case(case: DirtyRestartCase) -> Result<(), TestCaseEr .await?; } + // Destination event recording does not guarantee that apply-side response + // processing has finished. Wait for the quiescent pass to promote the table + // before requesting shutdown. + wait_for_table_state_type( + &restarted_store, + table_id, + TableStateType::Ready, + DIRTY_RESTART_TIMEOUT, + ) + .await + .map_err(|error| { + TestCaseError::fail(format!("failed to wait for users table readiness: {error}")) + })?; + tokio::time::timeout(DIRTY_RESTART_TIMEOUT, restarted_pipeline.shutdown_and_wait()) .await .map_err(|_| TestCaseError::fail("timed out waiting for restarted pipeline shutdown"))? From ea9e7f7c0e101d56df51e1e9ed54770c94505fae Mon Sep 17 00:00:00 2001 From: Jordan McQueen Date: Wed, 26 Aug 2026 03:02:52 +0900 Subject: [PATCH 4/4] test: clarify dirty restart event wait --- crates/etl/tests/pipeline_dirty_restart.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/etl/tests/pipeline_dirty_restart.rs b/crates/etl/tests/pipeline_dirty_restart.rs index 4a4dd8b6b..ba2e04277 100644 --- a/crates/etl/tests/pipeline_dirty_restart.rs +++ b/crates/etl/tests/pipeline_dirty_restart.rs @@ -132,7 +132,8 @@ async fn wait_for_notification( .map_err(|_| TestCaseError::fail(format!("timed out waiting for {description}"))) } -/// Inserts one autocommit user transaction and waits for its acknowledgement. +/// Inserts one autocommit user transaction and waits for destination event +/// recording. async fn insert_user_and_wait( database: &mut PgDatabase, users_table_name: &TableName, @@ -145,7 +146,8 @@ async fn insert_user_and_wait( insert_users_data(database, users_table_name, user_number..=user_number).await; - wait_for_notification(&delivered, format!("acknowledgement of user {user_id}")).await + wait_for_notification(&delivered, format!("destination event recording for user {user_id}")) + .await } /// Waits until the old apply worker releases its replication slot connection.