From c2614e910bc074bb927379ef3f880b9e054e2817 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:11:42 +0000 Subject: [PATCH 1/3] fix: let Recall's final delivery rows use the remaining shutdown budget The delivery summary rows written when a session closes were capped at 500 ms even when most of shutdown_timeout remained, so one slow hosted store response marked an otherwise fully delivered transcript incomplete. They now get the rest of the shutdown budget, and at least 500 ms; close still returns by the shutdown deadline. --- .../src/writer/_helpers/delivery.rs | 4 +- .../sqlbuild-recall/src/writer/constants.rs | 2 +- .../src/writer/tests/helpers.rs | 16 +++++- .../src/writer/tests/test_session.rs | 44 +++++++++++++-- .../src/writer/tests/test_types.rs | 53 +++++++++++++++++++ 5 files changed, 110 insertions(+), 9 deletions(-) diff --git a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs index 65baa52389..540e10515e 100644 --- a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs +++ b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs @@ -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..e35e10cf90 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; otherwise they use what remains of the shutdown budget. 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..524049c97b 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,17 @@ pub(crate) fn left_out_evidence(outcome: &str) -> Vec { RecallValue::Null, ] } + +/// Seconds `close` took and whether delivery reported complete, when the final delivery rows take `delay`. +pub(crate) fn close_with_slow_final_rows(shutdown: Duration, delay: Duration) -> (f64, bool) { + let session = start_recall_session(SessionConfig { + store: SessionStore::Hosted(Arc::new(SlowFinalRowsStore { delay })), + 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(); + (started.elapsed().as_secs_f64(), summary.flush_complete) +} diff --git a/crates/sqlbuild-recall/src/writer/tests/test_session.rs b/crates/sqlbuild-recall/src/writer/tests/test_session.rs index c8a9c5d218..22493675cf 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,39 @@ 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) = + close_with_slow_final_rows(test_case.shutdown, test_case.final_rows_delay); + 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..696fbb767d 100644 --- a/crates/sqlbuild-recall/src/writer/tests/test_types.rs +++ b/crates/sqlbuild-recall/src/writer/tests/test_types.rs @@ -361,3 +361,56 @@ 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, +} + +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()); + } + 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) + } +} From fe03f15151b264023cf4df2487f711158b53d1e9 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:55:24 +0000 Subject: [PATCH 2/3] test: pin the deadline Recall gives its final delivery rows --- .../src/writer/_helpers/delivery.rs | 2 +- .../src/writer/tests/helpers.rs | 22 +++++++++++++++---- .../src/writer/tests/test_session.rs | 7 +++++- .../src/writer/tests/test_types.rs | 6 +++++ 4 files changed, 31 insertions(+), 6 deletions(-) diff --git a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs index 540e10515e..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()) diff --git a/crates/sqlbuild-recall/src/writer/tests/helpers.rs b/crates/sqlbuild-recall/src/writer/tests/helpers.rs index 524049c97b..fdf3dbe55d 100644 --- a/crates/sqlbuild-recall/src/writer/tests/helpers.rs +++ b/crates/sqlbuild-recall/src/writer/tests/helpers.rs @@ -716,10 +716,14 @@ pub(crate) fn left_out_evidence(outcome: &str) -> Vec { ] } -/// Seconds `close` took and whether delivery reported complete, when the final delivery rows take `delay`. -pub(crate) fn close_with_slow_final_rows(shutdown: Duration, delay: Duration) -> (f64, bool) { +/// 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(SlowFinalRowsStore { delay })), + store: SessionStore::Hosted(Arc::new(store.clone())), shutdown_timeout: shutdown, maintenance: false, ..config(PathBuf::from("unused"), "slow-final-rows") @@ -727,5 +731,15 @@ pub(crate) fn close_with_slow_final_rows(shutdown: Duration, delay: Duration) -> session.append_output(OutputStream::Stdout, b"printed\n"); let started = std::time::Instant::now(); let summary = session.close(); - (started.elapsed().as_secs_f64(), summary.flush_complete) + 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 22493675cf..bce9001c10 100644 --- a/crates/sqlbuild-recall/src/writer/tests/test_session.rs +++ b/crates/sqlbuild-recall/src/writer/tests/test_session.rs @@ -127,8 +127,13 @@ fn given_slow_final_delivery_rows_when_closed_then_the_shutdown_budget_bounds_th ]; for test_case in test_cases { - let (elapsed, flush_complete) = + 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", diff --git a/crates/sqlbuild-recall/src/writer/tests/test_types.rs b/crates/sqlbuild-recall/src/writer/tests/test_types.rs index 696fbb767d..9d16bfe4f4 100644 --- a/crates/sqlbuild-recall/src/writer/tests/test_types.rs +++ b/crates/sqlbuild-recall/src/writer/tests/test_types.rs @@ -374,6 +374,8 @@ pub(crate) struct SlowFinalRowsTestCase { #[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 { @@ -381,6 +383,10 @@ impl RecallStore for SlowFinalRowsStore { 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 From 870ff52b94d9db530e6f075f65c4892cdd2c69ff Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:58:59 +0000 Subject: [PATCH 3/3] docs: note why Recall's final-row floor may outlast close --- crates/sqlbuild-recall/src/writer/constants.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/sqlbuild-recall/src/writer/constants.rs b/crates/sqlbuild-recall/src/writer/constants.rs index e35e10cf90..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; -/// Least time the final delivery rows get; otherwise they use what remains of 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;