v0.3.7-pre.009

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-app-backfill-desk/src/backfill_run.rs
-// version: 1
+// version: 2
 
 //! Single-active-run admission and launch ownership for Backfill Desk.
 
@@ -9,6 +9,8 @@ pub(crate) struct BackfillRunLaunch {
     pub(crate) job_id: ksp_job_api::JobId,
     /// Concrete Backfill runtime owning discovery, hydration, persistence and latest-value publication.
     pub(crate) runtime: ksp_job_backfill_lib::BackfillJobRuntime,
+    /// Independent latest-value source retained by the monitoring bridge.
+    pub(crate) snapshots: ksp_job_backfill_lib::BackfillSnapshotSource,
     /// Store facade temporarily borrowed from the application Store runtime for the duration of this run.
     pub(crate) store: ksp_store_lib::Store,
     /// Shareable HTTP Transport pool selected by the active composite profile.
@@ -26,6 +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>>>,
     next_sequence: std::sync::atomic::AtomicU64,
 }
 
@@ -35,6 +38,7 @@ impl crate::BackfillRunState {
     pub(crate) const fn new() -> Self {
         return Self {
             active: std::sync::Mutex::new(std::option::Option::None),
+            last_terminal: std::sync::Mutex::new(std::option::Option::None),
             next_sequence: std::sync::atomic::AtomicU64::new(1),
         };
     }
@@ -107,10 +111,58 @@ impl crate::BackfillRunState {
             },
         };
         let cancellation_requested = current.handle.is_cancellation_requested();
+        let source = current.handle.snapshots();
+        let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
+        if notification.state().is_terminal() {
+            let terminal = self.last_terminal.lock();
+            let mut 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 monitoring state lock is poisoned",
+                    ));
+                },
+            };
+            *terminal = std::option::Option::Some(notification);
+        }
         *active = std::option::Option::None;
         return std::result::Result::Ok(cancellation_requested);
     }
 
+    /// Returns the complete current active snapshot or the retained latest terminal snapshot for resynchronization.
+    pub(crate) fn current_notification(
+        &self,
+    ) -> ksp_core_lib::Result<std::option::Option<ksp_job_api::JobNotification<ksp_job_backfill_lib::BackfillJobSnapshot>>> {
+        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 monitoring state lock is poisoned",
+                ));
+            },
+        };
+        if let std::option::Option::Some(current) = active.as_ref() {
+            let source = current.handle.snapshots();
+            let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
+            return std::result::Result::Ok(std::option::Option::Some(notification));
+        }
+        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 monitoring state lock is poisoned",
+                ));
+            },
+        };
+        return std::result::Result::Ok(terminal.clone());
+    }
+
     /// Rolls back a just-installed admission when execution resources cannot be acquired before spawn.
     pub(crate) fn rollback(&self, job_id: &ksp_job_api::JobId) -> ksp_core_lib::Result<()> {
         let finished = self.finish(job_id);