// file: crates/ksp-job-backfill-lib/src/discovery.rs // version: 6 /// 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 crate::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: crate::BackfillCandidateIdentity, discovered_slot: std::option::Option, } impl crate::BackfillCandidate { /// Creates one candidate from a network-scoped identity and optional discovery slot. #[must_use] pub fn new(identity: crate::BackfillCandidateIdentity, discovered_slot: std::option::Option) -> Self { return Self { identity, discovered_slot }; } /// Returns the network-scoped candidate identity. #[must_use] pub const fn identity(&self) -> &crate::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 { 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 crate::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, pages_fetched: usize, boundary: crate::BackfillDiscoveryBoundary, } impl crate::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, pages_fetched: usize, boundary: crate::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 { 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) -> &[crate::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) -> crate::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 crate::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::marker::Send + '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 { 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, std::option::Option::None).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; } /// Discovers one bounded candidate set while allowing cooperative cancellation of in-flight RPC futures. pub(crate) async fn discover_backfill_candidates_cancellable( transport: &ksp_onchain_transport_lib::HttpTransportPool, request: &crate::BackfillRequest, cancellation: &crate::BackfillCancellationSignal, ) -> ksp_core_lib::Result { return discover_with_source(transport, request, std::option::Option::Some(cancellation)).await; } async fn discover_with_source( source: &S, request: &crate::BackfillRequest, cancellation: std::option::Option<&crate::BackfillCancellationSignal>, ) -> ksp_core_lib::Result where S: SignaturePageSource, { if cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) { return std::result::Result::Err(cancelled_error()); } return match request.scope().kind() { crate::BackfillScopeKind::ExplicitSignatures => discover_explicit(request), crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => discover_older(source, request, cancellation).await, crate::BackfillScopeKind::AfterAddress => discover_after(source, request, cancellation).await, }; } fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result { 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 = crate::BackfillCandidateIdentity::new(request.network().clone(), signature.clone()); candidates.push(crate::BackfillCandidate::new(identity, std::option::Option::None)); } return std::result::Result::Ok(crate::BackfillDiscovery::new( request.network().clone(), request.scope_fingerprint(), candidates, 0, crate::BackfillDiscoveryBoundary::ExplicitInput, )); } async fn discover_older( source: &S, request: &crate::BackfillRequest, cancellation: std::option::Option<&crate::BackfillCancellationSignal>, ) -> ksp_core_lib::Result 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 => 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")); }, }; let mut candidates = std::vec::Vec::with_capacity(request.max_candidates()); let mut seen = std::collections::HashSet::::with_capacity(request.max_candidates()); let mut pages_fetched = 0_usize; let boundary = loop { if candidates.len() >= request.max_candidates() { break crate::BackfillDiscoveryBoundary::CandidateLimit; } if pages_fetched >= request.max_pages() { break crate::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 = fetch_signature_page(source, request.role(), address, config, cancellation).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 = crate::BackfillCandidateIdentity::new(request.network().clone(), signature); candidates.push(crate::BackfillCandidate::new(identity, std::option::Option::Some(entry.slot))); if candidates.len() >= request.max_candidates() { break; } } } if page_len < page_limit { break crate::BackfillDiscoveryBoundary::RpcBoundary; } let next_before = match next_before { std::option::Option::Some(next_before) => next_before, std::option::Option::None => break crate::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(crate::BackfillDiscovery::new(request.network().clone(), request.scope_fingerprint(), candidates, pages_fetched, boundary)); } async fn discover_after( source: &S, request: &crate::BackfillRequest, cancellation: std::option::Option<&crate::BackfillCancellationSignal>, ) -> ksp_core_lib::Result 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::::None; let mut nearest = std::collections::VecDeque::::with_capacity(request.max_candidates()); let mut seen = std::collections::HashSet::::new(); let mut pages_fetched = 0_usize; let boundary = loop { if pages_fetched >= request.max_pages() { break crate::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 = fetch_signature_page(source, request.role(), address, config, cancellation).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 = crate::BackfillCandidateIdentity::new(request.network().clone(), signature); nearest.push_back(crate::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 crate::BackfillDiscoveryBoundary::RpcBoundary; } let next_before = match next_before { std::option::Option::Some(next_before) => next_before, std::option::Option::None => break crate::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(crate::BackfillDiscovery::new( request.network().clone(), request.scope_fingerprint(), nearest.into_iter().collect(), pages_fetched, boundary, )); } async fn fetch_signature_page( source: &S, role: &ksp_onchain_transport_lib::HttpRoleName, address: &ksp_core_lib::Pubkey, config: ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig, cancellation: std::option::Option<&crate::BackfillCancellationSignal>, ) -> ksp_core_lib::Result> where S: SignaturePageSource, { let future = source.fetch_signature_page(role, address, config); return match cancellation { std::option::Option::Some(signal) => signal.run_cancellable(future).await, std::option::Option::None => future.await, }; } fn cancelled_error() -> ksp_core_lib::Error { return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CANCELLED, "Backfill discovery cancelled before RPC completion"); } fn validated_signature(value: &str) -> ksp_core_lib::Result { 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;