v0.3.7-pre.010

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-app-backfill-desk/src/backfill_run.rs
-// version: 2
+// version: 3
 
 //! Single-active-run admission and launch ownership for Backfill Desk.
 
@@ -130,6 +130,91 @@ impl crate::BackfillRunState {
         return std::result::Result::Ok(cancellation_requested);
     }
 
+    /// Requests cooperative cancellation for the exact backend-issued active Job identity.
+    pub(crate) fn cancel(&self, job_id: &str) -> ksp_core_lib::Result<crate::BackfillCancelResponseDto> {
+        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 cancellation state lock is poisoned",
+                ));
+            },
+        };
+        if let std::option::Option::Some(current) = active.as_ref() {
+            if current.job_id.as_str() != job_id {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_BACKFILL_RUN_MISMATCH,
+                    "Backfill Desk cancellation target does not match the active Job",
+                ));
+            }
+            let source = current.handle.snapshots();
+            let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
+            let state = notification.state();
+            if state.is_terminal() {
+                return std::result::Result::Ok(crate::BackfillCancelResponseDto {
+                    accepted: false,
+                    job_id: current.job_id.as_str().to_owned(),
+                    state: state.code().to_owned(),
+                });
+            }
+            let accepted = current.handle.cancel();
+            let state_code = if current.handle.is_cancellation_requested() { "cancelling" } else { state.code() };
+            return std::result::Result::Ok(crate::BackfillCancelResponseDto {
+                accepted,
+                job_id: current.job_id.as_str().to_owned(),
+                state: state_code.to_owned(),
+            });
+        }
+        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 cancellation state lock is poisoned",
+                ));
+            },
+        };
+        if let std::option::Option::Some(notification) = terminal.as_ref() {
+            if notification.id().as_str() != job_id {
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_BACKFILL_RUN_MISMATCH,
+                    "Backfill Desk cancellation target does not match the retained terminal Job",
+                ));
+            }
+            return std::result::Result::Ok(crate::BackfillCancelResponseDto {
+                accepted: false,
+                job_id: notification.id().as_str().to_owned(),
+                state: notification.state().code().to_owned(),
+            });
+        }
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_BACKFILL_RUN_NOT_ACTIVE,
+            "Backfill Desk has no active Backfill run to cancel",
+        ));
+    }
+
+    /// Requests best-effort cooperative cancellation of whichever run is active after shutdown wins admission.
+    pub(crate) fn cancel_active_for_shutdown(&self) -> ksp_core_lib::Result<bool> {
+        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 shutdown cancellation state lock is poisoned",
+                ));
+            },
+        };
+        return match active.as_ref() {
+            std::option::Option::Some(current) => std::result::Result::Ok(current.handle.cancel()),
+            std::option::Option::None => std::result::Result::Ok(false),
+        };
+    }
+
     /// Returns the complete current active snapshot or the retained latest terminal snapshot for resynchronization.
     pub(crate) fn current_notification(
         &self,