diff --git a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs index 65baa52389..229dc3cd71 100644 --- a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs +++ b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs @@ -64,7 +64,7 @@ fn retry_limit(shared: &Shared, started: Instant) -> Instant { .unwrap_or_else(|| started + Duration::from_millis(RUNNING_RETRY_LIMIT_MS)) } -/// Each SQLite busy wait ends by `give_up`, so no write outlasts the shutdown budget. +/// Each SQLite busy wait ends by `give_up`: the shutdown deadline, or 500 ms on for the final rows. fn bound_busy_wait(store: &SqliteStore, give_up: Instant) { let remaining: Duration = give_up .saturating_duration_since(Instant::now()) @@ -396,8 +396,8 @@ pub(crate) fn deliver(shared: &Arc, config: &SessionConfig) { state.rejecting = true; let rows = final_rows(config, &state); drop(state); - let give_up = (Instant::now() + Duration::from_millis(FINAL_ROWS_BUDGET_MS)) - .min(retry_limit(shared, Instant::now())); + let give_up = retry_limit(shared, Instant::now()) + .max(Instant::now() + Duration::from_millis(FINAL_ROWS_BUDGET_MS)); if let Err(error) = write_batch(store.as_ref(), &rows, give_up) { locked(&shared.state).degraded.get_or_insert(error.message); } diff --git a/crates/sqlbuild-recall/src/writer/constants.rs b/crates/sqlbuild-recall/src/writer/constants.rs index db4e1d877c..cc9d475e27 100644 --- a/crates/sqlbuild-recall/src/writer/constants.rs +++ b/crates/sqlbuild-recall/src/writer/constants.rs @@ -19,7 +19,7 @@ pub const RUNNING_RETRY_LIMIT_MS: u64 = 30_000; pub const SETUP_RETRY_MS: u64 = 50; /// Longest wait between wall-clock checks, so a resumed process beats promptly. pub const CLOCK_CHECK_MS: u64 = 1_000; -/// Budget for the final delivery rows, inside the shutdown budget. +/// Least time the final delivery rows get; rows landing after close then leave the CLI under-reporting, not the store. pub const FINAL_ROWS_BUDGET_MS: u64 = 500; /// Maintenance is skipped when less than this remains of the shutdown budget. pub const MINIMUM_MAINTENANCE_MS: u64 = 200; diff --git a/crates/sqlbuild-recall/src/writer/tests/helpers.rs b/crates/sqlbuild-recall/src/writer/tests/helpers.rs index 98d4956e20..fdf3dbe55d 100644 --- a/crates/sqlbuild-recall/src/writer/tests/helpers.rs +++ b/crates/sqlbuild-recall/src/writer/tests/helpers.rs @@ -19,7 +19,7 @@ use crate::writer::models::{ }; use crate::writer::tests::test_types::{ CopyingReplacement, CreatingStore, OutputBytesStore, PartialStore, PrefixReplacement, - RecordedSession, RecordingReplacement, + RecordedSession, RecordingReplacement, SlowFinalRowsStore, }; /// The envelope of invocation `invocation_id`. @@ -715,3 +715,31 @@ pub(crate) fn left_out_evidence(outcome: &str) -> Vec { RecallValue::Null, ] } + +/// Seconds `close` took, whether delivery reported complete, and how long after close began the final rows' deadline fell. +pub(crate) fn close_with_slow_final_rows(shutdown: Duration, delay: Duration) -> (f64, bool, f64) { + let store = SlowFinalRowsStore { + delay, + final_give_up: Arc::default(), + }; + let session = start_recall_session(SessionConfig { + store: SessionStore::Hosted(Arc::new(store.clone())), + shutdown_timeout: shutdown, + maintenance: false, + ..config(PathBuf::from("unused"), "slow-final-rows") + }); + session.append_output(OutputStream::Stdout, b"printed\n"); + let started = std::time::Instant::now(); + let summary = session.close(); + let elapsed = started.elapsed().as_secs_f64(); + let give_up = store + .final_give_up + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the final rows reached the store"); + ( + elapsed, + summary.flush_complete, + give_up.saturating_duration_since(started).as_secs_f64(), + ) +} diff --git a/crates/sqlbuild-recall/src/writer/tests/test_session.rs b/crates/sqlbuild-recall/src/writer/tests/test_session.rs index c8a9c5d218..bce9001c10 100644 --- a/crates/sqlbuild-recall/src/writer/tests/test_session.rs +++ b/crates/sqlbuild-recall/src/writer/tests/test_session.rs @@ -8,10 +8,11 @@ use crate::writer::models::{Priority, SetupOutcome}; #[cfg(unix)] use crate::writer::tests::helpers::unwritable_session; use crate::writer::tests::helpers::{ - artifact_counts, close_under_held_lock, creation_notices, event_counts, insert_heartbeat, - invocation_during_discovery, output_stored_while_running, partially_inserted, - pending_event_counts, record_run, recorded_session, replaced_session, replacement_order, - sent_before_the_window, session_on_corrupt_store, written_to_core_release_store, + artifact_counts, close_under_held_lock, close_with_slow_final_rows, creation_notices, + event_counts, insert_heartbeat, invocation_during_discovery, output_stored_while_running, + partially_inserted, pending_event_counts, record_run, recorded_session, replaced_session, + replacement_order, sent_before_the_window, session_on_corrupt_store, + written_to_core_release_store, }; #[cfg(unix)] use crate::writer::tests::test_types::UnwritableTestCase; @@ -19,7 +20,7 @@ use crate::writer::tests::test_types::{ ArtifactSizeTestCase, CoreReleaseStoreTestCase, CorruptStoreTestCase, CreationNoticeTestCase, DiscoveryTestCase, EarlyBatchTestCase, EventBuildTestCase, HeldLockTestCase, PartialInsertTestCase, PendingEventTestCase, RecordedSession, ReplacementOrderTestCase, - ReplacementTestCase, SessionTestCase, StoredWhileRunningTestCase, + ReplacementTestCase, SessionTestCase, SlowFinalRowsTestCase, StoredWhileRunningTestCase, }; const CLOSE_SLACK_SECONDS: f64 = 0.3; @@ -108,6 +109,44 @@ fn given_a_held_write_lock_when_closed_then_the_shutdown_budget_caps_the_delay() } } +#[test] +fn given_slow_final_delivery_rows_when_closed_then_the_shutdown_budget_bounds_them() { + let test_cases = [ + SlowFinalRowsTestCase { + description: "final rows slower than half a second but inside the budget record complete", + shutdown: Duration::from_secs(5), + final_rows_delay: Duration::from_millis(1_500), + expected_flush_complete: true, + }, + SlowFinalRowsTestCase { + description: "final rows slower than the shutdown budget still exit on time", + shutdown: Duration::from_millis(1_000), + final_rows_delay: Duration::from_secs(10), + expected_flush_complete: false, + }, + ]; + + for test_case in test_cases { + let (elapsed, flush_complete, final_rows_deadline) = + close_with_slow_final_rows(test_case.shutdown, test_case.final_rows_delay); + assert!( + final_rows_deadline <= test_case.shutdown.as_secs_f64() + CLOSE_SLACK_SECONDS, + "{}: the final rows may wait {final_rows_deadline:.3}s", + test_case.description + ); + assert!( + elapsed <= test_case.shutdown.as_secs_f64() + CLOSE_SLACK_SECONDS, + "{}: {elapsed:.3}s", + test_case.description + ); + assert_eq!( + flush_complete, test_case.expected_flush_complete, + "{}", + test_case.description + ); + } +} + #[test] fn given_a_file_that_is_not_a_database_when_started_then_it_is_moved_aside_once() { let test_cases = [CorruptStoreTestCase { diff --git a/crates/sqlbuild-recall/src/writer/tests/test_types.rs b/crates/sqlbuild-recall/src/writer/tests/test_types.rs index ad776499f5..9d16bfe4f4 100644 --- a/crates/sqlbuild-recall/src/writer/tests/test_types.rs +++ b/crates/sqlbuild-recall/src/writer/tests/test_types.rs @@ -361,3 +361,62 @@ pub(crate) struct FailedStatementTestCase { pub(crate) description: &'static str, pub(crate) expected_rows: Vec>, } + +pub(crate) struct SlowFinalRowsTestCase { + pub(crate) description: &'static str, + pub(crate) shutdown: Duration, + /// How long the store takes to accept the final delivery rows. + pub(crate) final_rows_delay: Duration, + pub(crate) expected_flush_complete: bool, +} + +/// A hosted store whose final delivery rows take `delay`, giving up at the deadline like HTTP. +#[derive(Clone, Debug)] +pub(crate) struct SlowFinalRowsStore { + pub(crate) delay: Duration, + /// The deadline the store was given for the final delivery rows. + pub(crate) final_give_up: Arc>>, +} + +impl RecallStore for SlowFinalRowsStore { + fn insert(&self, rows: &[RecallRow], give_up: Instant) -> Result { + if !rows.iter().any(|row| row.table == Table::Delivery) { + return Ok(InsertOutcome::default()); + } + *self + .final_give_up + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(give_up); + let answered: Instant = Instant::now() + self.delay; + std::thread::sleep( + answered + .min(give_up) + .saturating_duration_since(Instant::now()), + ); + if answered > give_up { + return Err(StoreError::new( + StoreErrorKind::Busy, + "the deadline expired", + )); + } + Ok(InsertOutcome::default()) + } +} + +impl RecallStoreOpener for SlowFinalRowsStore { + fn open( + &self, + _envelope: &RowEnvelope, + _give_up: Instant, + ) -> Result, StoreError> { + Ok(Box::new(self.clone())) + } + + fn batch_rows(&self) -> usize { + 1_000 + } + + fn batch_window(&self) -> Duration { + Duration::from_millis(10) + } +}