v0.3.6-pre.008

crates/ksp-job-backfill-lib/src/checkpoint.rs

@@ -0,0 +1,218 @@
+// file: crates/ksp-job-backfill-lib/src/checkpoint.rs
+// version: 1
+
+/// Opaque caller-owned checkpoint for one controlled Backfill resumption.
+///
+/// The checkpoint carries no payload, endpoint, provider, URL or secret. Persistence of this
+/// value is deliberately external to Store in v0.3.6; the Backfill library only validates and
+/// consumes checkpoints supplied back by its caller.
+#[derive(Clone, Eq, PartialEq)]
+pub struct BackfillCheckpoint {
+    job_id: ksp_job_api::JobId,
+    scope_fingerprint: crate::BackfillScopeFingerprint,
+    completed_prefix: usize,
+    resume_before: std::option::Option<crate::BackfillSignature>,
+}
+
+impl BackfillCheckpoint {
+    /// Returns the logical Job identity that owns this checkpoint.
+    #[must_use]
+    pub const fn job_id(&self) -> &ksp_job_api::JobId {
+        return &self.job_id;
+    }
+
+    /// Returns the semantic scope fingerprint bound to this checkpoint.
+    #[must_use]
+    pub const fn scope_fingerprint(&self) -> crate::BackfillScopeFingerprint {
+        return self.scope_fingerprint;
+    }
+
+    /// Returns the number of candidates proven durable in one contiguous prefix.
+    #[must_use]
+    pub const fn completed_prefix(&self) -> usize {
+        return self.completed_prefix;
+    }
+
+    /// Creates one internally proven checkpoint.
+    pub(crate) fn new(
+        job_id: ksp_job_api::JobId,
+        scope_fingerprint: crate::BackfillScopeFingerprint,
+        completed_prefix: usize,
+        resume_before: std::option::Option<crate::BackfillSignature>,
+    ) -> Self {
+        return Self { job_id, scope_fingerprint, completed_prefix, resume_before };
+    }
+
+    /// Returns the internal exclusive `before` cursor used only by controlled Before resumption.
+    pub(crate) const fn resume_before(&self) -> std::option::Option<&crate::BackfillSignature> {
+        return self.resume_before.as_ref();
+    }
+}
+
+impl std::fmt::Debug for BackfillCheckpoint {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("BackfillCheckpoint")
+            .field("job_id", &self.job_id)
+            .field("scope_fingerprint", &self.scope_fingerprint)
+            .field("completed_prefix", &self.completed_prefix)
+            .field("has_resume_before", &self.resume_before.is_some())
+            .finish();
+    }
+}
+
+/// Private bounded bitmap tracking durable completions and their contiguous prefix.
+pub(crate) struct CompletionFrontier {
+    durable: std::vec::Vec<bool>,
+    contiguous_completed: usize,
+}
+
+impl CompletionFrontier {
+    /// Creates one empty frontier for the exact bounded candidate count.
+    pub(crate) fn new(candidate_count: usize) -> Self {
+        return Self { durable: vec![false; candidate_count], contiguous_completed: 0 };
+    }
+
+    /// Seeds a previously proven replay prefix before processing the remaining candidates.
+    pub(crate) fn seed_prefix(&mut self, completed_prefix: usize) -> ksp_core_lib::Result<()> {
+        if completed_prefix > self.durable.len() {
+            return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix"));
+        }
+        for index in 0..completed_prefix {
+            self.durable[index] = true;
+        }
+        self.contiguous_completed = completed_prefix;
+        return std::result::Result::Ok(());
+    }
+
+    /// Marks one candidate durable and advances only through the now-contiguous prefix.
+    pub(crate) fn mark_durable(&mut self, index: usize) -> ksp_core_lib::Result<()> {
+        let slot = match self.durable.get_mut(index) {
+            std::option::Option::Some(slot) => slot,
+            std::option::Option::None => return std::result::Result::Err(checkpoint_error("frontier.index")),
+        };
+        *slot = true;
+        while self.contiguous_completed < self.durable.len() && self.durable[self.contiguous_completed] {
+            self.contiguous_completed += 1;
+        }
+        return std::result::Result::Ok(());
+    }
+
+    /// Returns the number of durable candidates in the contiguous prefix.
+    pub(crate) const fn contiguous_completed(&self) -> usize {
+        return self.contiguous_completed;
+    }
+}
+
+/// Validates one checkpoint against the Job, semantic scope and scope-specific cursor shape.
+pub(crate) fn validate_request_checkpoint(
+    request_job_id: &ksp_job_api::JobId,
+    scope_fingerprint: crate::BackfillScopeFingerprint,
+    scope_kind: crate::BackfillScopeKind,
+    checkpoint: &BackfillCheckpoint,
+) -> ksp_core_lib::Result<()> {
+    if checkpoint.job_id() != request_job_id {
+        return std::result::Result::Err(checkpoint_error("checkpoint.job_id"));
+    }
+    if checkpoint.scope_fingerprint() != scope_fingerprint {
+        return std::result::Result::Err(checkpoint_error("checkpoint.scope_fingerprint"));
+    }
+    if scope_kind != crate::BackfillScopeKind::BeforeAddress && checkpoint.resume_before().is_some() {
+        return std::result::Result::Err(checkpoint_error("checkpoint.resume_before"));
+    }
+    return std::result::Result::Ok(());
+}
+
+/// Resolves the exclusive Before cursor from a validated checkpoint or the original scope anchor.
+pub(crate) fn resume_before_cursor(request: &crate::BackfillRequest) -> std::option::Option<std::string::String> {
+    if request.scope().kind() != crate::BackfillScopeKind::BeforeAddress {
+        return std::option::Option::None;
+    }
+    if let std::option::Option::Some(checkpoint) = request.checkpoint()
+        && let std::option::Option::Some(cursor) = checkpoint.resume_before()
+    {
+        return std::option::Option::Some(cursor.as_str().to_owned());
+    }
+    return request.scope().anchor().map(|anchor| return anchor.as_str().to_owned());
+}
+
+/// Resolves the replay prefix skipped only by After and Explicit execution.
+pub(crate) fn execution_resume_offset(request: &crate::BackfillRequest, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<usize> {
+    let checkpoint = match request.checkpoint() {
+        std::option::Option::Some(checkpoint) => checkpoint,
+        std::option::Option::None => return std::result::Result::Ok(0),
+    };
+    return match request.scope().kind() {
+        crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => {
+            if checkpoint.completed_prefix() > discovery.candidates().len() {
+                return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix"));
+            }
+            std::result::Result::Ok(checkpoint.completed_prefix())
+        },
+        crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => std::result::Result::Ok(0),
+    };
+}
+
+/// Builds the safe next checkpoint from one drained contiguous completion frontier.
+pub(crate) fn checkpoint_from_frontier(
+    request: &crate::BackfillRequest,
+    discovery: &crate::BackfillDiscovery,
+    frontier: &CompletionFrontier,
+) -> ksp_core_lib::Result<BackfillCheckpoint> {
+    let validation = validate_discovery_identity(request, discovery);
+    if let std::result::Result::Err(error) = validation {
+        return std::result::Result::Err(error);
+    }
+    let local_prefix = frontier.contiguous_completed();
+    if request.scope().kind() == crate::BackfillScopeKind::AfterAddress && discovery.boundary() == crate::BackfillDiscoveryBoundary::AfterAnchorNotReached {
+        if let std::option::Option::Some(checkpoint) = request.checkpoint() {
+            return std::result::Result::Ok(checkpoint.clone());
+        }
+        return std::result::Result::Ok(BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), 0, std::option::Option::None));
+    }
+    let (completed_prefix, resume_before) = match request.scope().kind() {
+        crate::BackfillScopeKind::LatestAddress => (local_prefix, std::option::Option::None),
+        crate::BackfillScopeKind::BeforeAddress => {
+            let previous = request.checkpoint().map_or(0, BackfillCheckpoint::completed_prefix);
+            let completed_prefix = match previous.checked_add(local_prefix) {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix")),
+            };
+            let resume_before = if local_prefix == 0 {
+                match request.checkpoint() {
+                    std::option::Option::Some(checkpoint) => checkpoint.resume_before().cloned(),
+                    std::option::Option::None => request.scope().anchor().cloned(),
+                }
+            } else {
+                let index = local_prefix - 1;
+                let candidate = match discovery.candidates().get(index) {
+                    std::option::Option::Some(candidate) => candidate,
+                    std::option::Option::None => return std::result::Result::Err(checkpoint_error("frontier.index")),
+                };
+                std::option::Option::Some(candidate.identity().signature().clone())
+            };
+            (completed_prefix, resume_before)
+        },
+        crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => (local_prefix, std::option::Option::None),
+    };
+    return std::result::Result::Ok(BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), completed_prefix, resume_before));
+}
+
+/// Validates that one discovery result belongs to the request network and semantic scope.
+pub(crate) fn validate_discovery_identity(request: &crate::BackfillRequest, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<()> {
+    if discovery.network() != request.network() {
+        return std::result::Result::Err(checkpoint_error("discovery.network"));
+    }
+    if discovery.scope_fingerprint() != request.scope_fingerprint() {
+        return std::result::Result::Err(checkpoint_error("discovery.scope_fingerprint"));
+    }
+    return std::result::Result::Ok(());
+}
+
+fn checkpoint_error(field: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CHECKPOINT_INVALID, "invalid Backfill checkpoint/frontier state").with_context("field", field);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/checkpoint.rs"]
+mod tests;

crates/ksp-job-backfill-lib/src/discovery.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-backfill-lib/src/discovery.rs
-// version: 2
+// version: 3
 
 /// Network-scoped identity of one discovered transaction candidate before canonical signature decoding.
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
@@ -100,6 +100,17 @@ pub struct BackfillDiscovery {
 }
 
 impl BackfillDiscovery {
+    /// Creates one internally validated bounded discovery value.
+    pub(crate) fn new(
+        network: ksp_store_lib::RawNetworkId,
+        scope_fingerprint: crate::BackfillScopeFingerprint,
+        candidates: std::vec::Vec<BackfillCandidate>,
+        pages_fetched: usize,
+        boundary: BackfillDiscoveryBoundary,
+    ) -> Self {
+        return Self { network, scope_fingerprint, candidates, pages_fetched, boundary };
+    }
+
     /// Returns the logical network shared by every candidate identity.
     #[must_use]
     pub const fn network(&self) -> &ksp_store_lib::RawNetworkId {
@@ -244,13 +255,13 @@ fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result<B
         let identity = BackfillCandidateIdentity::new(request.network().clone(), signature.clone());
         candidates.push(BackfillCandidate::new(identity, std::option::Option::None));
     }
-    return std::result::Result::Ok(BackfillDiscovery {
-        network: request.network().clone(),
-        scope_fingerprint: request.scope_fingerprint(),
+    return std::result::Result::Ok(BackfillDiscovery::new(
+        request.network().clone(),
+        request.scope_fingerprint(),
         candidates,
-        pages_fetched: 0,
-        boundary: BackfillDiscoveryBoundary::ExplicitInput,
-    });
+        0,
+        BackfillDiscoveryBoundary::ExplicitInput,
+    ));
 }
 
 async fn discover_older<S>(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery>
@@ -262,7 +273,7 @@ where
         std::option::Option::None => return std::result::Result::Err(discovery_invalid("scope.address")),
     };
     let mut before = match request.scope().kind() {
-        crate::BackfillScopeKind::BeforeAddress => request.scope().anchor().map(|value| return value.as_str().to_owned()),
+        crate::BackfillScopeKind::BeforeAddress => crate::resume_before_cursor(request),
         crate::BackfillScopeKind::LatestAddress => std::option::Option::None,
         crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => {
             return std::result::Result::Err(discovery_invalid("scope.kind"));
@@ -333,13 +344,7 @@ where
         }
         before = std::option::Option::Some(next_before.as_str().to_owned());
     };
-    return std::result::Result::Ok(BackfillDiscovery {
-        network: request.network().clone(),
-        scope_fingerprint: request.scope_fingerprint(),
-        candidates,
-        pages_fetched,
-        boundary,
-    });
+    return std::result::Result::Ok(BackfillDiscovery::new(request.network().clone(), request.scope_fingerprint(), candidates, pages_fetched, boundary));
 }
 
 async fn discover_after<S>(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery>
@@ -416,13 +421,13 @@ where
         }
         before = std::option::Option::Some(next_before.as_str().to_owned());
     };
-    return std::result::Result::Ok(BackfillDiscovery {
-        network: request.network().clone(),
-        scope_fingerprint: request.scope_fingerprint(),
-        candidates: nearest.into_iter().collect(),
+    return std::result::Result::Ok(BackfillDiscovery::new(
+        request.network().clone(),
+        request.scope_fingerprint(),
+        nearest.into_iter().collect(),
         pages_fetched,
         boundary,
-    });
+    ));
 }
 
 fn validated_signature(value: &str) -> ksp_core_lib::Result<crate::BackfillSignature> {

crates/ksp-job-backfill-lib/src/error.rs

@@ -1,10 +1,14 @@
 // file: crates/ksp-job-backfill-lib/src/error.rs
-// version: 3
+// version: 4
 
+/// Error code used when one Backfill checkpoint/frontier is incompatible with the current Job or semantic scope.
+pub const ERROR_CODE_BACKFILL_CHECKPOINT_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "checkpoint_invalid");
 /// Error code used when a signature page violates a bounded discovery invariant.
 pub const ERROR_CODE_BACKFILL_DISCOVERY_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "discovery_invalid");
 /// Error code used when paginated discovery cannot advance its exclusive RPC cursor safely.
 pub const ERROR_CODE_BACKFILL_DISCOVERY_STALLED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "discovery_stalled");
+/// Error code used when bounded concurrent candidate execution violates its internal admission or clock invariants.
+pub const ERROR_CODE_BACKFILL_EXECUTION_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "execution_invalid");
 /// Error code used when Store persistence returns an impossible Backfill state or targets a different network.
 pub const ERROR_CODE_BACKFILL_PERSISTENCE_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "persistence_invalid");
 /// Error code used when deterministic Transport-to-RAW conversion violates the v1 contract.

crates/ksp-job-backfill-lib/src/execution.rs

@@ -0,0 +1,272 @@
+// file: crates/ksp-job-backfill-lib/src/execution.rs
+// version: 1
+
+use futures_util::StreamExt; // rust-rules: trait-import
+
+/// Bounded result of one concurrent Backfill candidate execution pass.
+///
+/// This is intentionally smaller than the concrete latest-value snapshot planned for `pre.009`.
+/// It exposes only the admission/frontier/checkpoint facts required to prove safe resumption.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct BackfillExecutionBatch {
+    candidate_count: usize,
+    admitted_count: usize,
+    finished_count: usize,
+    durable_count: usize,
+    hole_count: usize,
+    maximum_in_flight: usize,
+    local_contiguous_completed: usize,
+    discovery_partial: bool,
+    checkpoint: crate::BackfillCheckpoint,
+    failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
+}
+
+impl BackfillExecutionBatch {
+    /// Returns the number of candidates present in the bounded discovery result.
+    #[must_use]
+    pub const fn candidate_count(&self) -> usize {
+        return self.candidate_count;
+    }
+
+    /// Returns the number of candidates admitted into hydration during this execution pass.
+    #[must_use]
+    pub const fn admitted_count(&self) -> usize {
+        return self.admitted_count;
+    }
+
+    /// Returns the number of admitted candidates whose hydration/persistence future reached a known outcome.
+    #[must_use]
+    pub const fn finished_count(&self) -> usize {
+        return self.finished_count;
+    }
+
+    /// Returns the number of candidates whose Store outcome is durable for checkpoint advancement.
+    #[must_use]
+    pub const fn durable_count(&self) -> usize {
+        return self.durable_count;
+    }
+
+    /// Returns the number of known candidate outcomes that intentionally block the contiguous frontier.
+    #[must_use]
+    pub const fn hole_count(&self) -> usize {
+        return self.hole_count;
+    }
+
+    /// Returns the greatest number of candidate futures simultaneously in flight.
+    #[must_use]
+    pub const fn maximum_in_flight(&self) -> usize {
+        return self.maximum_in_flight;
+    }
+
+    /// Returns the durable contiguous prefix completed inside the current discovery result.
+    #[must_use]
+    pub const fn local_contiguous_completed(&self) -> usize {
+        return self.local_contiguous_completed;
+    }
+
+    /// Returns the safe caller-owned checkpoint after draining all work already admitted.
+    #[must_use]
+    pub const fn checkpoint(&self) -> &crate::BackfillCheckpoint {
+        return &self.checkpoint;
+    }
+
+    /// Returns the first stable fatal error code that stopped new admissions, when one occurred.
+    #[must_use]
+    pub const fn failure_code(&self) -> std::option::Option<ksp_core_lib::ErrorCode> {
+        return self.failure_code;
+    }
+
+    /// Returns whether discovery or candidate outcomes left the pass incomplete.
+    #[must_use]
+    pub const fn is_partial(&self) -> bool {
+        return self.discovery_partial || self.hole_count != 0 || self.failure_code.is_some();
+    }
+}
+
+/// Executes one bounded discovered candidate set with request-owned hydration concurrency.
+///
+/// At most `request.hydration_concurrency()` candidate futures are kept in flight. Candidate
+/// completions may arrive out of order, but the checkpoint advances only across the contiguous
+/// durable prefix. `Missing` remains a non-fatal hole. Conflict or any Transport/conversion/Store
+/// error stops new admissions immediately; work already admitted is drained before this function
+/// returns so a submitted Store operation is never silently abandoned by the batch coordinator.
+pub async fn execute_backfill_discovery(
+    transport: &ksp_onchain_transport_lib::HttpTransportPool,
+    store: &ksp_store_lib::Store,
+    request: &crate::BackfillRequest,
+    discovery: &crate::BackfillDiscovery,
+) -> ksp_core_lib::Result<BackfillExecutionBatch> {
+    let validation = crate::validate_discovery_identity(request, discovery);
+    if let std::result::Result::Err(error) = validation {
+        return std::result::Result::Err(error);
+    }
+    let processor = RuntimeCandidateProcessor { transport, store, request };
+    return execute_with_processor(&processor, request, discovery).await;
+}
+
+type CandidateProcessFuture<'a> =
+    std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ksp_core_lib::Result<crate::BackfillPersistenceOutcome>> + 'a>>;
+
+trait CandidateProcessor {
+    fn process<'a>(&'a self, candidate: &'a crate::BackfillCandidate) -> CandidateProcessFuture<'a>;
+}
+
+struct RuntimeCandidateProcessor<'a> {
+    transport: &'a ksp_onchain_transport_lib::HttpTransportPool,
+    store: &'a ksp_store_lib::Store,
+    request: &'a crate::BackfillRequest,
+}
+
+impl CandidateProcessor for RuntimeCandidateProcessor<'_> {
+    fn process<'a>(&'a self, candidate: &'a crate::BackfillCandidate) -> CandidateProcessFuture<'a> {
+        return std::boxed::Box::pin(async move {
+            let received_at = match current_raw_timestamp() {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            let hydration = crate::hydrate_backfill_candidate(self.transport, self.request, candidate, received_at).await;
+            let hydration = match hydration {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            return crate::persist_backfill_hydration(self.store, hydration).await;
+        });
+    }
+}
+
+async fn execute_with_processor<P>(
+    processor: &P,
+    request: &crate::BackfillRequest,
+    discovery: &crate::BackfillDiscovery,
+) -> ksp_core_lib::Result<BackfillExecutionBatch>
+where
+    P: CandidateProcessor,
+{
+    let validation = crate::validate_discovery_identity(request, discovery);
+    if let std::result::Result::Err(error) = validation {
+        return std::result::Result::Err(error);
+    }
+    let resume_offset = match crate::execution_resume_offset(request, discovery) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut frontier = crate::CompletionFrontier::new(discovery.candidates().len());
+    if matches!(request.scope().kind(), crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures) {
+        let seeded = frontier.seed_prefix(resume_offset);
+        if let std::result::Result::Err(error) = seeded {
+            return std::result::Result::Err(error);
+        }
+    }
+    let mut next_index = resume_offset;
+    let mut admitted_count = 0_usize;
+    let mut finished_count = 0_usize;
+    let mut durable_count = 0_usize;
+    let mut hole_count = 0_usize;
+    let mut maximum_in_flight = 0_usize;
+    let mut failure_code = std::option::Option::<ksp_core_lib::ErrorCode>::None;
+    let mut in_flight = futures_util::stream::FuturesUnordered::new();
+    loop {
+        while failure_code.is_none() && next_index < discovery.candidates().len() && in_flight.len() < request.hydration_concurrency() {
+            let index = next_index;
+            let candidate = &discovery.candidates()[index];
+            let future = processor.process(candidate);
+            in_flight.push(async move {
+                return (index, future.await);
+            });
+            next_index += 1;
+            admitted_count = match checked_increment(admitted_count, "admitted_count") {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            maximum_in_flight = std::cmp::max(maximum_in_flight, in_flight.len());
+        }
+        let completed = in_flight.next().await;
+        let (index, result) = match completed {
+            std::option::Option::Some(completed) => completed,
+            std::option::Option::None => break,
+        };
+        finished_count = match checked_increment(finished_count, "finished_count") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        match result {
+            std::result::Result::Ok(outcome) => {
+                if persistence_advances_frontier(&outcome) {
+                    let marked = frontier.mark_durable(index);
+                    if let std::result::Result::Err(error) = marked {
+                        return std::result::Result::Err(error);
+                    }
+                    durable_count = match checked_increment(durable_count, "durable_count") {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => return std::result::Result::Err(error),
+                    };
+                } else {
+                    hole_count = match checked_increment(hole_count, "hole_count") {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => return std::result::Result::Err(error),
+                    };
+                    if outcome.entity() == crate::BackfillEntityPersistence::Conflict && failure_code.is_none() {
+                        failure_code = std::option::Option::Some(ksp_store_lib::ERROR_CODE_RAW_CONFLICT);
+                    }
+                }
+            },
+            std::result::Result::Err(error) => {
+                hole_count = match checked_increment(hole_count, "hole_count") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(increment_error) => return std::result::Result::Err(increment_error),
+                };
+                if failure_code.is_none() {
+                    failure_code = std::option::Option::Some(error.code());
+                }
+            },
+        }
+    }
+    let checkpoint = match crate::checkpoint_from_frontier(request, discovery, &frontier) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return std::result::Result::Ok(BackfillExecutionBatch {
+        candidate_count: discovery.candidates().len(),
+        admitted_count,
+        finished_count,
+        durable_count,
+        hole_count,
+        maximum_in_flight,
+        local_contiguous_completed: frontier.contiguous_completed(),
+        discovery_partial: discovery.is_partial(),
+        checkpoint,
+        failure_code,
+    });
+}
+
+fn persistence_advances_frontier(outcome: &crate::BackfillPersistenceOutcome) -> bool {
+    return matches!(
+        outcome.entity(),
+        crate::BackfillEntityPersistence::Inserted | crate::BackfillEntityPersistence::AlreadyPresent | crate::BackfillEntityPersistence::SkippedPurged
+    );
+}
+
+fn checked_increment(value: usize, field: &'static str) -> ksp_core_lib::Result<usize> {
+    return value.checked_add(1).ok_or_else(|| return execution_error(field));
+}
+
+fn current_raw_timestamp() -> ksp_core_lib::Result<ksp_store_lib::RawTimestamp> {
+    let duration = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
+    let duration = match duration {
+        std::result::Result::Ok(duration) => duration,
+        std::result::Result::Err(_) => return std::result::Result::Err(execution_error("clock.before_epoch")),
+    };
+    let millis = match u64::try_from(duration.as_millis()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(execution_error("clock.millis")),
+    };
+    return ksp_store_lib::RawTimestamp::from_unix_millis(millis);
+}
+
+fn execution_error(field: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_EXECUTION_INVALID, "invalid bounded Backfill execution state").with_context("field", field);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/execution.rs"]
+mod tests;

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-backfill-lib/src/lib.rs
-// version: 3
+// version: 4
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,16 +10,21 @@
 //! This tranche owns explicit admission, network-scoped candidate identity, deterministic
 //! `getSignaturesForAddress` pagination and canonical RAW v1 conversion through observed
 //! `getTransaction`. Transport retains provider/endpoint selection and retry; Store retains
-//! durable idempotence through the atomic Store facade. Concurrency, checkpointing, cancellation
-//! and concrete latest-value snapshots are added by later v0.3.6 tranches.
+//! durable idempotence through the atomic Store facade. This tranche also owns bounded concurrent
+//! candidate execution and caller-owned contiguous checkpoints. Cancellation and concrete latest-value
+//! snapshots are added by later v0.3.6 tranches.
 
+mod checkpoint;
 mod constants;
 mod conversion;
 mod discovery;
 mod error;
+mod execution;
 mod persistence;
 mod request;
 
+/// Opaque caller-owned checkpoint for one controlled Backfill resumption.
+pub use self::checkpoint::BackfillCheckpoint;
 /// Result of hydrating one deterministic candidate through observed `getTransaction`.
 pub use self::conversion::BackfillHydrationOutcome;
 /// Complete in-memory RAW transaction acquisition ready for later Store persistence.
@@ -40,10 +45,14 @@ pub use self::discovery::BackfillDiscovery;
 pub use self::discovery::BackfillDiscoveryBoundary;
 /// Discovers one bounded deterministic candidate set through the typed KSP Transport wrapper.
 pub use self::discovery::discover_backfill_candidates;
+/// Error code used when one checkpoint/frontier is incompatible with the current Job or semantic scope.
+pub use self::error::ERROR_CODE_BACKFILL_CHECKPOINT_INVALID;
 /// Error code used when a signature page violates a bounded discovery invariant.
 pub use self::error::ERROR_CODE_BACKFILL_DISCOVERY_INVALID;
 /// Error code used when paginated discovery cannot advance its exclusive RPC cursor safely.
 pub use self::error::ERROR_CODE_BACKFILL_DISCOVERY_STALLED;
+/// Error code used when bounded concurrent execution violates an internal admission or clock invariant.
+pub use self::error::ERROR_CODE_BACKFILL_EXECUTION_INVALID;
 /// Error code used when Store persistence returns an impossible Backfill state or targets a different network.
 pub use self::error::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID;
 /// Error code used when deterministic Transport-to-RAW conversion violates the v1 contract.
@@ -52,6 +61,10 @@ pub use self::error::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID;
 pub use self::error::ERROR_CODE_BACKFILL_REQUEST_INVALID;
 /// Error code used when one transaction signature text violates the bounded Base58-shape contract.
 pub use self::error::ERROR_CODE_BACKFILL_SIGNATURE_INVALID;
+/// Bounded result of one concurrent Backfill candidate execution pass.
+pub use self::execution::BackfillExecutionBatch;
+/// Executes one bounded discovered candidate set with request-owned hydration concurrency.
+pub use self::execution::execute_backfill_discovery;
 /// Canonical entity disposition produced by one Backfill Store persistence attempt.
 pub use self::persistence::BackfillEntityPersistence;
 /// Observation disposition produced by one Backfill Store persistence attempt.
@@ -85,6 +98,18 @@ pub use self::request::MAX_BACKFILL_SIGNATURE_TEXT_BYTES;
 /// Minimum Base58 text length possible for one canonical 64-byte Solana signature.
 pub use self::request::MIN_BACKFILL_SIGNATURE_TEXT_BYTES;
 
+/// Internal contiguous completion frontier used by bounded execution.
+pub(crate) use self::checkpoint::CompletionFrontier;
+/// Builds one safe caller-owned checkpoint from the current contiguous frontier.
+pub(crate) use self::checkpoint::checkpoint_from_frontier;
+/// Resolves the candidate offset skipped by one validated replay checkpoint.
+pub(crate) use self::checkpoint::execution_resume_offset;
+/// Resolves the internal exclusive Before cursor for discovery resumption.
+pub(crate) use self::checkpoint::resume_before_cursor;
+/// Validates that one discovery result belongs to the current request identity.
+pub(crate) use self::checkpoint::validate_discovery_identity;
+/// Validates one caller-owned checkpoint against Job and semantic scope identity.
+pub(crate) use self::checkpoint::validate_request_checkpoint;
 /// Owning tracing target used by the concrete Backfill runtime.
 pub(crate) use self::constants::TRACING_TARGET;
 /// Exact private Base58 decoder shared by the public signature wrapper and hydration path.

crates/ksp-job-backfill-lib/src/persistence.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-backfill-lib/src/persistence.rs
-// version: 1
+// version: 2
 
 /// Canonical entity disposition produced by one Backfill Store persistence attempt.
 #[non_exhaustive]
@@ -43,6 +43,15 @@ pub struct BackfillPersistenceOutcome {
 }
 
 impl BackfillPersistenceOutcome {
+    /// Creates one internally classified persistence result.
+    pub(crate) fn new(
+        reference: ksp_store_lib::RawTransactionReference,
+        entity: BackfillEntityPersistence,
+        observation: BackfillObservationPersistence,
+    ) -> Self {
+        return Self { reference, entity, observation };
+    }
+
     /// Returns the network-scoped canonical transaction identity classified by this result.
     #[must_use]
     pub fn reference(&self) -> &ksp_store_lib::RawTransactionReference {
@@ -112,11 +121,11 @@ where
         return std::result::Result::Err(persistence_error("store.network"));
     }
     return match hydration {
-        crate::BackfillHydrationOutcome::Missing(_) => std::result::Result::Ok(BackfillPersistenceOutcome {
+        crate::BackfillHydrationOutcome::Missing(_) => std::result::Result::Ok(BackfillPersistenceOutcome::new(
             reference,
-            entity: BackfillEntityPersistence::Missing,
-            observation: BackfillObservationPersistence::NotApplicable,
-        }),
+            BackfillEntityPersistence::Missing,
+            BackfillObservationPersistence::NotApplicable,
+        )),
         crate::BackfillHydrationOutcome::Available(acquisition) => {
             let (transaction, observation) = acquisition.into_parts();
             persist_available_with_port(port, reference, transaction, observation).await
@@ -144,11 +153,11 @@ where
         std::result::Result::Ok(outcome) => map_store_outcome(reference, outcome),
         std::result::Result::Err(error) => {
             if error.code() == ksp_store_lib::ERROR_CODE_RAW_CONFLICT {
-                return std::result::Result::Ok(BackfillPersistenceOutcome {
+                return std::result::Result::Ok(BackfillPersistenceOutcome::new(
                     reference,
-                    entity: BackfillEntityPersistence::Conflict,
-                    observation: BackfillObservationPersistence::NotRecorded,
-                });
+                    BackfillEntityPersistence::Conflict,
+                    BackfillObservationPersistence::NotRecorded,
+                ));
             }
             std::result::Result::Err(error)
         },
@@ -162,32 +171,32 @@ fn map_store_outcome(
     let entity = outcome.entity();
     let observation = outcome.observation();
     if entity == ksp_store_lib::RawEntityWriteOutcome::Inserted && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
-        return std::result::Result::Ok(BackfillPersistenceOutcome {
+        return std::result::Result::Ok(BackfillPersistenceOutcome::new(
             reference,
-            entity: BackfillEntityPersistence::Inserted,
-            observation: BackfillObservationPersistence::Inserted,
-        });
+            BackfillEntityPersistence::Inserted,
+            BackfillObservationPersistence::Inserted,
+        ));
     }
     if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
-        return std::result::Result::Ok(BackfillPersistenceOutcome {
+        return std::result::Result::Ok(BackfillPersistenceOutcome::new(
             reference,
-            entity: BackfillEntityPersistence::AlreadyPresent,
-            observation: BackfillObservationPersistence::Inserted,
-        });
+            BackfillEntityPersistence::AlreadyPresent,
+            BackfillObservationPersistence::Inserted,
+        ));
     }
     if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent {
-        return std::result::Result::Ok(BackfillPersistenceOutcome {
+        return std::result::Result::Ok(BackfillPersistenceOutcome::new(
             reference,
-            entity: BackfillEntityPersistence::AlreadyPresent,
-            observation: BackfillObservationPersistence::AlreadyPresent,
-        });
+            BackfillEntityPersistence::AlreadyPresent,
+            BackfillObservationPersistence::AlreadyPresent,
+        ));
     }
     if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded {
-        return std::result::Result::Ok(BackfillPersistenceOutcome {
+        return std::result::Result::Ok(BackfillPersistenceOutcome::new(
             reference,
-            entity: BackfillEntityPersistence::SkippedPurged,
-            observation: BackfillObservationPersistence::NotRecorded,
-        });
+            BackfillEntityPersistence::SkippedPurged,
+            BackfillObservationPersistence::NotRecorded,
+        ));
     }
     return std::result::Result::Err(persistence_error("store.outcome"));
 }

crates/ksp-job-backfill-lib/src/request.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-backfill-lib/src/request.rs
-// version: 2
+// version: 3
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -251,6 +251,7 @@ pub struct BackfillRequest {
     hydration_concurrency: usize,
     min_context_slot: std::option::Option<u64>,
     scope_fingerprint: BackfillScopeFingerprint,
+    checkpoint: std::option::Option<crate::BackfillCheckpoint>,
 }
 
 impl BackfillRequest {
@@ -308,6 +309,7 @@ impl BackfillRequest {
             hydration_concurrency,
             min_context_slot,
             scope_fingerprint,
+            checkpoint: std::option::Option::None,
         });
     }
 
@@ -379,6 +381,22 @@ impl BackfillRequest {
     pub const fn scope_fingerprint(&self) -> BackfillScopeFingerprint {
         return self.scope_fingerprint;
     }
+
+    /// Attaches one caller-owned checkpoint after validating Job and semantic scope identity.
+    pub fn with_checkpoint(mut self, checkpoint: crate::BackfillCheckpoint) -> ksp_core_lib::Result<Self> {
+        let validation = crate::validate_request_checkpoint(&self.job_id, self.scope_fingerprint, self.scope.kind(), &checkpoint);
+        if let std::result::Result::Err(error) = validation {
+            return std::result::Result::Err(error);
+        }
+        self.checkpoint = std::option::Option::Some(checkpoint);
+        return std::result::Result::Ok(self);
+    }
+
+    /// Returns the validated optional checkpoint supplied for controlled resumption.
+    #[must_use]
+    pub const fn checkpoint(&self) -> std::option::Option<&crate::BackfillCheckpoint> {
+        return self.checkpoint.as_ref();
+    }
 }
 
 impl std::fmt::Debug for BackfillRequest {
@@ -396,6 +414,7 @@ impl std::fmt::Debug for BackfillRequest {
             .field("hydration_concurrency", &self.hydration_concurrency)
             .field("min_context_slot", &self.min_context_slot)
             .field("scope_fingerprint", &self.scope_fingerprint)
+            .field("has_checkpoint", &self.checkpoint.is_some())
             .finish();
     }
 }