Merge 'core/mvcc: properly complete checkpoint in case of error' from Nikita Sivukhin

Call `self.mvstore.storage.on_checkpoint_end` in the
`cleanup_after_external_io_error` to properly cleanup checkpoint in
cased of IO error

Reviewed-by: Mikaël Francoeur (@LeMikaelF)

Closes #7413

Author: sivukhin

core/connection.rs

@@ -1721,7 +1721,7 @@ impl Connection {
                     Ok(TransitionResult::Done(result)) => return Ok(result),
                     Ok(TransitionResult::Io(iocompletions)) => {
                         if let Err(err) = iocompletions.wait(io.as_ref()) {
-                            ckpt_sm.cleanup_after_external_io_error();
+                            ckpt_sm.cleanup_after_external_io_error(err.clone())?;
                             return Err(err);
                         }
                     }

core/mvcc/database/checkpoint_state_machine.rs

@@ -688,7 +688,10 @@ impl<Clock: LogicalClock> CheckpointStateMachine<Clock> {
     /// Cleanup path for I/O errors that happen while waiting on completions outside
     /// of `step()`. This mirrors `step()` error handling and also resets pager/WAL
     /// checkpoint bookkeeping.
-    pub fn cleanup_after_external_io_error(&mut self) {
+    pub fn cleanup_after_external_io_error(&mut self, err: LimboError) -> Result<()> {
+        // run storage cleanup within proper checkpoint context (e.g. pager has pending read/write txn)
+        let result = self.mvstore.storage.on_checkpoint_end(Err(err));
+
         if self.lock_states.pager_write_tx {
             self.pager.rollback_tx(self.connection.as_ref());
             if self.update_transaction_state {
@@ -716,6 +719,8 @@ impl<Clock: LogicalClock> CheckpointStateMachine<Clock> {
             self.checkpoint_lock.unlock();
             self.lock_states.blocking_checkpoint_lock_held = false;
         }
+
+        result
     }
 
     /// Returns all checkpointable [RowVersion]s for that `table_id`
@@ -1558,6 +1563,7 @@ impl<Clock: LogicalClock> CheckpointStateMachine<Clock> {
                         *special_write,
                     )
                 };
+                tracing::debug!("WriteRow: num_columns={num_columns}, table_id={table_id:?}, special_write={special_write:?}");
 
                 // Handle CREATE TABLE / DROP TABLE / CREATE INDEX / DROP INDEX ops
                 if let Some(special_write) = special_write {
@@ -1703,6 +1709,8 @@ impl<Clock: LogicalClock> CheckpointStateMachine<Clock> {
                     })
                 };
 
+                tracing::debug!("WriteRow: resolved root page: root_page={root_page}");
+
                 // If a table was created, it now has a real root page allocated for it, but the 'root_page' field in the sqlite_schema record is still the table id.
                 // So we need to rewrite the row version to use the real root page.
                 if let Some(SpecialWrite::BTreeCreate {
@@ -2255,9 +2263,11 @@ impl<Clock: LogicalClock> StateTransition for CheckpointStateMachine<Clock> {
         let res = self.step_inner(&());
         match res {
             Err(ref err) => {
-                self.mvstore.storage.on_checkpoint_end(Err(err.clone()))?;
                 tracing::debug!("Error in checkpoint state machine: {err}");
-                self.cleanup_after_external_io_error();
+                // `cleanup_after_external_io_error` already emits the paired
+                // `on_checkpoint_end(Err(..))`, so don't call it here too — doing both
+                // double-fires the hook for a single failure.
+                self.cleanup_after_external_io_error(err.clone())?;
                 res
             }
             Ok(TransitionResult::Done(ref result)) => {

core/mvcc/database/mod.rs

@@ -1394,10 +1394,13 @@ pub struct DeleteRowStateMachine {
 impl<Clock: LogicalClock> CommitStateMachine<Clock> {
     pub(crate) fn cleanup_mvcc_checkpoint_state(&mut self) {
         if let CommitState::Checkpoint { state_machine } = &mut self.state {
-            state_machine
+            let _ = state_machine
                 .lock()
                 .inner_mut()
-                .cleanup_after_external_io_error();
+                .cleanup_after_external_io_error(LimboError::InternalError(
+                    "mvcc: cleanup_unfinished_commit".to_string(),
+                ))
+                .inspect_err(|e| tracing::error!("cleanup_after_external_io_error failed: {e}"));
         }
     }
 

core/mvcc/database/tests.rs

@@ -2852,7 +2852,9 @@ fn test_checkpoint_resamples_boundary_before_starting() {
         mvcc_store.durable_txid_max.load(Ordering::SeqCst),
         update_ts
     );
-    interrupted_checkpoint.cleanup_after_external_io_error();
+    interrupted_checkpoint
+        .cleanup_after_external_io_error(LimboError::Interrupt)
+        .unwrap();
 
     let mut finished = false;
     for _ in 0..50_000 {

core/storage/pager.rs

@@ -6124,7 +6124,7 @@ mod ptrmap_tests {
             target_idx,
             PendingRead {
                 page: synthetic_page.clone(),
-                disk_read: Some(stub_disk_read.clone()),
+                disk_read: Some(stub_disk_read),
             },
         );
 

core/vdbe/execute.rs

@@ -616,7 +616,7 @@ pub fn op_checkpoint(
                 Ok(TransitionResult::Done(result)) => break result,
                 Ok(TransitionResult::Io(iocompletions)) => {
                     if let Err(err) = iocompletions.wait(pager.io.as_ref()) {
-                        ckpt_sm.cleanup_after_external_io_error();
+                        ckpt_sm.cleanup_after_external_io_error(err.clone())?;
                         return Err(err);
                     }
                 }