v0.3.6-pre.005

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

@@ -0,0 +1,443 @@
+// file: crates/ksp-job-backfill-lib/src/discovery.rs
+// version: 1
+
+/// Network-scoped identity of one discovered transaction candidate before canonical signature decoding.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct BackfillCandidateIdentity {
+    network: ksp_store_lib::RawNetworkId,
+    signature: crate::BackfillSignature,
+}
+
+impl BackfillCandidateIdentity {
+    /// Creates one candidate identity from its logical network and encoded transaction signature.
+    #[must_use]
+    pub fn new(network: ksp_store_lib::RawNetworkId, signature: crate::BackfillSignature) -> Self {
+        return Self { network, signature };
+    }
+
+    /// Returns the logical network that scopes signature uniqueness.
+    #[must_use]
+    pub const fn network(&self) -> &ksp_store_lib::RawNetworkId {
+        return &self.network;
+    }
+
+    /// Returns the validated encoded transaction signature.
+    #[must_use]
+    pub const fn signature(&self) -> &crate::BackfillSignature {
+        return &self.signature;
+    }
+}
+
+/// One deterministic transaction candidate produced by bounded discovery.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct BackfillCandidate {
+    identity: BackfillCandidateIdentity,
+    discovered_slot: std::option::Option<u64>,
+}
+
+impl BackfillCandidate {
+    /// Creates one candidate from a network-scoped identity and optional discovery slot.
+    #[must_use]
+    pub fn new(identity: BackfillCandidateIdentity, discovered_slot: std::option::Option<u64>) -> Self {
+        return Self { identity, discovered_slot };
+    }
+
+    /// Returns the network-scoped candidate identity.
+    #[must_use]
+    pub const fn identity(&self) -> &BackfillCandidateIdentity {
+        return &self.identity;
+    }
+
+    /// Returns the slot reported by address discovery, when one was available.
+    #[must_use]
+    pub const fn discovered_slot(&self) -> std::option::Option<u64> {
+        return self.discovered_slot;
+    }
+}
+
+/// Reason bounded candidate discovery stopped.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum BackfillDiscoveryBoundary {
+    /// Explicit signatures required no RPC pagination.
+    ExplicitInput,
+    /// The caller's requested candidate limit was satisfied.
+    CandidateLimit,
+    /// The RPC returned a short or empty page, reaching the bounded remote history boundary.
+    RpcBoundary,
+    /// The maximum page count was exhausted before the requested address window completed.
+    PageLimit,
+    /// An `AfterAddress` scan exhausted its page bound before the exclusive anchor boundary was reached.
+    AfterAnchorNotReached,
+}
+
+impl BackfillDiscoveryBoundary {
+    /// Returns the stable diagnostic code for this boundary.
+    #[must_use]
+    pub const fn code(self) -> &'static str {
+        return match self {
+            Self::ExplicitInput => "explicit_input",
+            Self::CandidateLimit => "candidate_limit",
+            Self::RpcBoundary => "rpc_boundary",
+            Self::PageLimit => "page_limit",
+            Self::AfterAnchorNotReached => "after_anchor_not_reached",
+        };
+    }
+
+    /// Returns whether the discovery result is partial and must not advance a future checkpoint beyond the unresolved gap.
+    #[must_use]
+    pub const fn is_partial(self) -> bool {
+        return matches!(self, Self::PageLimit | Self::AfterAnchorNotReached);
+    }
+}
+
+/// Complete bounded output of one candidate discovery pass.
+pub struct BackfillDiscovery {
+    network: ksp_store_lib::RawNetworkId,
+    scope_fingerprint: crate::BackfillScopeFingerprint,
+    candidates: std::vec::Vec<BackfillCandidate>,
+    pages_fetched: usize,
+    boundary: BackfillDiscoveryBoundary,
+}
+
+impl BackfillDiscovery {
+    /// Returns the logical network shared by every candidate identity.
+    #[must_use]
+    pub const fn network(&self) -> &ksp_store_lib::RawNetworkId {
+        return &self.network;
+    }
+
+    /// Returns the semantic scope fingerprint copied from the validated request.
+    #[must_use]
+    pub const fn scope_fingerprint(&self) -> crate::BackfillScopeFingerprint {
+        return self.scope_fingerprint;
+    }
+
+    /// Returns discovered candidates in deterministic processing order.
+    #[must_use]
+    pub fn candidates(&self) -> &[BackfillCandidate] {
+        return self.candidates.as_slice();
+    }
+
+    /// Returns the number of `getSignaturesForAddress` pages fetched by this pass.
+    #[must_use]
+    pub const fn pages_fetched(&self) -> usize {
+        return self.pages_fetched;
+    }
+
+    /// Returns the reason discovery stopped.
+    #[must_use]
+    pub const fn boundary(&self) -> BackfillDiscoveryBoundary {
+        return self.boundary;
+    }
+
+    /// Returns whether the bounded pass stopped before a complete requested discovery window was proven.
+    #[must_use]
+    pub const fn is_partial(&self) -> bool {
+        return self.boundary.is_partial();
+    }
+}
+
+impl std::fmt::Debug for BackfillDiscovery {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("BackfillDiscovery")
+            .field("network", &self.network)
+            .field("scope_fingerprint", &self.scope_fingerprint)
+            .field("candidate_count", &self.candidates.len())
+            .field("pages_fetched", &self.pages_fetched)
+            .field("boundary", &self.boundary)
+            .finish();
+    }
+}
+
+struct SignaturePageEntry {
+    signature: std::string::String,
+    slot: u64,
+}
+
+type SignaturePageFuture<'a> = std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ksp_core_lib::Result<std::vec::Vec<SignaturePageEntry>>> + 'a>>;
+
+trait SignaturePageSource {
+    fn fetch_signature_page<'a>(
+        &'a self,
+        role: &'a ksp_onchain_transport_lib::HttpRoleName,
+        address: &'a ksp_core_lib::Pubkey,
+        config: ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig,
+    ) -> SignaturePageFuture<'a>;
+}
+
+impl SignaturePageSource for ksp_onchain_transport_lib::HttpTransportPool {
+    fn fetch_signature_page<'a>(
+        &'a self,
+        role: &'a ksp_onchain_transport_lib::HttpRoleName,
+        address: &'a ksp_core_lib::Pubkey,
+        config: ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig,
+    ) -> SignaturePageFuture<'a> {
+        return std::boxed::Box::pin(async move {
+            let result = self.get_signatures_for_address(role, address, std::option::Option::Some(&config)).await;
+            let infos = match result {
+                std::result::Result::Ok(infos) => infos,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            let mut entries = std::vec::Vec::with_capacity(infos.len());
+            for info in infos {
+                entries.push(SignaturePageEntry { signature: info.signature().to_owned(), slot: info.slot() });
+            }
+            return std::result::Result::Ok(entries);
+        });
+    }
+}
+
+/// Discovers one bounded deterministic candidate set using only the typed KSP Transport wrapper.
+///
+/// Provider, endpoint and protocol selection remain entirely owned by Transport. Candidate identity
+/// is scoped only by the request network plus transaction signature.
+pub async fn discover_backfill_candidates(
+    transport: &ksp_onchain_transport_lib::HttpTransportPool,
+    request: &crate::BackfillRequest,
+) -> ksp_core_lib::Result<BackfillDiscovery> {
+    ksp_logging_lib::debug!(
+        target: crate::TRACING_TARGET,
+        job_id = request.job_id().as_str(),
+        network = request.network().as_str(),
+        scope = request.scope().kind().code(),
+        page_size = request.page_size(),
+        max_pages = request.max_pages(),
+        max_candidates = request.max_candidates(),
+        "starting bounded Backfill candidate discovery"
+    );
+    let result = discover_with_source(transport, request).await;
+    if let std::result::Result::Ok(discovery) = &result {
+        ksp_logging_lib::debug!(
+            target: crate::TRACING_TARGET,
+            job_id = request.job_id().as_str(),
+            network = request.network().as_str(),
+            scope = request.scope().kind().code(),
+            pages_fetched = discovery.pages_fetched(),
+            candidate_count = discovery.candidates().len(),
+            boundary = discovery.boundary().code(),
+            partial = discovery.is_partial(),
+            "completed bounded Backfill candidate discovery"
+        );
+    }
+    return result;
+}
+
+async fn discover_with_source<S>(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery>
+where
+    S: SignaturePageSource,
+{
+    return match request.scope().kind() {
+        crate::BackfillScopeKind::ExplicitSignatures => discover_explicit(request),
+        crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => discover_older(source, request).await,
+        crate::BackfillScopeKind::AfterAddress => discover_after(source, request).await,
+    };
+}
+
+fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery> {
+    let signatures = match request.scope().signatures() {
+        std::option::Option::Some(signatures) => signatures,
+        std::option::Option::None => return std::result::Result::Err(discovery_invalid("scope.signatures")),
+    };
+    let mut candidates = std::vec::Vec::with_capacity(signatures.len());
+    for signature in signatures {
+        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(),
+        candidates,
+        pages_fetched: 0,
+        boundary: BackfillDiscoveryBoundary::ExplicitInput,
+    });
+}
+
+async fn discover_older<S>(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery>
+where
+    S: SignaturePageSource,
+{
+    let address = match request.scope().address() {
+        std::option::Option::Some(address) => address,
+        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| value.as_str().to_owned()),
+        crate::BackfillScopeKind::LatestAddress => std::option::Option::None,
+        crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => {
+            return std::result::Result::Err(discovery_invalid("scope.kind"));
+        },
+    };
+    let mut candidates = std::vec::Vec::with_capacity(request.max_candidates());
+    let mut seen = std::collections::HashSet::<crate::BackfillSignature>::with_capacity(request.max_candidates());
+    let mut pages_fetched = 0_usize;
+    let boundary = loop {
+        if candidates.len() >= request.max_candidates() {
+            break BackfillDiscoveryBoundary::CandidateLimit;
+        }
+        if pages_fetched >= request.max_pages() {
+            break BackfillDiscoveryBoundary::PageLimit;
+        }
+        let remaining = request.max_candidates() - candidates.len();
+        let page_limit = std::cmp::min(request.page_size(), remaining);
+        let config = ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig::new(
+            before.clone(),
+            std::option::Option::None,
+            std::option::Option::Some(page_limit),
+            std::option::Option::Some(request.commitment().transport()),
+            request.min_context_slot(),
+        );
+        let page_result = source.fetch_signature_page(request.role(), address, config).await;
+        let page = match page_result {
+            std::result::Result::Ok(page) => page,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        pages_fetched += 1;
+        if page.len() > page_limit {
+            return std::result::Result::Err(discovery_invalid("page.length"));
+        }
+        let page_len = page.len();
+        let next_before = match page.last() {
+            std::option::Option::Some(entry) => {
+                let signature_result = validated_signature(entry.signature.as_str());
+                match signature_result {
+                    std::result::Result::Ok(signature) => std::option::Option::Some(signature),
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                }
+            },
+            std::option::Option::None => std::option::Option::None,
+        };
+        for entry in page {
+            let signature_result = validated_signature(entry.signature.as_str());
+            let signature = match signature_result {
+                std::result::Result::Ok(signature) => signature,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            if seen.insert(signature.clone()) {
+                let identity = BackfillCandidateIdentity::new(request.network().clone(), signature);
+                candidates.push(BackfillCandidate::new(identity, std::option::Option::Some(entry.slot)));
+                if candidates.len() >= request.max_candidates() {
+                    break;
+                }
+            }
+        }
+        if page_len < page_limit {
+            break BackfillDiscoveryBoundary::RpcBoundary;
+        }
+        let next_before = match next_before {
+            std::option::Option::Some(next_before) => next_before,
+            std::option::Option::None => break BackfillDiscoveryBoundary::RpcBoundary,
+        };
+        if before.as_deref() == std::option::Option::Some(next_before.as_str()) {
+            return std::result::Result::Err(discovery_stalled());
+        }
+        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,
+    });
+}
+
+async fn discover_after<S>(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery>
+where
+    S: SignaturePageSource,
+{
+    let address = match request.scope().address() {
+        std::option::Option::Some(address) => address,
+        std::option::Option::None => return std::result::Result::Err(discovery_invalid("scope.address")),
+    };
+    let anchor = match request.scope().anchor() {
+        std::option::Option::Some(anchor) => anchor,
+        std::option::Option::None => return std::result::Result::Err(discovery_invalid("scope.anchor")),
+    };
+    let until = anchor.as_str().to_owned();
+    let mut before = std::option::Option::<std::string::String>::None;
+    let mut nearest = std::collections::VecDeque::<BackfillCandidate>::with_capacity(request.max_candidates());
+    let mut seen = std::collections::HashSet::<crate::BackfillSignature>::new();
+    let mut pages_fetched = 0_usize;
+    let boundary = loop {
+        if pages_fetched >= request.max_pages() {
+            break BackfillDiscoveryBoundary::AfterAnchorNotReached;
+        }
+        let config = ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig::new(
+            before.clone(),
+            std::option::Option::Some(until.clone()),
+            std::option::Option::Some(request.page_size()),
+            std::option::Option::Some(request.commitment().transport()),
+            request.min_context_slot(),
+        );
+        let page_result = source.fetch_signature_page(request.role(), address, config).await;
+        let page = match page_result {
+            std::result::Result::Ok(page) => page,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        pages_fetched += 1;
+        if page.len() > request.page_size() {
+            return std::result::Result::Err(discovery_invalid("page.length"));
+        }
+        let page_len = page.len();
+        let next_before = match page.last() {
+            std::option::Option::Some(entry) => {
+                let signature_result = validated_signature(entry.signature.as_str());
+                match signature_result {
+                    std::result::Result::Ok(signature) => std::option::Option::Some(signature),
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                }
+            },
+            std::option::Option::None => std::option::Option::None,
+        };
+        for entry in page {
+            let signature_result = validated_signature(entry.signature.as_str());
+            let signature = match signature_result {
+                std::result::Result::Ok(signature) => signature,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            if seen.insert(signature.clone()) {
+                let identity = BackfillCandidateIdentity::new(request.network().clone(), signature);
+                nearest.push_back(BackfillCandidate::new(identity, std::option::Option::Some(entry.slot)));
+                if nearest.len() > request.max_candidates() {
+                    nearest.pop_front();
+                }
+            }
+        }
+        if page_len < request.page_size() {
+            break BackfillDiscoveryBoundary::RpcBoundary;
+        }
+        let next_before = match next_before {
+            std::option::Option::Some(next_before) => next_before,
+            std::option::Option::None => break BackfillDiscoveryBoundary::RpcBoundary,
+        };
+        if before.as_deref() == std::option::Option::Some(next_before.as_str()) {
+            return std::result::Result::Err(discovery_stalled());
+        }
+        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(),
+        pages_fetched,
+        boundary,
+    });
+}
+
+fn validated_signature(value: &str) -> ksp_core_lib::Result<crate::BackfillSignature> {
+    return crate::BackfillSignature::new(value.to_owned());
+}
+
+fn discovery_invalid(field: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_DISCOVERY_INVALID, "invalid bounded Backfill discovery state").with_context("field", field);
+}
+
+fn discovery_stalled() -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_DISCOVERY_STALLED, "Backfill discovery cursor did not advance")
+        .with_context("field", "before");
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/discovery.rs"]
+mod tests;