fix: let Recall's final delivery rows use the remaining shutdown budget

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<Shared>, 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);
             }

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;

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> {
         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(),
+    )
+}

crates/sqlbuild-recall/src/writer/tests/test_session.rs

@@ -8,18 +8,19 @@ 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;
 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 {

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<Vec<RecallValue>>,
 }
+
+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<Mutex<Option<Instant>>>,
+}
+
+impl RecallStore for SlowFinalRowsStore {
+    fn insert(&self, rows: &[RecallRow], give_up: Instant) -> Result<InsertOutcome, StoreError> {
+        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<Box<dyn RecallStore>, StoreError> {
+        Ok(Box::new(self.clone()))
+    }
+
+    fn batch_rows(&self) -> usize {
+        1_000
+    }
+
+    fn batch_window(&self) -> Duration {
+        Duration::from_millis(10)
+    }
+}