From 760a6c807b7633d7304740fa87f05289d7072eca Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 02:13:59 +0000 Subject: [PATCH] test: keep completed rows past ring rotation in job output and staged tests Queue-storage clients built with the default settings rotate the ready ring every second. While a test is still running, or inside the two-second shutdown window, maintenance can rotate the ring and prune completed rows, so a later admin::get_job or count of a terminal job finds nothing. This made external_completion_stores_the_output_it_is_given fail intermittently in CI with JobNotFound { id: 1 }, and the same race affects failed_or_late_staging_never_commits_under_queue_storage, which counts a completed trigger after it finishes. job_output_test.rs gains a client_builder helper that sets a one-hour queue and lease rotation, as completion_stores_the_output_on_both_terminal_paths already did by hand, and every test that reads terminal rows after completion uses it. The shared client_builder in staged_follow_up_test.rs sets the same rotation for its queue-storage branch. No test in either file relies on rotation or pruning. --- awa/tests/job_output_test.rs | 39 ++++++++++++++++++------------ awa/tests/staged_follow_up_test.rs | 9 ++++++- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/awa/tests/job_output_test.rs b/awa/tests/job_output_test.rs index 7fd14a72a..410efb89e 100644 --- a/awa/tests/job_output_test.rs +++ b/awa/tests/job_output_test.rs @@ -38,6 +38,20 @@ fn queue_config() -> QueueConfig { } } +/// A queue-storage client that keeps terminal rows for the whole test. +/// +/// The default one-second ring rotation lets maintenance rotate the ready +/// ring and prune completed rows while the test runs, or inside its +/// `shutdown` window, so a later read of a terminal job finds nothing on a +/// slow machine. +fn client_builder(pool: &sqlx::PgPool) -> awa::ClientBuilder { + Client::builder(pool.clone()).queue_storage( + awa::QueueStorageConfig::default(), + Duration::from_secs(3600), + Duration::from_secs(3600), + ) +} + async fn insert( pool: &sqlx::PgPool, args: &impl JobArgs, @@ -84,14 +98,7 @@ async fn completion_stores_the_output_on_both_terminal_paths() { let queue = "output_paths"; let events: Arc>>> = Arc::default(); let seen = events.clone(); - // The default one-second ring rotation lets prune truncate completed - // rows before the table checks below run on a slow machine. - let client = Client::builder(pool.clone()) - .queue_storage( - awa::QueueStorageConfig::default(), - Duration::from_secs(3600), - Duration::from_secs(3600), - ) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(|args, ctx| { let output = ctx.output_slot(); @@ -204,7 +211,7 @@ async fn failures_keep_only_the_latest_output_and_dispatch_strips_it() { let queue = "output_failures"; let dispatch_errors: Arc>>> = Arc::default(); let seen = dispatch_errors.clone(); - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(move |_args, ctx| { let attempt = ctx.job.attempt; @@ -291,7 +298,7 @@ async fn snooze_discards_and_a_failed_set_clears_without_failing() { let queue = "output_snooze"; let errors: Arc>> = Arc::default(); let recorded = errors.clone(); - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(move |_args, ctx| { let attempt = ctx.job.attempt; @@ -381,7 +388,7 @@ async fn enqueue_specs_receive_the_row_without_its_output() { let db = TestDatabase::queue_storage().await; let pool = db.pool().clone(); let queue = "output_specs"; - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(|args, ctx| { let output = ctx.output_slot(); @@ -450,7 +457,7 @@ async fn external_completion_stores_the_output_it_is_given() { let db = TestDatabase::queue_storage().await; let pool = db.pool().clone(); let queue = "output_external"; - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register_worker(AwaitWorker) .build() @@ -524,7 +531,7 @@ async fn a_schema_without_output_columns_drops_completion_outputs() { .unwrap(); let queue = "output_pre_v052"; - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(|args, ctx| { let output = ctx.output_slot(); @@ -625,7 +632,7 @@ async fn panic_and_cancel_keep_the_output_in_their_error_entry() { let db = TestDatabase::queue_storage().await; let pool = db.pool().clone(); let queue = "output_panic_cancel"; - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(|args, ctx| { let output = ctx.output_slot(); @@ -688,7 +695,7 @@ async fn the_dlq_keeps_the_failed_output_and_a_retry_dispatches_without_it() { let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let leaked: Arc>> = Arc::default(); let (runs_seen, leaked_seen) = (runs.clone(), leaked.clone()); - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .dlq_enabled_by_default(true) .register::(move |_args, ctx| { @@ -851,7 +858,7 @@ async fn typed_outputs_round_trip_and_publish_their_type() { let queue = "output_typed"; let mismatches: Arc>> = Arc::default(); let seen = mismatches.clone(); - let client = Client::builder(pool.clone()) + let client = client_builder(&pool) .queue(queue, queue_config()) .register::(move |args, ctx| { // The wrong kind's slot fails, and clears like any failed set. diff --git a/awa/tests/staged_follow_up_test.rs b/awa/tests/staged_follow_up_test.rs index c570e7a16..5c8805ccb 100644 --- a/awa/tests/staged_follow_up_test.rs +++ b/awa/tests/staged_follow_up_test.rs @@ -48,7 +48,14 @@ fn client_builder(pool: &sqlx::PgPool, canonical: bool) -> awa::ClientBuilder { if canonical { builder.canonical_storage() } else { - builder + // The default one-second ring rotation lets maintenance rotate the + // ready ring and prune completed rows while the test runs, so a later + // count of a completed job finds nothing on a slow machine. + builder.queue_storage( + awa::QueueStorageConfig::default(), + Duration::from_secs(3600), + Duration::from_secs(3600), + ) } }