v0.3.6-pre.003

crates/ksp-job-api/src/error.rs

@@ -1,9 +1,12 @@
 // file: crates/ksp-job-api/src/error.rs
-// version: 1
+// version: 2
 
 /// Error code used when a Job identifier violates its bounded safe-code contract.
 pub const ERROR_CODE_JOB_ID_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_api", "job_id_invalid");
 /// Error code used when a Job kind code violates its bounded safe-code contract.
 pub const ERROR_CODE_JOB_KIND_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_api", "job_kind_invalid");
+/// Error code used when a Job notification sequence cannot advance without wrapping.
+pub const ERROR_CODE_JOB_NOTIFICATION_SEQUENCE_EXHAUSTED: ksp_core_lib::ErrorCode =
+    ksp_core_lib::ErrorCode::new("job_api", "job_notification_sequence_exhausted");
 /// Error code used when a requested Job lifecycle transition is not allowed.
 pub const ERROR_CODE_JOB_TRANSITION_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_api", "job_transition_invalid");

crates/ksp-job-api/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-api/src/lib.rs
-// version: 1
+// version: 2
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -8,14 +8,15 @@
 //! Passive runtime-neutral lifecycle contracts for bounded KSP Jobs.
 //!
 //! This foundation owns validated Job identity, explicit lifecycle transitions
-//! and cooperative cancellation intent. Notification delivery, concrete Job
-//! behavior, runtime spawning, Transport, Store and Worker contracts remain
-//! outside this crate.
+//! cooperative cancellation intent and runtime-neutral latest-value observation.
+//! Concrete Job behavior, runtime spawning, Transport, Store and Worker contracts
+//! remain outside this crate.
 
 mod cancellation;
 mod error;
 mod identity;
 mod lifecycle;
+mod notification;
 
 /// Runtime-neutral cloneable token carrying cooperative cancellation intent.
 pub use self::cancellation::JobCancellationToken;
@@ -23,6 +24,8 @@ pub use self::cancellation::JobCancellationToken;
 pub use self::error::ERROR_CODE_JOB_ID_INVALID;
 /// Error code used when a Job kind code violates its bounded safe-code contract.
 pub use self::error::ERROR_CODE_JOB_KIND_INVALID;
+/// Error code used when a Job notification sequence cannot advance without wrapping.
+pub use self::error::ERROR_CODE_JOB_NOTIFICATION_SEQUENCE_EXHAUSTED;
 /// Error code used when a requested Job lifecycle transition is not allowed.
 pub use self::error::ERROR_CODE_JOB_TRANSITION_INVALID;
 /// Bounded caller-supplied identity of one logical Job and its controlled resumptions.
@@ -39,6 +42,14 @@ pub use self::lifecycle::JobCompletion;
 pub use self::lifecycle::JobLifecycle;
 /// Current lifecycle state of one bounded Job.
 pub use self::lifecycle::JobState;
+/// Latest complete observable value for one Job at a monotone sequence position.
+pub use self::notification::JobNotification;
+/// Monotone sequence attached to one latest-value Job notification stream.
+pub use self::notification::JobNotificationSequence;
+/// Runtime-neutral future returned while observing a latest-value Job snapshot source.
+pub use self::notification::JobSnapshotFuture;
+/// Runtime-neutral read and change-wait contract for one latest-value Job snapshot stream.
+pub use self::notification::JobSnapshotSource;
 /// Common KSP error type used by Job-facing contracts.
 pub use ksp_core_lib::Error;
 /// Stable structured code identifying a KSP error category and condition.

crates/ksp-job-api/src/notification.rs

@@ -0,0 +1,126 @@
+// file: crates/ksp-job-api/src/notification.rs
+// version: 1
+
+/// Monotone sequence attached to one latest-value Job notification stream.
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct JobNotificationSequence(u64);
+
+impl JobNotificationSequence {
+    /// Creates the initial sequence position for one Job snapshot stream.
+    #[must_use]
+    pub const fn initial() -> Self {
+        return Self(0);
+    }
+
+    /// Returns the opaque numeric position carried by this sequence.
+    #[must_use]
+    pub const fn value(&self) -> u64 {
+        return self.0;
+    }
+
+    /// Advances the sequence exactly once or reports exhaustion without wrapping.
+    pub fn next(&self) -> crate::Result<Self> {
+        let next = match self.0.checked_add(1) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => {
+                return std::result::Result::Err(
+                    crate::Error::new(crate::ERROR_CODE_JOB_NOTIFICATION_SEQUENCE_EXHAUSTED, "Job notification sequence exhausted")
+                        .with_context("sequence", self.0.to_string()),
+                );
+            },
+        };
+        return std::result::Result::Ok(Self(next));
+    }
+
+    /// Reports whether this sequence is strictly newer than an observed sequence.
+    #[must_use]
+    pub const fn is_after(&self, observed: Self) -> bool {
+        return self.0 > observed.0;
+    }
+}
+
+/// Latest complete observable value for one Job at a monotone sequence position.
+#[derive(Clone, Eq, PartialEq)]
+pub struct JobNotification<S> {
+    id: crate::JobId,
+    kind: crate::JobKindCode,
+    sequence: crate::JobNotificationSequence,
+    snapshot: S,
+    state: crate::JobState,
+}
+
+impl<S> JobNotification<S> {
+    /// Creates one immutable latest-value notification from an already validated Job identity and snapshot.
+    #[must_use]
+    pub const fn new(id: crate::JobId, kind: crate::JobKindCode, sequence: crate::JobNotificationSequence, state: crate::JobState, snapshot: S) -> Self {
+        return Self { id, kind, sequence, snapshot, state };
+    }
+
+    /// Returns the logical Job identity.
+    #[must_use]
+    pub const fn id(&self) -> &crate::JobId {
+        return &self.id;
+    }
+
+    /// Returns the stable Job family code.
+    #[must_use]
+    pub const fn kind(&self) -> &crate::JobKindCode {
+        return &self.kind;
+    }
+
+    /// Returns the monotone sequence of this latest value.
+    #[must_use]
+    pub const fn sequence(&self) -> crate::JobNotificationSequence {
+        return self.sequence;
+    }
+
+    /// Returns the complete safe snapshot owned by the concrete Job contract.
+    #[must_use]
+    pub const fn snapshot(&self) -> &S {
+        return &self.snapshot;
+    }
+
+    /// Returns the lifecycle state represented by this snapshot.
+    #[must_use]
+    pub const fn state(&self) -> crate::JobState {
+        return self.state;
+    }
+}
+
+impl<S> std::fmt::Debug for JobNotification<S> {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("JobNotification")
+            .field("id", &self.id)
+            .field("kind", &self.kind)
+            .field("sequence", &self.sequence)
+            .field("state", &self.state)
+            .field("snapshot", &"<redacted>")
+            .finish();
+    }
+}
+
+/// Runtime-neutral future returned while observing a latest-value Job snapshot source.
+pub type JobSnapshotFuture<'a, S> = std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = crate::JobNotification<S>> + std::marker::Send + 'a>>;
+
+/// Runtime-neutral read and change-wait contract for one latest-value Job snapshot stream.
+pub trait JobSnapshotSource: std::marker::Send + std::marker::Sync {
+    /// Complete snapshot type retained by the concrete source.
+    type Snapshot: std::clone::Clone + std::marker::Send + std::marker::Sync + 'static;
+
+    /// Returns the complete current value without requiring replay of prior notifications.
+    #[must_use]
+    fn current(&self) -> crate::JobNotification<Self::Snapshot>;
+
+    /// Waits for a value newer than `observed`, returning the complete current snapshot after coalescing any intermediate updates.
+    fn wait_for_change(&self, observed: crate::JobNotificationSequence) -> crate::JobSnapshotFuture<'_, Self::Snapshot>;
+}
+
+#[cfg(test)]
+fn exhausted_notification_sequence() -> crate::JobNotificationSequence {
+    return crate::JobNotificationSequence(u64::MAX);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/notification.rs"]
+mod tests;

crates/ksp-job-api/tests/dependency_boundary.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-api/tests/dependency_boundary.rs
-// version: 1
+// version: 2
 
 //! Dependency and runtime-neutrality canaries for the Job API foundation.
 
@@ -25,13 +25,14 @@ fn pre_002_manifest_has_exact_core_only_dependency_graph() {
 }
 
 #[test]
-fn pre_002_production_sources_forbid_runtime_domain_and_wire_dependencies() {
+fn pre_003_production_sources_forbid_runtime_domain_and_wire_dependencies() {
     let sources = [
         include_str!("../src/cancellation.rs"),
         include_str!("../src/error.rs"),
         include_str!("../src/identity.rs"),
         include_str!("../src/lib.rs"),
         include_str!("../src/lifecycle.rs"),
+        include_str!("../src/notification.rs"),
     ];
     for source in sources {
         for forbidden in [

crates/ksp-job-api/tests/notifications.rs

@@ -0,0 +1,148 @@
+// file: crates/ksp-job-api/tests/notifications.rs
+// version: 1
+
+//! External-consumer canaries for latest-value Job observation.
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+struct TestSnapshot {
+    completed: u64,
+}
+
+#[derive(Clone)]
+struct TestSnapshotSource {
+    current: std::sync::Arc<ksp_job_api::JobNotification<TestSnapshot>>,
+}
+
+impl TestSnapshotSource {
+    fn new(current: ksp_job_api::JobNotification<TestSnapshot>) -> Self {
+        return Self { current: std::sync::Arc::new(current) };
+    }
+}
+
+impl ksp_job_api::JobSnapshotSource for TestSnapshotSource {
+    type Snapshot = TestSnapshot;
+
+    fn current(&self) -> ksp_job_api::JobNotification<Self::Snapshot> {
+        return self.current.as_ref().clone();
+    }
+
+    fn wait_for_change(&self, observed: ksp_job_api::JobNotificationSequence) -> ksp_job_api::JobSnapshotFuture<'_, Self::Snapshot> {
+        let current = ksp_job_api::JobSnapshotSource::current(self);
+        assert!(current.sequence().is_after(observed));
+        return std::boxed::Box::pin(std::future::ready(current));
+    }
+}
+
+struct TestWake;
+
+impl std::task::Wake for TestWake {
+    fn wake(self: std::sync::Arc<Self>) {
+        return;
+    }
+}
+
+fn poll_ready<S>(mut future: ksp_job_api::JobSnapshotFuture<'_, S>) -> std::option::Option<ksp_job_api::JobNotification<S>> {
+    let waker = std::task::Waker::from(std::sync::Arc::new(TestWake));
+    let mut context = std::task::Context::from_waker(&waker);
+    return match std::future::Future::poll(future.as_mut(), &mut context) {
+        std::task::Poll::Ready(value) => std::option::Option::Some(value),
+        std::task::Poll::Pending => std::option::Option::None,
+    };
+}
+
+fn notification_sequence(value: u64) -> std::option::Option<ksp_job_api::JobNotificationSequence> {
+    let mut sequence = ksp_job_api::JobNotificationSequence::initial();
+    for _ in 0..value {
+        sequence = match sequence.next() {
+            std::result::Result::Ok(next) => next,
+            std::result::Result::Err(_) => return std::option::Option::None,
+        };
+    }
+    return std::option::Option::Some(sequence);
+}
+
+fn notification(sequence_value: u64, state: ksp_job_api::JobState, completed: u64) -> std::option::Option<ksp_job_api::JobNotification<TestSnapshot>> {
+    let id = match ksp_job_api::JobId::new("external-job-001") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let kind = match ksp_job_api::JobKindCode::new("backfill_raw") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let sequence = match notification_sequence(sequence_value) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    return std::option::Option::Some(ksp_job_api::JobNotification::new(id, kind, sequence, state, TestSnapshot { completed }));
+}
+
+#[test]
+fn pre_003_external_notification_contract_is_consumable_from_crate_root() {
+    let notification = notification(3, ksp_job_api::JobState::Running, 2);
+    assert!(notification.is_some());
+    let notification = match notification {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert_eq!(notification.id().as_str(), "external-job-001");
+    assert_eq!(notification.kind().as_str(), "backfill_raw");
+    assert_eq!(notification.sequence().value(), 3);
+    assert_eq!(notification.state(), ksp_job_api::JobState::Running);
+    assert_eq!(notification.snapshot().completed, 2);
+    return;
+}
+
+#[test]
+fn pre_003_slow_and_independent_listeners_resynchronize_to_latest_value() {
+    let latest = notification(8, ksp_job_api::JobState::Running, 7);
+    assert!(latest.is_some());
+    let source = match latest {
+        std::option::Option::Some(value) => TestSnapshotSource::new(value),
+        std::option::Option::None => return,
+    };
+    let listener_a = source.clone();
+    let listener_b = source.clone();
+    let observed_a = match notification_sequence(2) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let observed_b = match notification_sequence(6) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let changed_a = poll_ready(ksp_job_api::JobSnapshotSource::wait_for_change(&listener_a, observed_a));
+    let changed_b = poll_ready(ksp_job_api::JobSnapshotSource::wait_for_change(&listener_b, observed_b));
+    assert!(changed_a.is_some());
+    assert!(changed_b.is_some());
+    let changed_a = match changed_a {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let changed_b = match changed_b {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert_eq!(changed_a.sequence().value(), 8);
+    assert_eq!(changed_b.sequence().value(), 8);
+    assert_eq!(changed_a.snapshot().completed, 7);
+    assert_eq!(changed_b.snapshot().completed, 7);
+    return;
+}
+
+#[test]
+fn pre_003_terminal_snapshot_remains_readable_from_shared_source() {
+    let terminal = notification(9, ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Partial), 8);
+    assert!(terminal.is_some());
+    let source = match terminal {
+        std::option::Option::Some(value) => TestSnapshotSource::new(value),
+        std::option::Option::None => return,
+    };
+    let cloned = source.clone();
+    let current = ksp_job_api::JobSnapshotSource::current(&cloned);
+    assert_eq!(current.sequence().value(), 9);
+    assert_eq!(current.state(), ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Partial));
+    assert!(current.state().is_terminal());
+    assert_eq!(current.snapshot().completed, 8);
+    return;
+}

crates/ksp-job-api/tests/release_completeness.rs

@@ -1,10 +1,10 @@
 // file: crates/ksp-job-api/tests/release_completeness.rs
-// version: 1
+// version: 2
 
-//! Completeness canaries for the initial Job API foundation.
+//! Completeness canaries for the current Job API foundation.
 
 #[test]
-fn pre_002_crate_root_export_inventory_is_exact() {
+fn pre_003_crate_root_export_inventory_is_exact() {
     let crate_root = include_str!("../src/lib.rs");
     let mut actual = std::vec::Vec::new();
     for line in crate_root.lines() {
@@ -18,6 +18,7 @@ fn pre_002_crate_root_export_inventory_is_exact() {
         "pub use self::cancellation::JobCancellationToken;",
         "pub use self::error::ERROR_CODE_JOB_ID_INVALID;",
         "pub use self::error::ERROR_CODE_JOB_KIND_INVALID;",
+        "pub use self::error::ERROR_CODE_JOB_NOTIFICATION_SEQUENCE_EXHAUSTED;",
         "pub use self::error::ERROR_CODE_JOB_TRANSITION_INVALID;",
         "pub use self::identity::JobId;",
         "pub use self::identity::JobKindCode;",
@@ -26,6 +27,10 @@ fn pre_002_crate_root_export_inventory_is_exact() {
         "pub use self::lifecycle::JobCompletion;",
         "pub use self::lifecycle::JobLifecycle;",
         "pub use self::lifecycle::JobState;",
+        "pub use self::notification::JobNotification;",
+        "pub use self::notification::JobNotificationSequence;",
+        "pub use self::notification::JobSnapshotFuture;",
+        "pub use self::notification::JobSnapshotSource;",
         "pub use ksp_core_lib::Error;",
         "pub use ksp_core_lib::ErrorCode;",
         "pub use ksp_core_lib::ErrorContext;",
@@ -38,7 +43,7 @@ fn pre_002_crate_root_export_inventory_is_exact() {
 }
 
 #[test]
-fn pre_002_production_module_inventory_is_exact() -> std::io::Result<()> {
+fn pre_003_production_module_inventory_is_exact() -> std::io::Result<()> {
     let source_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
     let entries = match std::fs::read_dir(source_root) {
         std::result::Result::Ok(value) => value,
@@ -66,35 +71,29 @@ fn pre_002_production_module_inventory_is_exact() -> std::io::Result<()> {
         }
     }
     names.sort_unstable();
-    assert_eq!(names, std::vec!["cancellation.rs", "error.rs", "identity.rs", "lib.rs", "lifecycle.rs"]);
+    assert_eq!(names, std::vec!["cancellation.rs", "error.rs", "identity.rs", "lib.rs", "lifecycle.rs", "notification.rs"]);
     return std::result::Result::Ok(());
 }
 
 #[test]
-fn pre_002_surface_does_not_open_notifications_backfill_or_worker_contracts() {
+fn pre_003_surface_opens_only_generic_notifications_without_backfill_or_worker_contracts() {
     let sources = [
         include_str!("../src/cancellation.rs"),
         include_str!("../src/identity.rs"),
         include_str!("../src/lib.rs"),
         include_str!("../src/lifecycle.rs"),
+        include_str!("../src/notification.rs"),
     ];
     for source in sources {
-        for forbidden in [
-            "JobNotification",
-            "JobNotificationSequence",
-            "JobSnapshot",
-            "JobSnapshotSource",
-            "BackfillRequest",
-            "WorkerControl",
-            "WorkerState",
-            "spawn(",
-            "RawTransaction",
-            "provider",
-            "endpoint",
-        ] {
+        for forbidden in ["BackfillRequest", "WorkerControl", "WorkerState", "spawn(", "RawTransaction", "provider", "endpoint"] {
             assert!(!source.contains(forbidden), "future or domain-specific Job contract leaked early: {forbidden}");
         }
     }
+    let notification_source = include_str!("../src/notification.rs");
+    assert!(notification_source.contains("pub struct JobNotification<S>"));
+    assert!(notification_source.contains("pub trait JobSnapshotSource"));
+    assert!(notification_source.contains("std::future::Future"));
+    assert!(!notification_source.contains("tokio::"));
     let lifecycle_source = include_str!("../src/lifecycle.rs");
     assert!(lifecycle_source.contains("#[derive(Eq, PartialEq)]\npub struct JobLifecycle"));
     assert!(!lifecycle_source.contains("#[derive(Clone, Eq, PartialEq)]\npub struct JobLifecycle"));

crates/ksp-job-api/unit_tests/notification.rs

@@ -0,0 +1,81 @@
+// file: crates/ksp-job-api/unit_tests/notification.rs
+// version: 1
+
+fn sequence(value: u64) -> crate::JobNotificationSequence {
+    let mut sequence = crate::JobNotificationSequence::initial();
+    for _ in 0..value {
+        sequence = match sequence.next() {
+            std::result::Result::Ok(next) => next,
+            std::result::Result::Err(_) => return sequence,
+        };
+    }
+    return sequence;
+}
+
+#[test]
+fn pre_003_notification_sequence_advances_strictly_and_orders_positions() {
+    let mut first = crate::JobNotificationSequence::initial();
+    for _ in 0..41 {
+        let next = first.next();
+        assert!(next.is_ok());
+        first = match next {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => return,
+        };
+    }
+    let second = first.next();
+    assert!(second.is_ok());
+    let second = match second {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert_eq!(first.value(), 41);
+    assert_eq!(second.value(), 42);
+    assert!(second.is_after(first));
+    assert!(!first.is_after(second));
+    assert!(second > first);
+    return;
+}
+
+#[test]
+fn pre_003_notification_sequence_exhaustion_is_explicit_and_non_wrapping() {
+    let exhausted = super::exhausted_notification_sequence().next();
+    assert!(exhausted.is_err());
+    let error = match exhausted {
+        std::result::Result::Err(value) => value,
+        std::result::Result::Ok(_) => return,
+    };
+    assert_eq!(error.code(), crate::ERROR_CODE_JOB_NOTIFICATION_SEQUENCE_EXHAUSTED);
+    assert_eq!(error.context().len(), 1);
+    assert_eq!(error.context()[0].key(), "sequence");
+    assert_eq!(error.context()[0].value(), u64::MAX.to_string());
+    return;
+}
+
+#[test]
+fn pre_003_notification_preserves_complete_value_and_redacts_snapshot_debug() {
+    let id = crate::JobId::new("job-notify-001");
+    let kind = crate::JobKindCode::new("backfill_raw");
+    assert!(id.is_ok());
+    assert!(kind.is_ok());
+    let id = match id {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let kind = match kind {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let notification =
+        crate::JobNotification::new(id.clone(), kind.clone(), sequence(7), crate::JobState::Running, "RAW-PAYLOAD-MUST-NOT-APPEAR-IN-DEBUG".to_string());
+    assert_eq!(notification.id(), &id);
+    assert_eq!(notification.kind(), &kind);
+    assert_eq!(notification.sequence().value(), 7);
+    assert_eq!(notification.state(), crate::JobState::Running);
+    assert_eq!(notification.snapshot(), "RAW-PAYLOAD-MUST-NOT-APPEAR-IN-DEBUG");
+    let debug = std::format!("{notification:?}");
+    assert!(debug.contains("JobNotification"));
+    assert!(debug.contains("<redacted>"));
+    assert!(!debug.contains("RAW-PAYLOAD-MUST-NOT-APPEAR-IN-DEBUG"));
+    return;
+}