v0.3.7-pre.011

crates/ksp-app-backfill-desk/src/backfill_run.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-app-backfill-desk/src/backfill_run.rs
-// version: 3
+// version: 4
 
 //! Single-active-run admission and launch ownership for Backfill Desk.
 
@@ -28,7 +28,7 @@ impl crate::BackfillRunLaunch {
 /// Single-run control slot retained by the application while one Backfill runtime is active.
 pub(crate) struct BackfillRunState {
     active: std::sync::Mutex<std::option::Option<ActiveBackfillRun>>,
-    last_terminal: std::sync::Mutex<std::option::Option<ksp_job_api::JobNotification<ksp_job_backfill_lib::BackfillJobSnapshot>>>,
+    last_terminal: std::sync::Mutex<std::option::Option<TerminalBackfillRun>>,
     next_sequence: std::sync::atomic::AtomicU64,
 }
 
@@ -61,7 +61,12 @@ impl crate::BackfillRunState {
     }
 
     /// Installs one control handle atomically and rejects concurrent Starts while another run owns the slot.
-    pub(crate) fn install(&self, job_id: ksp_job_api::JobId, handle: ksp_job_backfill_lib::BackfillJobHandle) -> ksp_core_lib::Result<()> {
+    pub(crate) fn install(
+        &self,
+        job_id: ksp_job_api::JobId,
+        handle: ksp_job_backfill_lib::BackfillJobHandle,
+        request: ksp_job_backfill_lib::BackfillRequest,
+    ) -> ksp_core_lib::Result<()> {
         let active = self.active.lock();
         let mut active = match active {
             std::result::Result::Ok(value) => value,
@@ -78,7 +83,7 @@ impl crate::BackfillRunState {
                 "Backfill Desk already has an active Backfill run",
             ));
         }
-        *active = std::option::Option::Some(ActiveBackfillRun { handle, job_id });
+        *active = std::option::Option::Some(ActiveBackfillRun { handle, job_id, request });
         return std::result::Result::Ok(());
     }
 
@@ -111,6 +116,7 @@ impl crate::BackfillRunState {
             },
         };
         let cancellation_requested = current.handle.is_cancellation_requested();
+        let request = current.request.clone();
         let source = current.handle.snapshots();
         let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
         if notification.state().is_terminal() {
@@ -124,7 +130,7 @@ impl crate::BackfillRunState {
                     ));
                 },
             };
-            *terminal = std::option::Option::Some(notification);
+            *terminal = std::option::Option::Some(TerminalBackfillRun { notification, request });
         }
         *active = std::option::Option::None;
         return std::result::Result::Ok(cancellation_requested);
@@ -178,7 +184,8 @@ impl crate::BackfillRunState {
                 ));
             },
         };
-        if let std::option::Option::Some(notification) = terminal.as_ref() {
+        if let std::option::Option::Some(terminal) = terminal.as_ref() {
+            let notification = &terminal.notification;
             if notification.id().as_str() != job_id {
                 return std::result::Result::Err(ksp_core_lib::Error::new(
                     crate::ERROR_CODE_BACKFILL_RUN_MISMATCH,
@@ -245,7 +252,58 @@ impl crate::BackfillRunState {
                 ));
             },
         };
-        return std::result::Result::Ok(terminal.clone());
+        return std::result::Result::Ok(terminal.as_ref().map(|terminal| return terminal.notification.clone()));
+    }
+
+    /// Rebuilds the last terminal request with its opaque checkpoint reissued onto one new Job identity.
+    pub(crate) fn resume_request(&self, job_id: ksp_job_api::JobId) -> ksp_core_lib::Result<ksp_job_backfill_lib::BackfillRequest> {
+        let active = self.active.lock();
+        let active = match active {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
+                    "Backfill Desk active-run resume state lock is poisoned",
+                ));
+            },
+        };
+        if active.is_some() {
+            return std::result::Result::Err(ksp_core_lib::Error::new(
+                crate::ERROR_CODE_BACKFILL_RUN_ACTIVE,
+                "Backfill Desk cannot resume while another Backfill run owns the single-run slot",
+            ));
+        }
+        drop(active);
+        let terminal = self.last_terminal.lock();
+        let terminal = match terminal {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
+                    "Backfill Desk terminal resume state lock is poisoned",
+                ));
+            },
+        };
+        let terminal = match terminal.as_ref() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_BACKFILL_RESUME_UNAVAILABLE,
+                    "Backfill Desk has no retained terminal run available for in-session Resume",
+                ));
+            },
+        };
+        let checkpoint = terminal.notification.snapshot().checkpoint();
+        let checkpoint = match checkpoint {
+            std::option::Option::Some(value) => value.clone(),
+            std::option::Option::None => {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_BACKFILL_RESUME_UNAVAILABLE,
+                    "Backfill Desk retained terminal run has no safe checkpoint for in-session Resume",
+                ));
+            },
+        };
+        return terminal.request.resume_for_job(job_id, checkpoint);
     }
 
     /// Rolls back a just-installed admission when execution resources cannot be acquired before spawn.
@@ -261,6 +319,13 @@ impl crate::BackfillRunState {
 struct ActiveBackfillRun {
     handle: ksp_job_backfill_lib::BackfillJobHandle,
     job_id: ksp_job_api::JobId,
+    request: ksp_job_backfill_lib::BackfillRequest,
+}
+
+#[derive(Clone)]
+struct TerminalBackfillRun {
+    notification: ksp_job_api::JobNotification<ksp_job_backfill_lib::BackfillJobSnapshot>,
+    request: ksp_job_backfill_lib::BackfillRequest,
 }
 
 #[cfg(test)]