From f2f5a0edc649b58717bdeddf1d72e5522b80c0c4 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Sat, 25 Jul 2026 10:42:24 +0200 Subject: [PATCH] v0.1.0-pre.031 --- CHANGELOG.md | 7 + README.md | 5 + ROADMAP.md | 13 +- kb-pipeline/Cargo.toml | 5 +- kb-pipeline/src/backfill.rs | 1984 +++++++++++++++++++++++++++++++++++ kb-pipeline/src/lib.rs | 21 +- 6 files changed, 2032 insertions(+), 3 deletions(-) create mode 100644 kb-pipeline/src/backfill.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index a661b21..fb45a18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,10 @@ +## 0.1.0-pre.031 + +- Migration complète de l’orchestration HTTP `backfill` dans `kb-pipeline`. +- Conservation des sources explicites et historiques par adresse, des directions `before`/`after`, de la pagination bornée, des retries et de la reprise déterministe. +- Adaptation aux façades `kb-onchain-transport`, `kb-store` et `kb-lib`. +- Conservation des 15 tests bot2, dont les deux tests async d’annulation coopérative. + ## 0.1.0-pre.030 diff --git a/README.md b/README.md index 88f7f53..d211152 100644 --- a/README.md +++ b/README.md @@ -98,3 +98,8 @@ La tranche `0.1.0-pre.029` migre la planification et l’extraction core de `kb- ### Pipeline migré `kb-pipeline` contient désormais l’extraction core et le replay contextuel de décodage avec matérialisation optionnelle. + + +### Backfill HTTP + +La tranche `0.1.0-pre.031` ajoute l’orchestration HTTP de backfill : signatures explicites ou historiques par adresse, pagination bornée, hydration via `getTransaction`, retries, annulation, reprise déterministe et persistance des observations. diff --git a/ROADMAP.md b/ROADMAP.md index 01bd857..89e5959 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -282,7 +282,18 @@ - [x] Planification et scopes de replay. - [x] Extraction canonique vers les tables core. -- [ ] Backfill et replay de décodage. +- [x] Backfill et replay de décodage. - [ ] Orchestrations Solana stateful et exécution. - [x] Migrer `decode_replay` dans `kb-pipeline`. + + +### 0.1.0-pre.031 — Backfill HTTP + +- [x] Porter les campagnes par signatures explicites et historique d’adresse. +- [x] Conserver les directions `before` et `after`, les ancres et la pagination RPC bornée. +- [x] Conserver la reprise au dernier candidat contigu, les retries, le pacing et l’annulation coopérative. +- [x] Adapter la persistance à `kb-store` et l’acquisition à `kb-onchain-transport`. +- [x] Conserver les 15 tests bot2. +- [ ] Valider avec Cargo sur la machine de développement. +- [ ] Porter ensuite les orchestrations Solana stateful et d’exécution. diff --git a/kb-pipeline/Cargo.toml b/kb-pipeline/Cargo.toml index 93c6728..da72994 100644 --- a/kb-pipeline/Cargo.toml +++ b/kb-pipeline/Cargo.toml @@ -1,5 +1,5 @@ # file: kb-pipeline/Cargo.toml -# version: 5 +# version: 7 [package] name = "kb-pipeline" @@ -24,5 +24,8 @@ tokio.workspace = true tracing.workspace = true uuid.workspace = true +[dev-dependencies] +bs58.workspace = true + [lints] workspace = true diff --git a/kb-pipeline/src/backfill.rs b/kb-pipeline/src/backfill.rs new file mode 100644 index 0000000..4a8318c --- /dev/null +++ b/kb-pipeline/src/backfill.rs @@ -0,0 +1,1984 @@ +// file: kb-pipeline/src/backfill.rs +// version: 10 + +//! Reusable HTTP transaction backfill orchestration. + +use futures_util::FutureExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import +use sha2::Digest; // rust-rules: trait-import + +const AFTER_HISTORY_PAGE_SIZE: u16 = 1000; +const CANCELLATION_POLL_INTERVAL_MS: u64 = 25; + +/// Address category used to identify one targeted signature history campaign. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackfillAddressKind { + /// Executable Solana program address. + Program, + /// Token mint address. + Token, + /// Pool account address. + Pool, +} + +impl BackfillAddressKind { + /// Returns the stable filter code prefix. + pub fn code(&self) -> &'static str { + return match self { + crate::BackfillAddressKind::Program => "program", + crate::BackfillAddressKind::Token => "token", + crate::BackfillAddressKind::Pool => "pool", + }; + } +} + +/// Chronological direction relative to one anchor signature. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackfillDirection { + /// Fetch signatures older than the anchor. + Before, + /// Fetch the nearest signatures newer than the anchor. + After, +} + +impl BackfillDirection { + /// Returns the stable direction code. + pub fn code(&self) -> &'static str { + return match self { + crate::BackfillDirection::Before => "before", + crate::BackfillDirection::After => "after", + }; + } +} + +/// Candidate source used by one backfill campaign. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum BackfillSource { + /// Explicit transaction signatures supplied by the operator. + ExplicitSignatures(std::vec::Vec), + /// Signatures discovered from one address history. + AddressHistory { + /// Semantic category of the address. + kind: crate::BackfillAddressKind, + /// Program, token mint or pool address. + address: std::string::String, + /// Optional anchor signature. `Before` without an anchor starts from the latest history page. + anchor_signature: std::option::Option, + /// Chronological direction relative to the anchor. + direction: crate::BackfillDirection, + /// Maximum number of signatures to hydrate. + limit: usize, + }, +} + +/// Complete bounded backfill request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackfillRequest { + /// Endpoint role used by the HTTP pool. + pub role: std::string::String, + /// Requested commitment for history and transaction calls. + pub commitment: std::string::String, + /// Candidate source. + pub source: crate::BackfillSource, + /// Maximum RPC page size between 1 and 1000. + pub page_size: u16, + /// Maximum number of history pages inspected. + pub max_pages: u32, + /// Operator concurrency cap before endpoint role limits are applied. + pub max_concurrent_requests: u32, + /// Number of retries after the initial HTTP attempt. + pub max_retries: u32, +} + +impl BackfillRequest { + /// Validates request bounds and source identifiers. + pub fn validate(&self) -> kb_core::Result<()> { + if self.role.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "backfill endpoint role must not be empty", + )); + } + let transaction_config_result = + kb_onchain_transport::GetTransactionConfig::new(self.commitment.clone(), 0); + if let std::result::Result::Err(error) = transaction_config_result { + return std::result::Result::Err(error); + } + if self.page_size == 0 || self.page_size > 1000 { + return std::result::Result::Err(kb_core::Error::config( + "backfill page size must be between 1 and 1000", + )); + } + if self.max_pages == 0 { + return std::result::Result::Err(kb_core::Error::config( + "backfill max pages must be greater than zero", + )); + } + if self.max_concurrent_requests == 0 { + return std::result::Result::Err(kb_core::Error::config( + "backfill concurrency must be greater than zero", + )); + } + match &self.source { + crate::BackfillSource::ExplicitSignatures(signatures) => { + if signatures.is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "explicit signature backfill requires at least one signature", + )); + } + for signature in signatures { + let validation_result = + kb_onchain_transport::validate_transaction_signature_text( + signature.as_str(), + "explicit backfill signature", + ); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + } + }, + crate::BackfillSource::AddressHistory { + address, + anchor_signature, + direction, + limit, + .. + } => { + let address_result = kb_onchain_transport::validate_solana_pubkey_text( + address.as_str(), + "backfill history address", + ); + if let std::result::Result::Err(error) = address_result { + return std::result::Result::Err(error); + } + if let std::option::Option::Some(signature) = anchor_signature { + let signature_result = + kb_onchain_transport::validate_transaction_signature_text( + signature.as_str(), + "backfill anchor signature", + ); + if let std::result::Result::Err(error) = signature_result { + return std::result::Result::Err(error); + } + } else if *direction == crate::BackfillDirection::After { + return std::result::Result::Err(kb_core::Error::config( + "newer address history backfill requires an anchor signature", + )); + } + if *limit == 0 { + return std::result::Result::Err(kb_core::Error::config( + "address history backfill limit must be greater than zero", + )); + } + }, + } + return std::result::Result::Ok(()); + } + + /// Returns a stable filter code for transaction observations. + pub fn filter_code(&self) -> std::string::String { + return match &self.source { + crate::BackfillSource::ExplicitSignatures(_) => "explicit_signatures".to_string(), + crate::BackfillSource::AddressHistory { kind, anchor_signature, direction, .. } => { + if *direction == crate::BackfillDirection::Before && anchor_signature.is_none() { + return format!("{}_latest", kind.code()); + } + format!("{}_{}", kind.code(), direction.code()) + }, + }; + } +} + +/// Backfill progress severity. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackfillProgressLevel { + /// Diagnostic detail. + Debug, + /// Normal campaign information. + Info, + /// Recoverable issue or retry. + Warning, + /// Non-recoverable item or campaign failure. + Error, +} + +impl BackfillProgressLevel { + /// Returns the stable lowercase level code. + pub fn code(&self) -> &'static str { + return match self { + crate::BackfillProgressLevel::Debug => "debug", + crate::BackfillProgressLevel::Info => "info", + crate::BackfillProgressLevel::Warning => "warn", + crate::BackfillProgressLevel::Error => "error", + }; + } +} + +/// One operator-visible progress event. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackfillProgressEvent { + /// UTC timestamp rendered in RFC 3339. + pub timestamp: std::string::String, + /// Severity. + pub level: crate::BackfillProgressLevel, + /// Human-readable message. + pub message: std::string::String, + /// Number of candidates completed when known. + pub completed: std::option::Option, + /// Total candidates when known. + pub total: std::option::Option, +} + +impl BackfillProgressEvent { + fn new( + level: crate::BackfillProgressLevel, + message: impl std::convert::Into, + completed: std::option::Option, + total: std::option::Option, + ) -> Self { + return Self { + timestamp: chrono::Utc::now().to_rfc3339(), + level, + message: message.into(), + completed, + total, + }; + } +} + +/// Progress and cancellation contract implemented by applications. +pub trait BackfillObserver: Sync { + /// Receives one progress event. + fn on_progress(&self, event: &crate::BackfillProgressEvent); + /// Returns true when the operator requested cancellation. + fn is_cancelled(&self) -> bool; +} + +/// Final counters and pagination information for one campaign. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackfillSummary { + /// Unique capture session identifier. + pub capture_session_id: std::string::String, + /// Stable filter code. + pub filter_code: std::string::String, + /// Endpoint role. + pub role: std::string::String, + /// Provider used for transaction hydration. + pub provider: std::string::String, + /// Endpoint code used for transaction hydration. + pub endpoint_code: std::string::String, + /// Number of history pages fetched. + pub pages_fetched: u64, + /// Number of unique candidates selected. + pub candidates_selected: u64, + /// Number of candidates admitted to the bounded execution queue. + pub candidates_started: u64, + /// Number of candidates that reached a terminal outcome. + pub candidates_completed: u64, + /// Number of admitted candidates interrupted before a terminal outcome. + pub candidates_cancelled: u64, + /// Number of selected candidates never admitted after cancellation. + pub candidates_not_started: u64, + /// Number of source transactions received and normalized. + pub transactions_received: u64, + /// Number of canonical transaction rows inserted. + pub canonical_inserted: u64, + /// Number of canonical inserts skipped by idempotence. + pub canonical_skipped: u64, + /// Number of signatures skipped because they already existed before hydration. + pub existing_skipped: u64, + /// Number of missing transaction results after retries. + pub missing: u64, + /// Number of failed candidates after retries or persistence errors. + pub failed: u64, + /// Number of observation rows inserted. + pub observations_inserted: u64, + /// Number of source attempts performed. + pub attempts: u64, + /// Whether the campaign stopped after a cancellation request. + pub cancelled: bool, + /// Optional cursor suitable for continuing an older-history scan. + pub resume_before_signature: std::option::Option, + /// Campaign start time. + pub started_at: std::string::String, + /// Campaign completion time. + pub finished_at: std::string::String, +} + +#[derive(Clone, Debug)] +struct BackfillCandidate { + signature: std::string::String, + slot: std::option::Option, + detected_at: chrono::DateTime, +} + +#[derive(Clone, Copy, Debug)] +struct HttpRoleLimits { + requests_per_second: u32, + max_concurrent_requests: u32, + pause_after_rate_limit_ms: u64, +} + +#[derive(Debug)] +struct RequestPacer { + interval: std::time::Duration, + next_start: tokio::sync::Mutex, +} + +impl RequestPacer { + fn new(requests_per_second: u32) -> Self { + let micros = 1_000_000_u64.div_ceil(u64::from(requests_per_second)); + return Self { + interval: std::time::Duration::from_micros(micros), + next_start: tokio::sync::Mutex::new(tokio::time::Instant::now()), + }; + } + + async fn wait(&self) { + let mut next_start = self.next_start.lock().await; + let now = tokio::time::Instant::now(); + if *next_start > now { + tokio::time::sleep(*next_start - now).await; + } + *next_start = tokio::time::Instant::now() + self.interval; + } +} + +async fn wait_until_cancelled(observer: &O) +where + O: crate::BackfillObserver + Sync, +{ + loop { + if observer.is_cancelled() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(CANCELLATION_POLL_INTERVAL_MS)).await; + } +} + +async fn await_or_cancelled(observer: &O, future: F) -> std::option::Option +where + O: crate::BackfillObserver + Sync, + F: std::future::Future, +{ + if observer.is_cancelled() { + return std::option::Option::None; + } + tokio::pin!(future); + let outcome = tokio::select! { + _ = wait_until_cancelled(observer) => std::option::Option::None, + value = &mut future => std::option::Option::Some(value), + }; + return outcome; +} + +async fn wait_with_cancellation(observer: &O, duration: std::time::Duration) -> bool +where + O: crate::BackfillObserver + Sync, +{ + let outcome = await_or_cancelled(observer, tokio::time::sleep(duration)).await; + return outcome.is_some(); +} + +#[derive(Clone, Debug)] +struct CandidateOutcome { + transactions_received: u64, + canonical_inserted: u64, + canonical_skipped: u64, + existing_skipped: u64, + missing: u64, + failed: u64, + observations_inserted: u64, + attempts: u64, + cancelled: bool, +} + +impl CandidateOutcome { + fn empty() -> Self { + return Self { + transactions_received: 0, + canonical_inserted: 0, + canonical_skipped: 0, + existing_skipped: 0, + missing: 0, + failed: 0, + observations_inserted: 0, + attempts: 0, + cancelled: false, + }; + } +} + +#[derive(Debug)] +struct CandidateTaskResult { + index: usize, + result: kb_core::Result, +} + +#[derive(Debug)] +struct CandidateCompletionFrontier { + completed: std::vec::Vec, + contiguous_completed: usize, +} + +impl CandidateCompletionFrontier { + fn new(candidate_count: usize) -> Self { + return Self { + completed: vec![false; candidate_count], + contiguous_completed: 0, + }; + } + + fn mark_completed(&mut self, index: usize) { + if let std::option::Option::Some(completed) = self.completed.get_mut(index) { + *completed = true; + } + while self.completed.get(self.contiguous_completed).copied().unwrap_or(false) { + self.contiguous_completed += 1; + } + } + + fn contiguous_completed(&self) -> usize { + return self.contiguous_completed; + } +} + +/// Executes one bounded HTTP backfill and persists canonical transactions plus observations. +pub async fn execute_http_backfill( + http_pool: &kb_onchain_transport::HttpEndpointPool, + store: &S, + request: &crate::BackfillRequest, + observer: &O, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, + O: crate::BackfillObserver + Sync, +{ + let validation_result = request.validate(); + if let std::result::Result::Err(error) = validation_result { + tracing::error!(target: crate::TRACING_TARGET, action = "validate_http_backfill", role = %request.role, error = %error, "HTTP backfill request validation failed"); + return std::result::Result::Err(error); + } + let started_at = chrono::Utc::now(); + let capture_session_id = uuid::Uuid::new_v4().to_string(); + let filter_code = request.filter_code(); + tracing::info!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, commitment = %request.commitment, page_size = request.page_size, max_pages = request.max_pages, requested_concurrency = request.max_concurrent_requests, max_retries = request.max_retries, "HTTP backfill campaign started"); + emit( + observer, + crate::BackfillProgressLevel::Info, + format!("backfill session {capture_session_id} started with filter {filter_code}"), + std::option::Option::None, + std::option::Option::None, + ); + let transaction_client_result = + http_pool.select_client_for_role_and_method(request.role.as_str(), "getTransaction"); + let transaction_client = match transaction_client_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "select_backfill_transaction_client", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, error = %error, "backfill transaction client selection failed"); + return std::result::Result::Err(error); + }, + }; + let transaction_limits_result = + role_limits(&transaction_client, request.role.as_str(), "get_transaction"); + let transaction_limits = match transaction_limits_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "resolve_backfill_role_limits", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, endpoint_name = %transaction_client.endpoint_name(), error = %error, "backfill role limit resolution failed"); + return std::result::Result::Err(error); + }, + }; + let discovery_result = collect_candidates(http_pool, request, observer).await; + let discovery = match discovery_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "collect_backfill_candidates", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, error = %error, "backfill candidate discovery failed"); + return std::result::Result::Err(error); + }, + }; + let total = match u64::try_from(discovery.candidates.len()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => u64::MAX, + }; + let ordered_signatures = discovery + .candidates + .iter() + .map(|candidate| return candidate.signature.clone()) + .collect::>(); + let discovery_resume_before_signature = discovery.resume_before_signature.clone(); + let mut summary = crate::BackfillSummary { + capture_session_id: capture_session_id.clone(), + filter_code: filter_code.clone(), + role: request.role.clone(), + provider: transaction_client.provider().to_string(), + endpoint_code: transaction_client.endpoint_name().to_string(), + pages_fetched: discovery.pages_fetched, + candidates_selected: total, + candidates_started: 0, + candidates_completed: 0, + candidates_cancelled: 0, + candidates_not_started: 0, + transactions_received: 0, + canonical_inserted: 0, + canonical_skipped: 0, + existing_skipped: 0, + missing: 0, + failed: 0, + observations_inserted: 0, + attempts: 0, + cancelled: false, + resume_before_signature: discovery_resume_before_signature.clone(), + started_at: started_at.to_rfc3339(), + finished_at: started_at.to_rfc3339(), + }; + if observer.is_cancelled() { + tracing::warn!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %capture_session_id, filter_code = %filter_code, candidates_selected = total, decision = "cancelled_before_admission", "HTTP backfill cancelled before candidate admission"); + summary.cancelled = true; + summary.candidates_not_started = total; + summary.resume_before_signature = resolved_resume_before_signature( + request, + ordered_signatures.as_slice(), + 0, + discovery_resume_before_signature, + true, + ); + summary.finished_at = chrono::Utc::now().to_rfc3339(); + return std::result::Result::Ok(summary); + } + let effective_concurrency = std::cmp::max( + 1_usize, + std::cmp::min( + request.max_concurrent_requests as usize, + transaction_limits.max_concurrent_requests as usize, + ), + ); + let pacer = std::sync::Arc::new(RequestPacer::new(transaction_limits.requests_per_second)); + let transaction_config_result = + kb_onchain_transport::GetTransactionConfig::new(request.commitment.clone(), 0); + let transaction_config = match transaction_config_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let mut pending = discovery.candidates.into_iter().enumerate(); + let mut in_flight = futures_util::stream::FuturesUnordered::new(); + let mut completion_frontier = CandidateCompletionFrontier::new(ordered_signatures.len()); + let create_task = |index: usize, candidate: BackfillCandidate| { + let pacer_value = std::sync::Arc::clone(&pacer); + let capture_session_id_value = capture_session_id.clone(); + let filter_code_value = filter_code.clone(); + let transaction_client_value = transaction_client.clone(); + let transaction_config_value = transaction_config.clone(); + return async move { + let result = hydrate_candidate( + store, + observer, + &transaction_client_value, + &transaction_config_value, + request, + &capture_session_id_value, + &filter_code_value, + candidate, + pacer_value.as_ref(), + transaction_limits.pause_after_rate_limit_ms, + ) + .await; + return CandidateTaskResult { index, result }; + } + .boxed(); + }; + while in_flight.len() < effective_concurrency && !observer.is_cancelled() { + let next_candidate = pending.next(); + let (index, candidate) = match next_candidate { + std::option::Option::Some(value) => value, + std::option::Option::None => break, + }; + summary.candidates_started += 1; + in_flight.push(create_task(index, candidate)); + } + while let std::option::Option::Some(task) = in_flight.next().await { + let index = task.index; + match task.result { + std::result::Result::Ok(outcome) => { + if outcome.cancelled { + summary.candidates_cancelled += 1; + summary.cancelled = true; + emit( + observer, + crate::BackfillProgressLevel::Debug, + format!("candidate {} cancelled before terminal processing", index + 1), + std::option::Option::Some(summary.candidates_completed), + std::option::Option::Some(total), + ); + } else { + summary.candidates_completed += 1; + summary.transactions_received += outcome.transactions_received; + summary.canonical_inserted += outcome.canonical_inserted; + summary.canonical_skipped += outcome.canonical_skipped; + summary.existing_skipped += outcome.existing_skipped; + summary.missing += outcome.missing; + summary.failed += outcome.failed; + summary.observations_inserted += outcome.observations_inserted; + summary.attempts += outcome.attempts; + completion_frontier.mark_completed(index); + emit( + observer, + crate::BackfillProgressLevel::Info, + format!("processed {}/{} candidates", summary.candidates_completed, total), + std::option::Option::Some(summary.candidates_completed), + std::option::Option::Some(total), + ); + } + }, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "hydrate_backfill_candidate", capture_session_id = %capture_session_id, filter_code = %filter_code, candidate_index = index, error = %error, "backfill candidate task failed without a normal outcome"); + summary.candidates_completed += 1; + summary.failed += 1; + completion_frontier.mark_completed(index); + emit( + observer, + crate::BackfillProgressLevel::Error, + format!("candidate {} failed without a normal outcome: {error}", index + 1), + std::option::Option::Some(summary.candidates_completed), + std::option::Option::Some(total), + ); + }, + } + if observer.is_cancelled() { + summary.cancelled = true; + } else { + let next_candidate = pending.next(); + if let std::option::Option::Some((next_index, next_candidate_value)) = next_candidate { + summary.candidates_started += 1; + in_flight.push(create_task(next_index, next_candidate_value)); + } + } + } + if observer.is_cancelled() { + summary.cancelled = true; + } + summary.candidates_not_started = total.saturating_sub(summary.candidates_started); + if summary.candidates_cancelled > 0 || summary.candidates_not_started > 0 { + summary.cancelled = true; + } + summary.resume_before_signature = resolved_resume_before_signature( + request, + ordered_signatures.as_slice(), + completion_frontier.contiguous_completed(), + discovery_resume_before_signature, + summary.cancelled, + ); + summary.finished_at = chrono::Utc::now().to_rfc3339(); + let completion_level = if summary.cancelled { + crate::BackfillProgressLevel::Warning + } else { + crate::BackfillProgressLevel::Info + }; + if summary.failed > 0 { + tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, failed = summary.failed, missing = summary.missing, cancelled = summary.cancelled, "HTTP backfill completed with failed candidates"); + } else if summary.cancelled { + tracing::warn!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, candidates_cancelled = summary.candidates_cancelled, candidates_not_started = summary.candidates_not_started, "HTTP backfill campaign cancelled"); + } else { + tracing::info!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, canonical_inserted = summary.canonical_inserted, canonical_skipped = summary.canonical_skipped, existing_skipped = summary.existing_skipped, missing = summary.missing, observations_inserted = summary.observations_inserted, "HTTP backfill campaign completed"); + } + emit( + observer, + completion_level, + format!( + "backfill completed: completed={}, candidate_cancelled={}, not_started={}, inserted={}, skipped={}, existing={}, missing={}, failed={}, observations={}", + summary.candidates_completed, + summary.candidates_cancelled, + summary.candidates_not_started, + summary.canonical_inserted, + summary.canonical_skipped, + summary.existing_skipped, + summary.missing, + summary.failed, + summary.observations_inserted + ), + std::option::Option::Some(summary.candidates_completed), + std::option::Option::Some(total), + ); + return std::result::Result::Ok(summary); +} + +fn resolved_resume_before_signature( + request: &crate::BackfillRequest, + ordered_signatures: &[std::string::String], + contiguous_completed: usize, + discovery_resume_before_signature: std::option::Option, + cancelled: bool, +) -> std::option::Option { + if !cancelled { + return discovery_resume_before_signature; + } + return match &request.source { + crate::BackfillSource::AddressHistory { + anchor_signature, + direction: crate::BackfillDirection::Before, + .. + } => { + if contiguous_completed == 0 { + return anchor_signature.clone(); + } + if let std::option::Option::Some(signature) = + ordered_signatures.get(contiguous_completed - 1) + { + return std::option::Option::Some(signature.clone()); + } + anchor_signature.clone() + }, + crate::BackfillSource::ExplicitSignatures(_) + | crate::BackfillSource::AddressHistory { + direction: crate::BackfillDirection::After, + .. + } => std::option::Option::None, + }; +} + +struct CandidateDiscovery { + candidates: std::vec::Vec, + pages_fetched: u64, + resume_before_signature: std::option::Option, +} + +async fn collect_candidates( + http_pool: &kb_onchain_transport::HttpEndpointPool, + request: &crate::BackfillRequest, + observer: &O, +) -> kb_core::Result +where + O: crate::BackfillObserver + Sync, +{ + return match &request.source { + crate::BackfillSource::ExplicitSignatures(signatures) => { + std::result::Result::Ok(explicit_candidates(signatures)) + }, + crate::BackfillSource::AddressHistory { + address, + anchor_signature, + direction, + limit, + .. + } => { + let client_result = http_pool.select_client_for_role_and_method( + request.role.as_str(), + "getSignaturesForAddress", + ); + let client = match client_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let limits_result = + role_limits(&client, request.role.as_str(), "get_signatures_for_address"); + let limits = match limits_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let pacer = RequestPacer::new(limits.requests_per_second); + match direction { + crate::BackfillDirection::Before => { + collect_before_candidates( + &client, + request, + observer, + address, + anchor_signature.as_deref(), + *limit, + &pacer, + limits.pause_after_rate_limit_ms, + ) + .await + }, + crate::BackfillDirection::After => { + collect_after_candidates( + &client, + request, + observer, + address, + match anchor_signature.as_deref() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::config( + "newer address history backfill requires an anchor signature", + )); + }, + }, + *limit, + &pacer, + limits.pause_after_rate_limit_ms, + ) + .await + }, + } + }, + }; +} + +fn explicit_candidates(signatures: &[std::string::String]) -> CandidateDiscovery { + let mut seen = std::collections::HashSet::::new(); + let mut candidates = std::vec::Vec::::new(); + let detected_at = chrono::Utc::now(); + for signature in signatures { + let trimmed = signature.trim(); + if trimmed.is_empty() || !seen.insert(trimmed.to_string()) { + continue; + } + candidates.push(BackfillCandidate { + signature: trimmed.to_string(), + slot: std::option::Option::None, + detected_at, + }); + } + return CandidateDiscovery { + candidates, + pages_fetched: 0, + resume_before_signature: std::option::Option::None, + }; +} + +async fn collect_before_candidates( + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + observer: &O, + address: &str, + anchor_signature: std::option::Option<&str>, + limit: usize, + pacer: &RequestPacer, + pause_after_rate_limit_ms: u64, +) -> kb_core::Result +where + O: crate::BackfillObserver + Sync, +{ + let mut candidates = std::vec::Vec::::new(); + let mut seen = std::collections::HashSet::::new(); + let mut before = anchor_signature.map(str::to_string); + let mut pages_fetched = 0_u64; + while candidates.len() < limit && pages_fetched < u64::from(request.max_pages) { + if observer.is_cancelled() { + break; + } + let remaining = limit - candidates.len(); + let page_limit = std::cmp::min(usize::from(request.page_size), remaining); + let page_limit_u16 = match u16::try_from(page_limit) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::config(format!( + "backfill page limit conversion failed: {error}" + ))); + }, + }; + let config_result = kb_onchain_transport::GetSignaturesForAddressConfig::new( + request.commitment.clone(), + page_limit_u16, + before.clone(), + std::option::Option::None, + std::option::Option::None, + ); + let config = match config_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let page_result = fetch_signature_page( + client, + address, + &config, + request.max_retries, + pacer, + pause_after_rate_limit_ms, + observer, + ) + .await; + let page = match page_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if observer.is_cancelled() { + break; + } + pages_fetched += 1; + append_signature_candidates(&mut candidates, &mut seen, &page); + before = page.last().map(|item| return item.signature.clone()); + emit( + observer, + crate::BackfillProgressLevel::Debug, + format!("history page {pages_fetched} returned {} signatures", page.len()), + std::option::Option::None, + std::option::Option::None, + ); + if page.len() < page_limit || page.is_empty() { + break; + } + } + let resume_before_signature = before; + return std::result::Result::Ok(CandidateDiscovery { + candidates, + pages_fetched, + resume_before_signature, + }); +} + +async fn collect_after_candidates( + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + observer: &O, + address: &str, + anchor_signature: &str, + limit: usize, + pacer: &RequestPacer, + pause_after_rate_limit_ms: u64, +) -> kb_core::Result +where + O: crate::BackfillObserver + Sync, +{ + let page_size = AFTER_HISTORY_PAGE_SIZE; + let mut nearest_newer = std::collections::VecDeque::::with_capacity(limit); + let mut seen = std::collections::HashSet::::new(); + let mut before = std::option::Option::None; + let until = std::option::Option::Some(anchor_signature.to_string()); + let mut pages_fetched = 0_u64; + let mut scanned = 0_u64; + let mut boundary_reached = false; + while pages_fetched < u64::from(request.max_pages) { + if observer.is_cancelled() { + break; + } + let config_result = kb_onchain_transport::GetSignaturesForAddressConfig::new( + request.commitment.clone(), + page_size, + before.clone(), + until.clone(), + std::option::Option::None, + ); + let config = match config_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let page_result = fetch_signature_page( + client, + address, + &config, + request.max_retries, + pacer, + pause_after_rate_limit_ms, + observer, + ) + .await; + let page = match page_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if observer.is_cancelled() { + break; + } + pages_fetched += 1; + let page_count = match u64::try_from(page.len()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::config(format!( + "after-history page length conversion failed: {error}" + ))); + }, + }; + scanned = scanned.saturating_add(page_count); + append_nearest_newer_candidates(&mut nearest_newer, &mut seen, &page, limit); + before = page.last().map(|item| return item.signature.clone()); + let oldest_scanned_slot = page.last().map(|item| return item.slot); + emit( + observer, + crate::BackfillProgressLevel::Debug, + format!( + "newer-history page={pages_fetched} page_size={} scanned={scanned} oldest_scanned_slot={oldest_scanned_slot:?}", + page.len() + ), + std::option::Option::None, + std::option::Option::None, + ); + if page.len() < usize::from(page_size) { + boundary_reached = true; + break; + } + } + if !boundary_reached && !observer.is_cancelled() { + return std::result::Result::Err(kb_core::Error::config(format!( + "anchor boundary was not reached within {} pages after scanning {scanned} signatures; increase max_pages", + request.max_pages + ))); + } + return std::result::Result::Ok(CandidateDiscovery { + candidates: nearest_newer.into_iter().collect(), + pages_fetched, + resume_before_signature: std::option::Option::None, + }); +} + +fn append_nearest_newer_candidates( + candidates: &mut std::collections::VecDeque, + seen: &mut std::collections::HashSet, + page: &[kb_onchain_transport::AddressSignatureInfo], + limit: usize, +) { + let detected_at = chrono::Utc::now(); + for item in page { + if !seen.insert(item.signature.clone()) { + continue; + } + candidates.push_back(BackfillCandidate { + signature: item.signature.clone(), + slot: std::option::Option::Some(item.slot), + detected_at, + }); + if candidates.len() > limit { + candidates.pop_front(); + } + } +} + +fn append_signature_candidates( + candidates: &mut std::vec::Vec, + seen: &mut std::collections::HashSet, + page: &[kb_onchain_transport::AddressSignatureInfo], +) { + let detected_at = chrono::Utc::now(); + for item in page { + if !seen.insert(item.signature.clone()) { + continue; + } + candidates.push(BackfillCandidate { + signature: item.signature.clone(), + slot: std::option::Option::Some(item.slot), + detected_at, + }); + } +} + +async fn fetch_signature_page( + client: &kb_onchain_transport::HttpClient, + address: &str, + config: &kb_onchain_transport::GetSignaturesForAddressConfig, + max_retries: u32, + pacer: &RequestPacer, + pause_after_rate_limit_ms: u64, + observer: &O, +) -> kb_core::Result> +where + O: crate::BackfillObserver + Sync, +{ + let mut attempt = 0_u32; + loop { + if observer.is_cancelled() { + return std::result::Result::Ok(std::vec::Vec::new()); + } + let pacing_result = await_or_cancelled(observer, pacer.wait()).await; + if pacing_result.is_none() { + return std::result::Result::Ok(std::vec::Vec::new()); + } + let request_result = + await_or_cancelled(observer, client.get_signatures_for_address(address, config)).await; + let result = match request_result { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Ok(std::vec::Vec::new()), + }; + match result { + std::result::Result::Ok(page) => return std::result::Result::Ok(page), + std::result::Result::Err(error) => { + if attempt >= max_retries { + return std::result::Result::Err(error); + } + let error_text = error.to_string(); + let delay_ms = + retry_delay_ms(error_text.as_str(), attempt, pause_after_rate_limit_ms); + emit( + observer, + crate::BackfillProgressLevel::Warning, + format!( + "getSignaturesForAddress retry {} after {} ms: {}", + attempt + 1, + delay_ms, + error_text + ), + std::option::Option::None, + std::option::Option::None, + ); + let wait_completed = + wait_with_cancellation(observer, std::time::Duration::from_millis(delay_ms)) + .await; + if !wait_completed { + return std::result::Result::Ok(std::vec::Vec::new()); + } + attempt += 1; + }, + } + } +} + +async fn hydrate_candidate( + store: &S, + observer: &O, + client: &kb_onchain_transport::HttpClient, + config: &kb_onchain_transport::GetTransactionConfig, + request: &crate::BackfillRequest, + capture_session_id: &str, + filter_code: &str, + candidate: BackfillCandidate, + pacer: &RequestPacer, + pause_after_rate_limit_ms: u64, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, + O: crate::BackfillObserver + Sync, +{ + let mut outcome = CandidateOutcome::empty(); + if observer.is_cancelled() { + outcome.cancelled = true; + return std::result::Result::Ok(outcome); + } + let signature_model = kb_lib::MdSignature(candidate.signature.clone()); + let exists_result = + kb_store::RawTransactionStore::has_raw_transaction_signature(store, &signature_model).await; + let exists = match exists_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if exists { + outcome.existing_skipped = 1; + return std::result::Result::Ok(outcome); + } + let mut attempt = 0_u32; + loop { + if observer.is_cancelled() { + outcome.cancelled = true; + return std::result::Result::Ok(outcome); + } + let pacing_result = await_or_cancelled(observer, pacer.wait()).await; + if pacing_result.is_none() { + outcome.cancelled = true; + return std::result::Result::Ok(outcome); + } + outcome.attempts += 1; + let request_result = await_or_cancelled( + observer, + client.get_transaction_acquisition(candidate.signature.as_str(), config), + ) + .await; + let acquisition_result = match request_result { + std::option::Option::Some(value) => value, + std::option::Option::None => { + outcome.cancelled = true; + return std::result::Result::Ok(outcome); + }, + }; + let received_at = chrono::Utc::now(); + match acquisition_result { + std::result::Result::Ok(acquisition) => { + let metadata_result = source_payload_metadata(&acquisition.source_json); + let (payload_size_bytes, source_payload_hash) = match metadata_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(error); + }, + }; + match acquisition.canonical_transaction { + std::option::Option::Some(transaction) => { + outcome.transactions_received = 1; + let normalized_at = chrono::Utc::now(); + let raw_insert_result = + kb_store::RawTransactionInsert::from_canonical(&transaction); + let raw_insert = match raw_insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + let observation_result = persist_failed_observation( + store, + client, + request, + capture_session_id, + filter_code, + &candidate, + attempt, + received_at, + std::option::Option::Some(normalized_at), + payload_size_bytes, + source_payload_hash, + "canonical_normalization_failed", + error.to_string(), + ) + .await; + if let std::result::Result::Ok(count) = observation_result { + outcome.observations_inserted += count; + } + tracing::error!(target: crate::TRACING_TARGET, action = "normalize_backfill_transaction", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "canonical transaction normalization failed"); + outcome.failed = 1; + return std::result::Result::Ok(outcome); + }, + }; + let persist_result = kb_store::RawTransactionStore::insert_raw_transaction( + store, + &raw_insert, + ) + .await; + let persist_outcome = match persist_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + let observation_result = persist_failed_observation( + store, + client, + request, + capture_session_id, + filter_code, + &candidate, + attempt, + received_at, + std::option::Option::Some(normalized_at), + payload_size_bytes, + source_payload_hash, + "canonical_persist_failed", + error.to_string(), + ) + .await; + if let std::result::Result::Ok(count) = observation_result { + outcome.observations_inserted += count; + } + tracing::error!(target: crate::TRACING_TARGET, action = "persist_backfill_transaction", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "canonical transaction persistence failed"); + outcome.failed = 1; + return std::result::Result::Ok(outcome); + }, + }; + outcome.canonical_inserted += persist_outcome.inserted_count; + outcome.canonical_skipped += persist_outcome.skipped_count; + let observation_result = persist_success_observation( + store, + client, + request, + capture_session_id, + filter_code, + &candidate, + transaction.slot, + attempt, + received_at, + normalized_at, + payload_size_bytes, + source_payload_hash, + ) + .await; + let observation_count = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "persist_backfill_observation", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "successful backfill observation persistence failed"); + outcome.failed = 1; + emit( + observer, + crate::BackfillProgressLevel::Error, + format!( + "observation persistence failed for {}: {}", + candidate.signature, error + ), + std::option::Option::None, + std::option::Option::None, + ); + return std::result::Result::Ok(outcome); + }, + }; + outcome.observations_inserted += observation_count; + return std::result::Result::Ok(outcome); + }, + std::option::Option::None => { + let observation_result = persist_missing_observation( + store, + client, + request, + capture_session_id, + filter_code, + &candidate, + attempt, + received_at, + payload_size_bytes, + source_payload_hash, + ) + .await; + let observation_count = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(error); + }, + }; + outcome.observations_inserted += observation_count; + if attempt >= request.max_retries { + outcome.missing = 1; + return std::result::Result::Ok(outcome); + } + }, + } + }, + std::result::Result::Err(error) => { + let error_text = error.to_string(); + let observation_result = persist_failed_observation( + store, + client, + request, + capture_session_id, + filter_code, + &candidate, + attempt, + received_at, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + error_code(error_text.as_str()), + error_text.clone(), + ) + .await; + let observation_count = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(observation_error) => { + return std::result::Result::Err(observation_error); + }, + }; + outcome.observations_inserted += observation_count; + if attempt >= request.max_retries { + tracing::error!(target: crate::TRACING_TARGET, action = "hydrate_backfill_candidate", capture_session_id, filter_code, endpoint_name = %client.endpoint_name(), provider = %client.provider(), signature = %candidate.signature, attempt, max_retries = request.max_retries, error_code = error_code(error_text.as_str()), error_message = %error_text, "getTransaction failed after all retries"); + outcome.failed = 1; + return std::result::Result::Ok(outcome); + } + let delay_ms = + retry_delay_ms(error_text.as_str(), attempt, pause_after_rate_limit_ms); + emit( + observer, + crate::BackfillProgressLevel::Warning, + format!( + "getTransaction retry {} for {} after {} ms: {}", + attempt + 1, + candidate.signature, + delay_ms, + error_text + ), + std::option::Option::None, + std::option::Option::None, + ); + let wait_completed = + wait_with_cancellation(observer, std::time::Duration::from_millis(delay_ms)) + .await; + if !wait_completed { + outcome.cancelled = true; + return std::result::Result::Ok(outcome); + } + }, + } + attempt += 1; + } +} + +async fn persist_success_observation( + store: &S, + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + capture_session_id: &str, + filter_code: &str, + candidate: &BackfillCandidate, + slot: u64, + attempt: u32, + received_at: chrono::DateTime, + normalized_at: chrono::DateTime, + payload_size_bytes: std::option::Option, + source_payload_hash: std::option::Option, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, +{ + let observation_result = base_observation( + client, + request, + capture_session_id, + filter_code, + candidate, + attempt, + received_at, + ); + let observation_base = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let identity_result = observation_base + .with_transaction_identity(candidate.signature.clone(), std::option::Option::Some(slot)); + let observation_with_identity = match identity_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let observation_with_timings = observation_with_identity.with_timings( + std::option::Option::Some(candidate.detected_at), + std::option::Option::Some(normalized_at), + ); + let payload_result = + observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash); + let observation = match payload_result { + std::result::Result::Ok(value) => { + value.with_status(kb_store::TransactionObservationStatus::Persisted) + }, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return insert_observation(store, &observation).await; +} + +async fn persist_missing_observation( + store: &S, + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + capture_session_id: &str, + filter_code: &str, + candidate: &BackfillCandidate, + attempt: u32, + received_at: chrono::DateTime, + payload_size_bytes: std::option::Option, + source_payload_hash: std::option::Option, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, +{ + let observation_result = base_observation( + client, + request, + capture_session_id, + filter_code, + candidate, + attempt, + received_at, + ); + let observation_base = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let identity_result = + observation_base.with_transaction_identity(candidate.signature.clone(), candidate.slot); + let observation_with_identity = match identity_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let observation_with_timings = observation_with_identity + .with_timings(std::option::Option::Some(candidate.detected_at), std::option::Option::None); + let payload_result = + observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash); + let observation = match payload_result { + std::result::Result::Ok(value) => { + value.with_status(kb_store::TransactionObservationStatus::Missing) + }, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return insert_observation(store, &observation).await; +} + +async fn persist_failed_observation( + store: &S, + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + capture_session_id: &str, + filter_code: &str, + candidate: &BackfillCandidate, + attempt: u32, + received_at: chrono::DateTime, + normalized_at: std::option::Option>, + payload_size_bytes: std::option::Option, + source_payload_hash: std::option::Option, + error_code: &str, + error_message: std::string::String, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, +{ + let observation_result = base_observation( + client, + request, + capture_session_id, + filter_code, + candidate, + attempt, + received_at, + ); + let observation_base = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let identity_result = + observation_base.with_transaction_identity(candidate.signature.clone(), candidate.slot); + let observation_with_identity = match identity_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let observation_with_timings = observation_with_identity + .with_timings(std::option::Option::Some(candidate.detected_at), normalized_at); + let payload_result = + observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash); + let observation_with_payload = match payload_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let error_result = observation_with_payload + .with_error(error_code.to_string(), std::option::Option::Some(error_message)); + let observation = match error_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return insert_observation(store, &observation).await; +} + +fn base_observation( + client: &kb_onchain_transport::HttpClient, + request: &crate::BackfillRequest, + capture_session_id: &str, + filter_code: &str, + candidate: &BackfillCandidate, + attempt: u32, + received_at: chrono::DateTime, +) -> kb_core::Result { + let observation_key = format!( + "backfill:{}:{}:{}:{}", + capture_session_id, + client.endpoint_name(), + candidate.signature, + attempt + ); + let observation_result = kb_store::TransactionObservationInsert::new( + observation_key, + client.provider().to_string(), + "solana_http_json_rpc".to_string(), + "getTransaction".to_string(), + kb_store::TransactionObservationOrigin::Backfill, + received_at, + ); + let observation = match observation_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let endpoint_result = observation.with_endpoint_code(client.endpoint_name().to_string()); + let endpoint_observation = match endpoint_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let commitment_result = endpoint_observation.with_commitment(request.commitment.clone()); + let commitment_observation = match commitment_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return commitment_observation.with_capture_context( + std::option::Option::Some(capture_session_id.to_string()), + std::option::Option::Some(filter_code.to_string()), + ); +} + +async fn insert_observation( + store: &S, + observation: &kb_store::TransactionObservationInsert, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, +{ + let insert_result = + kb_store::RawTransactionStore::insert_transaction_observation(store, observation).await; + let outcome = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(outcome.inserted_count); +} + +fn source_payload_metadata( + source: &serde_json::Value, +) -> kb_core::Result<(std::option::Option, std::option::Option)> { + let bytes_result = serde_json::to_vec(source); + let bytes = match bytes_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot serialize source getTransaction payload for metrics: {error}" + ))); + }, + }; + let size = match u64::try_from(bytes.len()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "source payload size conversion failed: {error}" + ))); + }, + }; + let digest = sha2::Sha256::digest(bytes.as_slice()); + return std::result::Result::Ok(( + std::option::Option::Some(size), + std::option::Option::Some(lower_hex(digest.as_ref())), + )); +} + +fn lower_hex(bytes: &[u8]) -> std::string::String { + const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef"; + let mut output = std::string::String::with_capacity(bytes.len().saturating_mul(2)); + for byte in bytes { + output.push(char::from(HEX_DIGITS[usize::from(byte >> 4)])); + output.push(char::from(HEX_DIGITS[usize::from(byte & 0x0f)])); + } + return output; +} + +fn role_limits( + client: &kb_onchain_transport::HttpClient, + required_role: &str, + request_kind: &str, +) -> kb_core::Result { + for role in &client.endpoint_config().roles { + if !role.enabled || role.role != required_role { + continue; + } + let supports = + role.request_kinds.iter().any(|kind| return kind == request_kind || kind == "*"); + if !supports { + continue; + } + return std::result::Result::Ok(HttpRoleLimits { + requests_per_second: role.requests_per_second, + max_concurrent_requests: role.max_concurrent_requests, + pause_after_rate_limit_ms: role.pause_after_rate_limit_ms, + }); + } + return std::result::Result::Err(kb_core::Error::config(format!( + "endpoint '{}' does not expose role '{}' for '{}'", + client.endpoint_name(), + required_role, + request_kind + ))); +} + +fn retry_delay_ms(error_text: &str, attempt: u32, pause_after_rate_limit_ms: u64) -> u64 { + if error_text.contains("429") || error_text.to_ascii_lowercase().contains("rate limit") { + return pause_after_rate_limit_ms; + } + let exponent = std::cmp::min(attempt, 4); + return 250_u64.saturating_mul(1_u64 << exponent); +} + +fn error_code(error_text: &str) -> &'static str { + if error_text.contains("429") || error_text.to_ascii_lowercase().contains("rate limit") { + return "http_rate_limited"; + } + if error_text.to_ascii_lowercase().contains("timeout") { + return "http_timeout"; + } + return "http_rpc_failed"; +} + +fn emit( + observer: &O, + level: crate::BackfillProgressLevel, + message: impl std::convert::Into, + completed: std::option::Option, + total: std::option::Option, +) where + O: crate::BackfillObserver + Sync, +{ + let event = crate::BackfillProgressEvent::new(level, message, completed, total); + observer.on_progress(&event); +} + +#[cfg(test)] +mod tests { + fn signature(byte: u8) -> std::string::String { + return bs58::encode([byte; 64]).into_string(); + } + + fn pubkey(byte: u8) -> std::string::String { + return bs58::encode([byte; 32]).into_string(); + } + + #[test] + fn explicit_candidates_are_deduplicated_in_input_order() { + let first = signature(1); + let second = signature(2); + let discovery = super::explicit_candidates(&[first.clone(), second.clone(), first.clone()]); + assert_eq!(discovery.candidates.len(), 2); + assert_eq!(discovery.candidates[0].signature, first); + assert_eq!(discovery.candidates[1].signature, second); + } + + #[test] + fn address_request_requires_non_zero_limit() { + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Program, + address: pubkey(3), + anchor_signature: std::option::Option::Some(signature(4)), + direction: crate::BackfillDirection::Before, + limit: 0, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + assert!(request.validate().is_err()); + } + + #[test] + fn before_address_request_accepts_missing_anchor() { + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Program, + address: pubkey(40), + anchor_signature: std::option::Option::None, + direction: crate::BackfillDirection::Before, + limit: 10, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + assert!(request.validate().is_ok()); + assert_eq!(request.filter_code(), "program_latest"); + } + + #[test] + fn after_address_request_rejects_missing_anchor() { + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Program, + address: pubkey(41), + anchor_signature: std::option::Option::None, + direction: crate::BackfillDirection::After, + limit: 10, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + assert!(request.validate().is_err()); + } + + #[test] + fn filter_code_distinguishes_target_and_direction() { + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Pool, + address: pubkey(5), + anchor_signature: std::option::Option::Some(signature(6)), + direction: crate::BackfillDirection::After, + limit: 10, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + assert_eq!(request.filter_code(), "pool_after"); + } + + #[test] + fn nearest_newer_candidates_keep_only_entries_closest_to_anchor() { + let page = (1_u8..=6_u8) + .map(|byte| { + return kb_onchain_transport::AddressSignatureInfo { + signature: signature(byte), + slot: u64::from(byte), + err: std::option::Option::None, + memo: std::option::Option::None, + block_time: std::option::Option::None, + confirmation_status: std::option::Option::Some("confirmed".to_string()), + }; + }) + .collect::>(); + let mut candidates = std::collections::VecDeque::new(); + let mut seen = std::collections::HashSet::new(); + super::append_nearest_newer_candidates(&mut candidates, &mut seen, &page, 3); + let signatures = candidates + .into_iter() + .map(|candidate| return candidate.signature) + .collect::>(); + assert_eq!(signatures, vec![signature(4), signature(5), signature(6)]); + } + + #[test] + fn nearest_newer_candidates_deduplicate_page_entries() { + let duplicate = signature(7); + let page = vec![ + kb_onchain_transport::AddressSignatureInfo { + signature: duplicate.clone(), + slot: 7, + err: std::option::Option::None, + memo: std::option::Option::None, + block_time: std::option::Option::None, + confirmation_status: std::option::Option::Some("confirmed".to_string()), + }, + kb_onchain_transport::AddressSignatureInfo { + signature: duplicate, + slot: 7, + err: std::option::Option::None, + memo: std::option::Option::None, + block_time: std::option::Option::None, + confirmation_status: std::option::Option::Some("confirmed".to_string()), + }, + ]; + let mut candidates = std::collections::VecDeque::new(); + let mut seen = std::collections::HashSet::new(); + super::append_nearest_newer_candidates(&mut candidates, &mut seen, &page, 5); + assert_eq!(candidates.len(), 1); + } + + #[test] + fn lower_hex_encodes_digest_bytes_without_lower_hex_trait() { + assert_eq!(super::lower_hex(&[0x00, 0xab, 0xcd, 0xef]), "00abcdef"); + } + + #[test] + fn completion_frontier_advances_only_across_contiguous_results() { + let mut frontier = super::CandidateCompletionFrontier::new(4); + frontier.mark_completed(1); + assert_eq!(frontier.contiguous_completed(), 0); + frontier.mark_completed(0); + assert_eq!(frontier.contiguous_completed(), 2); + frontier.mark_completed(3); + assert_eq!(frontier.contiguous_completed(), 2); + frontier.mark_completed(2); + assert_eq!(frontier.contiguous_completed(), 4); + } + + #[test] + fn cancelled_before_resume_uses_last_contiguous_candidate() { + let anchor = signature(20); + let candidates = vec![signature(21), signature(22), signature(23)]; + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Program, + address: pubkey(24), + anchor_signature: std::option::Option::Some(anchor), + direction: crate::BackfillDirection::Before, + limit: 3, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + let resume = super::resolved_resume_before_signature( + &request, + candidates.as_slice(), + 2, + std::option::Option::Some(signature(25)), + true, + ); + assert_eq!(resume, std::option::Option::Some(candidates[1].clone())); + } + + #[test] + fn cancelled_before_resume_keeps_anchor_when_nothing_completed() { + let anchor = signature(30); + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Token, + address: pubkey(31), + anchor_signature: std::option::Option::Some(anchor.clone()), + direction: crate::BackfillDirection::Before, + limit: 2, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + let resume = super::resolved_resume_before_signature( + &request, + &[signature(32), signature(33)], + 0, + std::option::Option::Some(signature(34)), + true, + ); + assert_eq!(resume, std::option::Option::Some(anchor)); + } + + #[test] + fn cancelled_latest_scan_without_completed_candidate_restarts_from_latest() { + let request = crate::BackfillRequest { + role: "history_backfill".to_string(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::AddressHistory { + kind: crate::BackfillAddressKind::Program, + address: pubkey(42), + anchor_signature: std::option::Option::None, + direction: crate::BackfillDirection::Before, + limit: 2, + }, + page_size: 100, + max_pages: 10, + max_concurrent_requests: 2, + max_retries: 2, + }; + let resume = super::resolved_resume_before_signature( + &request, + &[signature(43), signature(44)], + 0, + std::option::Option::Some(signature(45)), + true, + ); + assert_eq!(resume, std::option::Option::None); + } + + #[test] + fn retry_delay_uses_configured_pause_for_rate_limit() { + assert_eq!(super::retry_delay_ms("HTTP 429", 2, 1500), 1500); + assert_eq!(super::retry_delay_ms("timeout", 2, 1500), 1000); + } + + struct TestObserver { + cancelled: std::sync::atomic::AtomicBool, + } + + impl TestObserver { + fn new(cancelled: bool) -> Self { + return Self { + cancelled: std::sync::atomic::AtomicBool::new(cancelled), + }; + } + + fn cancel(&self) { + self.cancelled.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + + impl crate::BackfillObserver for TestObserver { + fn on_progress(&self, _event: &crate::BackfillProgressEvent) {} + + fn is_cancelled(&self) -> bool { + return self.cancelled.load(std::sync::atomic::Ordering::SeqCst); + } + } + + #[tokio::test] + async fn cancellable_retry_wait_returns_immediately_when_already_cancelled() { + let observer = TestObserver::new(true); + let started = std::time::Instant::now(); + let completed = + super::wait_with_cancellation(&observer, std::time::Duration::from_secs(60)).await; + assert!(!completed); + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + } + + #[tokio::test] + async fn long_running_rpc_future_is_cancelled_cooperatively() { + let observer = TestObserver::new(false); + let operation = super::await_or_cancelled( + &observer, + tokio::time::sleep(std::time::Duration::from_secs(60)), + ); + let cancellation = async { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + observer.cancel(); + }; + let (result, ()) = tokio::join!(operation, cancellation); + assert!(result.is_none()); + } +} diff --git a/kb-pipeline/src/lib.rs b/kb-pipeline/src/lib.rs index f0d7520..5d3e62a 100644 --- a/kb-pipeline/src/lib.rs +++ b/kb-pipeline/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-pipeline/src/lib.rs -// version: 4 +// version: 5 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -7,11 +7,30 @@ //! Pipeline orchestration boundary. +mod backfill; mod constants; mod core_extraction; mod decode_replay; mod plan; +/// Address category used by one targeted backfill campaign. +pub use self::backfill::BackfillAddressKind; +/// Chronological direction relative to one anchor signature. +pub use self::backfill::BackfillDirection; +/// Observer notified during HTTP backfill campaigns. +pub use self::backfill::BackfillObserver; +/// Progress event emitted by HTTP backfill campaigns. +pub use self::backfill::BackfillProgressEvent; +/// Severity of one HTTP backfill progress event. +pub use self::backfill::BackfillProgressLevel; +/// Complete bounded HTTP backfill request. +pub use self::backfill::BackfillRequest; +/// Candidate source used by one HTTP backfill campaign. +pub use self::backfill::BackfillSource; +/// Summary returned by one HTTP backfill campaign. +pub use self::backfill::BackfillSummary; +/// Executes one bounded HTTP transaction backfill campaign. +pub use self::backfill::execute_http_backfill; /// Canonical tracing target for pipeline orchestration. pub(crate) use self::constants::TRACING_TARGET; /// Stable core extraction processor name.