Sasedev/khadhroony-solana-project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.3.6-pre.009-fix.001

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-01 16:17:22 +02:00
parent b861a1e3b8
commit 75b2d7e7f1
16 changed files with 366 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
crates/ksp-job-backfill-lib/src/checkpoint.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/checkpoint.rs
// version: 1
// version: 2
/// Opaque caller-owned checkpoint for one controlled Backfill resumption.
///
@@ -14,7 +14,7 @@ pub struct BackfillCheckpoint {
resume_before: std::option::Option<crate::BackfillSignature>,
}
impl BackfillCheckpoint {
impl crate::BackfillCheckpoint {
/// Returns the logical Job identity that owns this checkpoint.
#[must_use]
pub const fn job_id(&self) -> &ksp_job_api::JobId {
@@ -49,7 +49,7 @@ impl BackfillCheckpoint {
}
}
impl std::fmt::Debug for BackfillCheckpoint {
impl std::fmt::Debug for crate::BackfillCheckpoint {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
.debug_struct("BackfillCheckpoint")
@@ -67,7 +67,7 @@ pub(crate) struct CompletionFrontier {
contiguous_completed: usize,
}
impl CompletionFrontier {
impl crate::CompletionFrontier {
/// Creates one empty frontier for the exact bounded candidate count.
pub(crate) fn new(candidate_count: usize) -> Self {
return Self { durable: vec![false; candidate_count], contiguous_completed: 0 };
@@ -109,7 +109,7 @@ pub(crate) fn validate_request_checkpoint(
request_job_id: &ksp_job_api::JobId,
scope_fingerprint: crate::BackfillScopeFingerprint,
scope_kind: crate::BackfillScopeKind,
checkpoint: &BackfillCheckpoint,
checkpoint: &crate::BackfillCheckpoint,
) -> ksp_core_lib::Result<()> {
if checkpoint.job_id() != request_job_id {
return std::result::Result::Err(checkpoint_error("checkpoint.job_id"));
@@ -157,9 +157,9 @@ pub(crate) fn execution_resume_offset(request: &crate::BackfillRequest, discover
pub(crate) fn checkpoint_from_frontier(
request: &crate::BackfillRequest,
discovery: &crate::BackfillDiscovery,
frontier: &CompletionFrontier,
) -> ksp_core_lib::Result<BackfillCheckpoint> {
let validation = validate_discovery_identity(request, discovery);
frontier: &crate::CompletionFrontier,
) -> ksp_core_lib::Result<crate::BackfillCheckpoint> {
let validation = crate::validate_discovery_identity(request, discovery);
if let std::result::Result::Err(error) = validation {
return std::result::Result::Err(error);
}
@@ -168,12 +168,12 @@ pub(crate) fn checkpoint_from_frontier(
if let std::option::Option::Some(checkpoint) = request.checkpoint() {
return std::result::Result::Ok(checkpoint.clone());
}
return std::result::Result::Ok(BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), 0, std::option::Option::None));
return std::result::Result::Ok(crate::BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), 0, std::option::Option::None));
}
let (completed_prefix, resume_before) = match request.scope().kind() {
crate::BackfillScopeKind::LatestAddress => (local_prefix, std::option::Option::None),
crate::BackfillScopeKind::BeforeAddress => {
let previous = request.checkpoint().map_or(0, BackfillCheckpoint::completed_prefix);
let previous = request.checkpoint().map_or(0, crate::BackfillCheckpoint::completed_prefix);
let completed_prefix = match previous.checked_add(local_prefix) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix")),
@@ -195,7 +195,7 @@ pub(crate) fn checkpoint_from_frontier(
},
crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => (local_prefix, std::option::Option::None),
};
return std::result::Result::Ok(BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), completed_prefix, resume_before));
return std::result::Result::Ok(crate::BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), completed_prefix, resume_before));
}
/// Validates that one discovery result belongs to the request network and semantic scope.

28
crates/ksp-job-backfill-lib/src/conversion.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/conversion.rs
// version: 3
// version: 4
use sha2::Digest; // rust-rules: trait-import
@@ -24,7 +24,7 @@ struct BackfillRawAcquisitionInner {
observation: ksp_store_lib::RawTransactionObservation,
}
impl BackfillRawAcquisition {
impl crate::BackfillRawAcquisition {
/// Returns the canonical RAW transaction produced from the typed Transport response.
#[must_use]
pub const fn transaction(&self) -> &ksp_store_lib::RawTransaction {
@@ -49,12 +49,12 @@ impl BackfillRawAcquisition {
#[derive(Debug)]
pub enum BackfillHydrationOutcome {
/// The RPC returned one complete transaction and conversion produced canonical RAW plus provenance.
Available(BackfillRawAcquisition),
Available(crate::BackfillRawAcquisition),
/// The RPC returned JSON `null`; only the canonical transaction identity exists and no provenance is fabricated.
Missing(ksp_store_lib::RawTransactionReference),
}
impl BackfillHydrationOutcome {
impl crate::BackfillHydrationOutcome {
/// Returns the network-scoped transaction identity represented by this hydration outcome.
#[must_use]
pub fn reference(&self) -> &ksp_store_lib::RawTransactionReference {
@@ -81,7 +81,7 @@ pub async fn hydrate_backfill_candidate(
request: &crate::BackfillRequest,
candidate: &crate::BackfillCandidate,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<BackfillHydrationOutcome> {
) -> ksp_core_lib::Result<crate::BackfillHydrationOutcome> {
let reference = canonical_reference(request, candidate);
let reference = match reference {
std::result::Result::Ok(reference) => reference,
@@ -102,7 +102,7 @@ pub async fn hydrate_backfill_candidate(
let transaction = observed.into_value();
let transaction = match transaction {
std::option::Option::Some(transaction) => transaction,
std::option::Option::None => return std::result::Result::Ok(BackfillHydrationOutcome::Missing(reference)),
std::option::Option::None => return std::result::Result::Ok(crate::BackfillHydrationOutcome::Missing(reference)),
};
let fields = CanonicalTransactionFields {
slot: transaction.slot(),
@@ -114,7 +114,7 @@ pub async fn hydrate_backfill_candidate(
};
let acquisition = convert_available_fields(request, reference, fields, provider.as_str(), endpoint.as_str(), received_at);
return match acquisition {
std::result::Result::Ok(acquisition) => std::result::Result::Ok(BackfillHydrationOutcome::Available(acquisition)),
std::result::Result::Ok(acquisition) => std::result::Result::Ok(crate::BackfillHydrationOutcome::Available(acquisition)),
std::result::Result::Err(error) => std::result::Result::Err(error),
};
}
@@ -183,7 +183,7 @@ fn convert_available_fields(
provider: &str,
endpoint: &str,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<BackfillRawAcquisition> {
) -> ksp_core_lib::Result<crate::BackfillRawAcquisition> {
let block_time = convert_block_time(fields.block_time);
let block_time = match block_time {
std::result::Result::Ok(block_time) => block_time,
@@ -195,13 +195,17 @@ fn convert_available_fields(
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let hash: [u8; 32] = sha2::Sha256::digest(bytes.as_slice()).into();
let format_id = ksp_store_lib::RawFormatId::new(RAW_TRANSACTION_FORMAT_ID);
let format_id = ksp_store_lib::RawFormatId::new(crate::RAW_TRANSACTION_FORMAT_ID);
let format_id = match format_id {
std::result::Result::Ok(format_id) => format_id,
std::result::Result::Err(_) => return std::result::Result::Err(conversion_error("payload.format_id")),
};
let payload =
ksp_store_lib::RawPayload::try_new(format_id, RAW_TRANSACTION_FORMAT_VERSION, bytes.into_boxed_slice(), ksp_store_lib::RawContentHash::new(hash));
let payload = ksp_store_lib::RawPayload::try_new(
format_id,
crate::RAW_TRANSACTION_FORMAT_VERSION,
bytes.into_boxed_slice(),
ksp_store_lib::RawContentHash::new(hash),
);
let payload = match payload {
std::result::Result::Ok(payload) => payload,
std::result::Result::Err(error) => return std::result::Result::Err(error),
@@ -214,7 +218,7 @@ fn convert_available_fields(
let observation_key = observation_key(request, &reference, provider, endpoint);
let transaction = ksp_store_lib::RawTransaction::new(reference.clone(), fields.slot, block_time, payload);
let observation = ksp_store_lib::RawTransactionObservation::new(observation_key, reference, provenance);
return std::result::Result::Ok(BackfillRawAcquisition { inner: Box::new(BackfillRawAcquisitionInner { transaction, observation }) });
return std::result::Result::Ok(crate::BackfillRawAcquisition { inner: Box::new(BackfillRawAcquisitionInner { transaction, observation }) });
}
fn convert_block_time(value: std::option::Option<i64>) -> ksp_core_lib::Result<std::option::Option<ksp_store_lib::RawTimestamp>> {

78
crates/ksp-job-backfill-lib/src/discovery.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/discovery.rs
// version: 4
// version: 5
/// Network-scoped identity of one discovered transaction candidate before canonical signature decoding.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
@@ -8,7 +8,7 @@ pub struct BackfillCandidateIdentity {
signature: crate::BackfillSignature,
}
impl BackfillCandidateIdentity {
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 {
@@ -31,20 +31,20 @@ impl BackfillCandidateIdentity {
/// One deterministic transaction candidate produced by bounded discovery.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillCandidate {
identity: BackfillCandidateIdentity,
identity: crate::BackfillCandidateIdentity,
discovered_slot: std::option::Option<u64>,
}
impl BackfillCandidate {
impl crate::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 {
pub fn new(identity: crate::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 {
pub const fn identity(&self) -> &crate::BackfillCandidateIdentity {
return &self.identity;
}
@@ -70,7 +70,7 @@ pub enum BackfillDiscoveryBoundary {
AfterAnchorNotReached,
}
impl BackfillDiscoveryBoundary {
impl crate::BackfillDiscoveryBoundary {
/// Returns the stable diagnostic code for this boundary.
#[must_use]
pub const fn code(self) -> &'static str {
@@ -94,19 +94,19 @@ impl BackfillDiscoveryBoundary {
pub struct BackfillDiscovery {
network: ksp_store_lib::RawNetworkId,
scope_fingerprint: crate::BackfillScopeFingerprint,
candidates: std::vec::Vec<BackfillCandidate>,
candidates: std::vec::Vec<crate::BackfillCandidate>,
pages_fetched: usize,
boundary: BackfillDiscoveryBoundary,
boundary: crate::BackfillDiscoveryBoundary,
}
impl BackfillDiscovery {
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<BackfillCandidate>,
candidates: std::vec::Vec<crate::BackfillCandidate>,
pages_fetched: usize,
boundary: BackfillDiscoveryBoundary,
boundary: crate::BackfillDiscoveryBoundary,
) -> Self {
return Self { network, scope_fingerprint, candidates, pages_fetched, boundary };
}
@@ -125,7 +125,7 @@ impl BackfillDiscovery {
/// Returns discovered candidates in deterministic processing order.
#[must_use]
pub fn candidates(&self) -> &[BackfillCandidate] {
pub fn candidates(&self) -> &[crate::BackfillCandidate] {
return self.candidates.as_slice();
}
@@ -137,7 +137,7 @@ impl BackfillDiscovery {
/// Returns the reason discovery stopped.
#[must_use]
pub const fn boundary(&self) -> BackfillDiscoveryBoundary {
pub const fn boundary(&self) -> crate::BackfillDiscoveryBoundary {
return self.boundary;
}
@@ -148,7 +148,7 @@ impl BackfillDiscovery {
}
}
impl std::fmt::Debug for BackfillDiscovery {
impl std::fmt::Debug for crate::BackfillDiscovery {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
.debug_struct("BackfillDiscovery")
@@ -206,7 +206,7 @@ impl SignaturePageSource for ksp_onchain_transport_lib::HttpTransportPool {
pub async fn discover_backfill_candidates(
transport: &ksp_onchain_transport_lib::HttpTransportPool,
request: &crate::BackfillRequest,
) -> ksp_core_lib::Result<BackfillDiscovery> {
) -> ksp_core_lib::Result<crate::BackfillDiscovery> {
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
job_id = request.job_id().as_str(),
@@ -239,7 +239,7 @@ pub(crate) async fn discover_backfill_candidates_cancellable(
transport: &ksp_onchain_transport_lib::HttpTransportPool,
request: &crate::BackfillRequest,
cancellation: &crate::BackfillCancellationSignal,
) -> ksp_core_lib::Result<BackfillDiscovery> {
) -> ksp_core_lib::Result<crate::BackfillDiscovery> {
return discover_with_source(transport, request, std::option::Option::Some(cancellation)).await;
}
@@ -247,7 +247,7 @@ async fn discover_with_source<S>(
source: &S,
request: &crate::BackfillRequest,
cancellation: std::option::Option<&crate::BackfillCancellationSignal>,
) -> ksp_core_lib::Result<BackfillDiscovery>
) -> ksp_core_lib::Result<crate::BackfillDiscovery>
where
S: SignaturePageSource,
{
@@ -261,22 +261,22 @@ where
};
}
fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result<BackfillDiscovery> {
fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result<crate::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));
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(BackfillDiscovery::new(
return std::result::Result::Ok(crate::BackfillDiscovery::new(
request.network().clone(),
request.scope_fingerprint(),
candidates,
0,
BackfillDiscoveryBoundary::ExplicitInput,
crate::BackfillDiscoveryBoundary::ExplicitInput,
));
}
@@ -284,7 +284,7 @@ async fn discover_older<S>(
source: &S,
request: &crate::BackfillRequest,
cancellation: std::option::Option<&crate::BackfillCancellationSignal>,
) -> ksp_core_lib::Result<BackfillDiscovery>
) -> ksp_core_lib::Result<crate::BackfillDiscovery>
where
S: SignaturePageSource,
{
@@ -304,10 +304,10 @@ where
let mut pages_fetched = 0_usize;
let boundary = loop {
if candidates.len() >= request.max_candidates() {
break BackfillDiscoveryBoundary::CandidateLimit;
break crate::BackfillDiscoveryBoundary::CandidateLimit;
}
if pages_fetched >= request.max_pages() {
break BackfillDiscoveryBoundary::PageLimit;
break crate::BackfillDiscoveryBoundary::PageLimit;
}
let remaining = request.max_candidates() - candidates.len();
let page_limit = std::cmp::min(request.page_size(), remaining);
@@ -345,33 +345,33 @@ where
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)));
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 BackfillDiscoveryBoundary::RpcBoundary;
break crate::BackfillDiscoveryBoundary::RpcBoundary;
}
let next_before = match next_before {
std::option::Option::Some(next_before) => next_before,
std::option::Option::None => break BackfillDiscoveryBoundary::RpcBoundary,
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(BackfillDiscovery::new(request.network().clone(), request.scope_fingerprint(), candidates, pages_fetched, boundary));
return std::result::Result::Ok(crate::BackfillDiscovery::new(request.network().clone(), request.scope_fingerprint(), candidates, pages_fetched, boundary));
}
async fn discover_after<S>(
source: &S,
request: &crate::BackfillRequest,
cancellation: std::option::Option<&crate::BackfillCancellationSignal>,
) -> ksp_core_lib::Result<BackfillDiscovery>
) -> ksp_core_lib::Result<crate::BackfillDiscovery>
where
S: SignaturePageSource,
{
@@ -385,12 +385,12 @@ where
};
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 nearest = std::collections::VecDeque::<crate::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;
break crate::BackfillDiscoveryBoundary::AfterAnchorNotReached;
}
let config = ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig::new(
before.clone(),
@@ -426,26 +426,26 @@ where
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)));
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 BackfillDiscoveryBoundary::RpcBoundary;
break crate::BackfillDiscoveryBoundary::RpcBoundary;
}
let next_before = match next_before {
std::option::Option::Some(next_before) => next_before,
std::option::Option::None => break BackfillDiscoveryBoundary::RpcBoundary,
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(BackfillDiscovery::new(
return std::result::Result::Ok(crate::BackfillDiscovery::new(
request.network().clone(),
request.scope_fingerprint(),
nearest.into_iter().collect(),

20
crates/ksp-job-backfill-lib/src/execution.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/execution.rs
// version: 2
// version: 3
use futures_util::StreamExt; // rust-rules: trait-import
@@ -27,7 +27,7 @@ pub struct BackfillExecutionBatch {
failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
}
impl BackfillExecutionBatch {
impl crate::BackfillExecutionBatch {
/// Returns the number of candidates present in the bounded discovery result.
#[must_use]
pub const fn candidate_count(&self) -> usize {
@@ -162,7 +162,7 @@ pub(crate) struct BackfillExecutionProgress {
checkpoint: crate::BackfillCheckpoint,
}
impl BackfillExecutionProgress {
impl crate::BackfillExecutionProgress {
/// Returns the cumulative admission count.
pub(crate) const fn admitted_count(&self) -> usize {
return self.admitted_count;
@@ -243,7 +243,7 @@ pub async fn execute_backfill_discovery(
store: &ksp_store_lib::Store,
request: &crate::BackfillRequest,
discovery: &crate::BackfillDiscovery,
) -> ksp_core_lib::Result<BackfillExecutionBatch> {
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch> {
let validation = crate::validate_discovery_identity(request, discovery);
if let std::result::Result::Err(error) = validation {
return std::result::Result::Err(error);
@@ -260,7 +260,7 @@ pub(crate) async fn execute_backfill_discovery_cancellable(
discovery: &crate::BackfillDiscovery,
cancellation: &crate::BackfillCancellationSignal,
publisher: &crate::BackfillRuntimePublisher,
) -> ksp_core_lib::Result<BackfillExecutionBatch> {
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch> {
let processor = RuntimeCandidateProcessor { transport, store, request, cancellation: std::option::Option::Some(cancellation) };
return execute_with_processor(&processor, request, discovery, std::option::Option::Some(cancellation), std::option::Option::Some(publisher)).await;
}
@@ -311,7 +311,7 @@ async fn execute_with_processor<P>(
discovery: &crate::BackfillDiscovery,
cancellation: std::option::Option<&crate::BackfillCancellationSignal>,
publisher: std::option::Option<&crate::BackfillRuntimePublisher>,
) -> ksp_core_lib::Result<BackfillExecutionBatch>
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch>
where
P: CandidateProcessor,
{
@@ -484,7 +484,7 @@ where
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return std::result::Result::Ok(BackfillExecutionBatch {
return std::result::Result::Ok(crate::BackfillExecutionBatch {
candidate_count: discovery.candidates().len(),
admitted_count,
finished_count,
@@ -524,12 +524,12 @@ fn progress_from_state(
cancelled_count: usize,
hole_count: usize,
maximum_in_flight: usize,
) -> ksp_core_lib::Result<BackfillExecutionProgress> {
) -> ksp_core_lib::Result<crate::BackfillExecutionProgress> {
let checkpoint = match crate::checkpoint_from_frontier(request, discovery, frontier) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return std::result::Result::Ok(BackfillExecutionProgress {
return std::result::Result::Ok(crate::BackfillExecutionProgress {
admitted_count,
finished_count,
inserted_count,
@@ -549,7 +549,7 @@ fn progress_from_state(
fn publish_progress(
publisher: std::option::Option<&crate::BackfillRuntimePublisher>,
progress: &BackfillExecutionProgress,
progress: &crate::BackfillExecutionProgress,
cancelling: bool,
draining: bool,
) -> ksp_core_lib::Result<()> {

6
crates/ksp-job-backfill-lib/src/lib.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/lib.rs
// version: 5
// version: 6
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -139,9 +139,5 @@ pub(crate) use self::execution::BackfillExecutionProgress;
pub(crate) use self::execution::execute_backfill_discovery_cancellable;
/// Internal concrete cancellation signal shared by discovery and execution.
pub(crate) use self::runtime::BackfillCancellationSignal;
/// Internal atomic terminal/cancellation arbitration shared by runtime tests and coordinator.
pub(crate) use self::runtime::BackfillRuntimeControl;
/// Internal execution progress publisher feeding the latest-value snapshot source.
pub(crate) use self::runtime::BackfillRuntimePublisher;
/// Internal terminal race result used by coordinator canaries.
pub(crate) use self::runtime::TerminalClaim;

60
crates/ksp-job-backfill-lib/src/persistence.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/persistence.rs
// version: 2
// version: 3
/// Canonical entity disposition produced by one Backfill Store persistence attempt.
#[non_exhaustive]
@@ -38,16 +38,16 @@ pub enum BackfillObservationPersistence {
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillPersistenceOutcome {
reference: ksp_store_lib::RawTransactionReference,
entity: BackfillEntityPersistence,
observation: BackfillObservationPersistence,
entity: crate::BackfillEntityPersistence,
observation: crate::BackfillObservationPersistence,
}
impl BackfillPersistenceOutcome {
impl crate::BackfillPersistenceOutcome {
/// Creates one internally classified persistence result.
pub(crate) fn new(
reference: ksp_store_lib::RawTransactionReference,
entity: BackfillEntityPersistence,
observation: BackfillObservationPersistence,
entity: crate::BackfillEntityPersistence,
observation: crate::BackfillObservationPersistence,
) -> Self {
return Self { reference, entity, observation };
}
@@ -60,13 +60,13 @@ impl BackfillPersistenceOutcome {
/// Returns the canonical transaction persistence disposition.
#[must_use]
pub const fn entity(&self) -> BackfillEntityPersistence {
pub const fn entity(&self) -> crate::BackfillEntityPersistence {
return self.entity;
}
/// Returns the acquisition-observation persistence disposition.
#[must_use]
pub const fn observation(&self) -> BackfillObservationPersistence {
pub const fn observation(&self) -> crate::BackfillObservationPersistence {
return self.observation;
}
}
@@ -81,7 +81,7 @@ impl BackfillPersistenceOutcome {
pub async fn persist_backfill_hydration(
store: &ksp_store_lib::Store,
hydration: crate::BackfillHydrationOutcome,
) -> ksp_core_lib::Result<BackfillPersistenceOutcome> {
) -> ksp_core_lib::Result<crate::BackfillPersistenceOutcome> {
return persist_hydration_with_port(store, hydration).await;
}
@@ -112,7 +112,7 @@ impl RawTransactionPersistencePort for ksp_store_lib::Store {
}
}
async fn persist_hydration_with_port<P>(port: &P, hydration: crate::BackfillHydrationOutcome) -> ksp_core_lib::Result<BackfillPersistenceOutcome>
async fn persist_hydration_with_port<P>(port: &P, hydration: crate::BackfillHydrationOutcome) -> ksp_core_lib::Result<crate::BackfillPersistenceOutcome>
where
P: RawTransactionPersistencePort,
{
@@ -121,10 +121,10 @@ where
return std::result::Result::Err(persistence_error("store.network"));
}
return match hydration {
crate::BackfillHydrationOutcome::Missing(_) => std::result::Result::Ok(BackfillPersistenceOutcome::new(
crate::BackfillHydrationOutcome::Missing(_) => std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::Missing,
BackfillObservationPersistence::NotApplicable,
crate::BackfillEntityPersistence::Missing,
crate::BackfillObservationPersistence::NotApplicable,
)),
crate::BackfillHydrationOutcome::Available(acquisition) => {
let (transaction, observation) = acquisition.into_parts();
@@ -138,7 +138,7 @@ async fn persist_available_with_port<P>(
reference: ksp_store_lib::RawTransactionReference,
transaction: ksp_store_lib::RawTransaction,
observation: ksp_store_lib::RawTransactionObservation,
) -> ksp_core_lib::Result<BackfillPersistenceOutcome>
) -> ksp_core_lib::Result<crate::BackfillPersistenceOutcome>
where
P: RawTransactionPersistencePort,
{
@@ -153,10 +153,10 @@ where
std::result::Result::Ok(outcome) => map_store_outcome(reference, outcome),
std::result::Result::Err(error) => {
if error.code() == ksp_store_lib::ERROR_CODE_RAW_CONFLICT {
return std::result::Result::Ok(BackfillPersistenceOutcome::new(
return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::Conflict,
BackfillObservationPersistence::NotRecorded,
crate::BackfillEntityPersistence::Conflict,
crate::BackfillObservationPersistence::NotRecorded,
));
}
std::result::Result::Err(error)
@@ -167,35 +167,35 @@ where
fn map_store_outcome(
reference: ksp_store_lib::RawTransactionReference,
outcome: ksp_store_lib::RawAcquisitionWriteOutcome,
) -> ksp_core_lib::Result<BackfillPersistenceOutcome> {
) -> ksp_core_lib::Result<crate::BackfillPersistenceOutcome> {
let entity = outcome.entity();
let observation = outcome.observation();
if entity == ksp_store_lib::RawEntityWriteOutcome::Inserted && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
return std::result::Result::Ok(BackfillPersistenceOutcome::new(
return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::Inserted,
BackfillObservationPersistence::Inserted,
crate::BackfillEntityPersistence::Inserted,
crate::BackfillObservationPersistence::Inserted,
));
}
if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
return std::result::Result::Ok(BackfillPersistenceOutcome::new(
return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::AlreadyPresent,
BackfillObservationPersistence::Inserted,
crate::BackfillEntityPersistence::AlreadyPresent,
crate::BackfillObservationPersistence::Inserted,
));
}
if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent {
return std::result::Result::Ok(BackfillPersistenceOutcome::new(
return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::AlreadyPresent,
BackfillObservationPersistence::AlreadyPresent,
crate::BackfillEntityPersistence::AlreadyPresent,
crate::BackfillObservationPersistence::AlreadyPresent,
));
}
if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded {
return std::result::Result::Ok(BackfillPersistenceOutcome::new(
return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(
reference,
BackfillEntityPersistence::SkippedPurged,
BackfillObservationPersistence::NotRecorded,
crate::BackfillEntityPersistence::SkippedPurged,
crate::BackfillObservationPersistence::NotRecorded,
));
}
return std::result::Result::Err(persistence_error("store.outcome"));

74
crates/ksp-job-backfill-lib/src/request.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/request.rs
// version: 3
// version: 4
use sha2::Digest; // rust-rules: trait-import
@@ -25,7 +25,7 @@ pub enum BackfillCommitment {
Finalized,
}
impl BackfillCommitment {
impl crate::BackfillCommitment {
/// Returns the stable Backfill commitment code.
#[must_use]
pub const fn code(self) -> &'static str {
@@ -52,7 +52,7 @@ impl BackfillCommitment {
#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct BackfillSignature(std::string::String);
impl BackfillSignature {
impl crate::BackfillSignature {
/// Creates one bounded Base58-shaped signature text.
pub fn new(value: impl std::convert::Into<std::string::String>) -> ksp_core_lib::Result<Self> {
let value = value.into();
@@ -74,7 +74,7 @@ impl BackfillSignature {
}
}
impl std::fmt::Debug for BackfillSignature {
impl std::fmt::Debug for crate::BackfillSignature {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.write_str("BackfillSignature(..)");
}
@@ -93,7 +93,7 @@ pub enum BackfillScopeKind {
ExplicitSignatures,
}
impl BackfillScopeKind {
impl crate::BackfillScopeKind {
/// Returns the stable scope code used by diagnostics and scope fingerprinting.
#[must_use]
pub const fn code(self) -> &'static str {
@@ -109,9 +109,9 @@ impl BackfillScopeKind {
#[derive(Clone, Eq, PartialEq)]
enum BackfillScopeValue {
LatestAddress { address: ksp_core_lib::Pubkey },
BeforeAddress { address: ksp_core_lib::Pubkey, anchor: BackfillSignature },
AfterAddress { address: ksp_core_lib::Pubkey, anchor: BackfillSignature },
ExplicitSignatures { signatures: std::vec::Vec<BackfillSignature> },
BeforeAddress { address: ksp_core_lib::Pubkey, anchor: crate::BackfillSignature },
AfterAddress { address: ksp_core_lib::Pubkey, anchor: crate::BackfillSignature },
ExplicitSignatures { signatures: std::vec::Vec<crate::BackfillSignature> },
}
/// Validated bounded discovery scope for one historical Backfill Job.
@@ -120,7 +120,7 @@ pub struct BackfillScope {
value: BackfillScopeValue,
}
impl BackfillScope {
impl crate::BackfillScope {
/// Creates a scope starting at the newest known history for one address.
#[must_use]
pub const fn latest_address(address: ksp_core_lib::Pubkey) -> Self {
@@ -129,18 +129,18 @@ impl BackfillScope {
/// Creates a scope reading history older than one exclusive address anchor.
#[must_use]
pub fn before_address(address: ksp_core_lib::Pubkey, anchor: BackfillSignature) -> Self {
pub fn before_address(address: ksp_core_lib::Pubkey, anchor: crate::BackfillSignature) -> Self {
return Self { value: BackfillScopeValue::BeforeAddress { address, anchor } };
}
/// Creates a scope reading the bounded newer history closest to one exclusive address anchor.
#[must_use]
pub fn after_address(address: ksp_core_lib::Pubkey, anchor: BackfillSignature) -> Self {
pub fn after_address(address: ksp_core_lib::Pubkey, anchor: crate::BackfillSignature) -> Self {
return Self { value: BackfillScopeValue::AfterAddress { address, anchor } };
}
/// Creates an explicit signature scope with stable first-occurrence deduplication.
pub fn explicit_signatures(signatures: std::vec::Vec<BackfillSignature>) -> ksp_core_lib::Result<Self> {
pub fn explicit_signatures(signatures: std::vec::Vec<crate::BackfillSignature>) -> ksp_core_lib::Result<Self> {
if signatures.is_empty() || signatures.len() > crate::MAX_BACKFILL_CANDIDATES {
return std::result::Result::Err(request_error("scope.signatures"));
}
@@ -159,12 +159,12 @@ impl BackfillScope {
/// Returns the stable category of this scope.
#[must_use]
pub const fn kind(&self) -> BackfillScopeKind {
pub const fn kind(&self) -> crate::BackfillScopeKind {
return match &self.value {
BackfillScopeValue::LatestAddress { .. } => BackfillScopeKind::LatestAddress,
BackfillScopeValue::BeforeAddress { .. } => BackfillScopeKind::BeforeAddress,
BackfillScopeValue::AfterAddress { .. } => BackfillScopeKind::AfterAddress,
BackfillScopeValue::ExplicitSignatures { .. } => BackfillScopeKind::ExplicitSignatures,
BackfillScopeValue::LatestAddress { .. } => crate::BackfillScopeKind::LatestAddress,
BackfillScopeValue::BeforeAddress { .. } => crate::BackfillScopeKind::BeforeAddress,
BackfillScopeValue::AfterAddress { .. } => crate::BackfillScopeKind::AfterAddress,
BackfillScopeValue::ExplicitSignatures { .. } => crate::BackfillScopeKind::ExplicitSignatures,
};
}
@@ -181,7 +181,7 @@ impl BackfillScope {
/// Returns the exclusive anchor used by before/after scopes.
#[must_use]
pub const fn anchor(&self) -> std::option::Option<&BackfillSignature> {
pub const fn anchor(&self) -> std::option::Option<&crate::BackfillSignature> {
return match &self.value {
BackfillScopeValue::BeforeAddress { anchor, .. } | BackfillScopeValue::AfterAddress { anchor, .. } => std::option::Option::Some(anchor),
BackfillScopeValue::LatestAddress { .. } | BackfillScopeValue::ExplicitSignatures { .. } => std::option::Option::None,
@@ -190,7 +190,7 @@ impl BackfillScope {
/// Returns the stable deduplicated explicit signature list, when this is an explicit scope.
#[must_use]
pub fn signatures(&self) -> std::option::Option<&[BackfillSignature]> {
pub fn signatures(&self) -> std::option::Option<&[crate::BackfillSignature]> {
return match &self.value {
BackfillScopeValue::ExplicitSignatures { signatures } => std::option::Option::Some(signatures.as_slice()),
BackfillScopeValue::LatestAddress { .. } | BackfillScopeValue::BeforeAddress { .. } | BackfillScopeValue::AfterAddress { .. } => {
@@ -200,7 +200,7 @@ impl BackfillScope {
}
}
impl std::fmt::Debug for BackfillScope {
impl std::fmt::Debug for crate::BackfillScope {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut debug = formatter.debug_struct("BackfillScope");
debug.field("kind", &self.kind());
@@ -223,7 +223,7 @@ impl std::fmt::Debug for BackfillScope {
#[derive(Clone, Copy, Eq, Hash, PartialEq)]
pub struct BackfillScopeFingerprint([u8; 32]);
impl BackfillScopeFingerprint {
impl crate::BackfillScopeFingerprint {
/// Returns the exact deterministic fingerprint bytes.
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
@@ -231,7 +231,7 @@ impl BackfillScopeFingerprint {
}
}
impl std::fmt::Debug for BackfillScopeFingerprint {
impl std::fmt::Debug for crate::BackfillScopeFingerprint {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.write_str("BackfillScopeFingerprint(..)");
}
@@ -243,26 +243,26 @@ pub struct BackfillRequest {
job_id: ksp_job_api::JobId,
network: ksp_store_lib::RawNetworkId,
role: ksp_onchain_transport_lib::HttpRoleName,
commitment: BackfillCommitment,
scope: BackfillScope,
commitment: crate::BackfillCommitment,
scope: crate::BackfillScope,
page_size: usize,
max_pages: usize,
max_candidates: usize,
hydration_concurrency: usize,
min_context_slot: std::option::Option<u64>,
scope_fingerprint: BackfillScopeFingerprint,
scope_fingerprint: crate::BackfillScopeFingerprint,
checkpoint: std::option::Option<crate::BackfillCheckpoint>,
}
impl BackfillRequest {
impl crate::BackfillRequest {
/// Creates and validates one fully explicit bounded Backfill request.
#[allow(clippy::too_many_arguments)]
pub fn new(
job_id: ksp_job_api::JobId,
network: ksp_store_lib::RawNetworkId,
role: ksp_onchain_transport_lib::HttpRoleName,
commitment: BackfillCommitment,
scope: BackfillScope,
commitment: crate::BackfillCommitment,
scope: crate::BackfillScope,
page_size: usize,
max_pages: usize,
max_candidates: usize,
@@ -284,7 +284,7 @@ impl BackfillRequest {
if role.as_str().is_empty() || role.as_str().trim() != role.as_str() {
return std::result::Result::Err(request_error("role"));
}
if scope.kind() == BackfillScopeKind::ExplicitSignatures {
if scope.kind() == crate::BackfillScopeKind::ExplicitSignatures {
if min_context_slot.is_some() {
return std::result::Result::Err(request_error("min_context_slot"));
}
@@ -333,13 +333,13 @@ impl BackfillRequest {
/// Returns the narrowed commitment used by discovery and hydration.
#[must_use]
pub const fn commitment(&self) -> BackfillCommitment {
pub const fn commitment(&self) -> crate::BackfillCommitment {
return self.commitment;
}
/// Returns the validated discovery scope.
#[must_use]
pub const fn scope(&self) -> &BackfillScope {
pub const fn scope(&self) -> &crate::BackfillScope {
return &self.scope;
}
@@ -378,7 +378,7 @@ impl BackfillRequest {
/// Transport role, provider, endpoint and protocol are deliberately excluded. They describe
/// acquisition provenance, not transaction or scope identity.
#[must_use]
pub const fn scope_fingerprint(&self) -> BackfillScopeFingerprint {
pub const fn scope_fingerprint(&self) -> crate::BackfillScopeFingerprint {
return self.scope_fingerprint;
}
@@ -399,7 +399,7 @@ impl BackfillRequest {
}
}
impl std::fmt::Debug for BackfillRequest {
impl std::fmt::Debug for crate::BackfillRequest {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
.debug_struct("BackfillRequest")
@@ -421,13 +421,13 @@ impl std::fmt::Debug for BackfillRequest {
fn fingerprint_scope(
network: &ksp_store_lib::RawNetworkId,
commitment: BackfillCommitment,
scope: &BackfillScope,
commitment: crate::BackfillCommitment,
scope: &crate::BackfillScope,
page_size: usize,
max_pages: usize,
max_candidates: usize,
min_context_slot: std::option::Option<u64>,
) -> BackfillScopeFingerprint {
) -> crate::BackfillScopeFingerprint {
let mut hasher = sha2::Sha256::new();
hasher.update(b"ksp.job.backfill.scope.v1\0");
hash_bytes(&mut hasher, network.as_str().as_bytes());
@@ -455,7 +455,7 @@ fn fingerprint_scope(
std::option::Option::None => hasher.update([0_u8]),
}
let bytes: [u8; 32] = hasher.finalize().into();
return BackfillScopeFingerprint(bytes);
return crate::BackfillScopeFingerprint(bytes);
}
fn hash_bytes(hasher: &mut sha2::Sha256, value: &[u8]) {

120
crates/ksp-job-backfill-lib/src/runtime.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/src/runtime.rs
// version: 2
// version: 3
/// Stable Job kind code used by the concrete historical RAW transaction Backfill runtime.
pub const BACKFILL_JOB_KIND_CODE: &str = "solana.raw_transaction.backfill";
@@ -26,7 +26,7 @@ pub enum BackfillJobPhase {
Finished,
}
impl BackfillJobPhase {
impl crate::BackfillJobPhase {
/// Returns the stable safe code for this concrete runtime phase.
#[must_use]
pub const fn code(self) -> &'static str {
@@ -43,7 +43,7 @@ impl BackfillJobPhase {
/// Complete safe latest-value snapshot of one concrete historical RAW transaction Backfill Job.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillJobSnapshot {
phase: BackfillJobPhase,
phase: crate::BackfillJobPhase,
scope_kind: crate::BackfillScopeKind,
discovery_boundary: std::option::Option<crate::BackfillDiscoveryBoundary>,
candidates_selected: usize,
@@ -64,10 +64,10 @@ pub struct BackfillJobSnapshot {
failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
}
impl BackfillJobSnapshot {
impl crate::BackfillJobSnapshot {
fn initial(request: &crate::BackfillRequest) -> Self {
return Self {
phase: BackfillJobPhase::Created,
phase: crate::BackfillJobPhase::Created,
scope_kind: request.scope().kind(),
discovery_boundary: std::option::Option::None,
candidates_selected: 0,
@@ -91,7 +91,7 @@ impl BackfillJobSnapshot {
/// Returns the concrete execution phase represented by this snapshot.
#[must_use]
pub const fn phase(&self) -> BackfillJobPhase {
pub const fn phase(&self) -> crate::BackfillJobPhase {
return self.phase;
}
@@ -207,11 +207,11 @@ impl BackfillJobSnapshot {
/// Cloneable runtime-neutral-facing latest-value source for concrete Backfill snapshots.
#[derive(Clone)]
pub struct BackfillSnapshotSource {
receiver: tokio::sync::watch::Receiver<ksp_job_api::JobNotification<BackfillJobSnapshot>>,
receiver: tokio::sync::watch::Receiver<ksp_job_api::JobNotification<crate::BackfillJobSnapshot>>,
}
impl ksp_job_api::JobSnapshotSource for BackfillSnapshotSource {
type Snapshot = BackfillJobSnapshot;
impl ksp_job_api::JobSnapshotSource for crate::BackfillSnapshotSource {
type Snapshot = crate::BackfillJobSnapshot;
fn current(&self) -> ksp_job_api::JobNotification<Self::Snapshot> {
return self.receiver.borrow().clone();
@@ -234,7 +234,7 @@ impl ksp_job_api::JobSnapshotSource for BackfillSnapshotSource {
}
}
impl std::fmt::Debug for BackfillSnapshotSource {
impl std::fmt::Debug for crate::BackfillSnapshotSource {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let current = self.receiver.borrow();
return formatter.debug_struct("BackfillSnapshotSource").field("sequence", &current.sequence()).field("state", &current.state()).finish();
@@ -245,10 +245,10 @@ impl std::fmt::Debug for BackfillSnapshotSource {
#[derive(Clone)]
pub struct BackfillJobHandle {
control: BackfillRuntimeControl,
snapshots: BackfillSnapshotSource,
snapshots: crate::BackfillSnapshotSource,
}
impl BackfillJobHandle {
impl crate::BackfillJobHandle {
/// Requests cooperative cancellation and returns `true` only when accepted before terminal publication.
#[must_use]
pub fn cancel(&self) -> bool {
@@ -257,7 +257,7 @@ impl BackfillJobHandle {
/// Returns an independent latest-value snapshot source for one listener.
#[must_use]
pub fn snapshots(&self) -> BackfillSnapshotSource {
pub fn snapshots(&self) -> crate::BackfillSnapshotSource {
return self.snapshots.clone();
}
@@ -268,12 +268,12 @@ impl BackfillJobHandle {
}
}
impl std::fmt::Debug for BackfillJobHandle {
impl std::fmt::Debug for crate::BackfillJobHandle {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
.debug_struct("BackfillJobHandle")
.field("cancellation_requested", &self.is_cancellation_requested())
.field("snapshot", &<BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&self.snapshots))
.field("snapshot", &<crate::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&self.snapshots))
.finish();
}
}
@@ -282,19 +282,19 @@ impl std::fmt::Debug for BackfillJobHandle {
pub struct BackfillJobRuntime {
request: crate::BackfillRequest,
control: BackfillRuntimeControl,
cancellation: BackfillCancellationSignal,
publisher: BackfillRuntimePublisher,
handle: BackfillJobHandle,
cancellation: crate::BackfillCancellationSignal,
publisher: crate::BackfillRuntimePublisher,
handle: crate::BackfillJobHandle,
}
impl BackfillJobRuntime {
impl crate::BackfillJobRuntime {
/// Creates one runtime in `Created` state and its stable latest-value channel.
pub fn new(request: crate::BackfillRequest) -> ksp_core_lib::Result<Self> {
let kind = match ksp_job_api::JobKindCode::new(BACKFILL_JOB_KIND_CODE) {
let kind = match ksp_job_api::JobKindCode::new(crate::BACKFILL_JOB_KIND_CODE) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let initial_snapshot = BackfillJobSnapshot::initial(&request);
let initial_snapshot = crate::BackfillJobSnapshot::initial(&request);
let initial = ksp_job_api::JobNotification::new(
request.job_id().clone(),
kind.clone(),
@@ -305,16 +305,16 @@ impl BackfillJobRuntime {
let (snapshot_sender, snapshot_receiver) = tokio::sync::watch::channel(initial);
let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false);
let control = BackfillRuntimeControl::new(cancel_sender);
let cancellation = BackfillCancellationSignal::new(control.token(), cancel_receiver);
let snapshots = BackfillSnapshotSource { receiver: snapshot_receiver };
let handle = BackfillJobHandle { control: control.clone(), snapshots: snapshots.clone() };
let publisher = BackfillRuntimePublisher { sender: snapshot_sender, id: request.job_id().clone(), kind };
let cancellation = crate::BackfillCancellationSignal::new(control.token(), cancel_receiver);
let snapshots = crate::BackfillSnapshotSource { receiver: snapshot_receiver };
let handle = crate::BackfillJobHandle { control: control.clone(), snapshots: snapshots.clone() };
let publisher = crate::BackfillRuntimePublisher { sender: snapshot_sender, id: request.job_id().clone(), kind };
return std::result::Result::Ok(Self { request, control, cancellation, publisher, handle });
}
/// Returns a cloneable control and latest-value observation handle before the runtime is moved into execution.
#[must_use]
pub fn handle(&self) -> BackfillJobHandle {
pub fn handle(&self) -> crate::BackfillJobHandle {
return self.handle.clone();
}
@@ -323,7 +323,7 @@ impl BackfillJobRuntime {
self,
transport: &ksp_onchain_transport_lib::HttpTransportPool,
store: &ksp_store_lib::Store,
) -> ksp_core_lib::Result<BackfillJobSnapshot> {
) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
if self.cancellation.is_requested() {
let claimed = self.control.claim_normal_terminal();
if claimed != TerminalClaim::Cancelled {
@@ -331,7 +331,7 @@ impl BackfillJobRuntime {
}
return self.publisher.publish_cancelled_from_created();
}
let started = self.publisher.publish_running(BackfillJobPhase::Discovering);
let started = self.publisher.publish_running(crate::BackfillJobPhase::Discovering);
if let std::result::Result::Err(error) = started {
self.control.claim_failed();
return std::result::Result::Err(error);
@@ -393,7 +393,7 @@ impl BackfillJobRuntime {
};
}
fn finish_cancelled(&self) -> ksp_core_lib::Result<BackfillJobSnapshot> {
fn finish_cancelled(&self) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
let terminal = self.control.claim_normal_terminal();
if terminal != TerminalClaim::Cancelled {
return std::result::Result::Err(runtime_error("terminal.cancelled"));
@@ -408,7 +408,7 @@ impl BackfillJobRuntime {
/// Internal atomic terminal/cancellation arbitration shared by runtime and external handle.
#[derive(Clone)]
pub(crate) struct BackfillRuntimeControl {
struct BackfillRuntimeControl {
state: std::sync::Arc<std::sync::atomic::AtomicU8>,
token: ksp_job_api::JobCancellationToken,
cancel_sender: tokio::sync::watch::Sender<bool>,
@@ -416,7 +416,7 @@ pub(crate) struct BackfillRuntimeControl {
impl BackfillRuntimeControl {
/// Creates one active control state paired with the cancellation wake channel.
pub(crate) fn new(cancel_sender: tokio::sync::watch::Sender<bool>) -> Self {
fn new(cancel_sender: tokio::sync::watch::Sender<bool>) -> Self {
return Self {
state: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(CONTROL_ACTIVE)),
token: ksp_job_api::JobCancellationToken::new(),
@@ -425,12 +425,12 @@ impl BackfillRuntimeControl {
}
/// Returns the runtime-neutral cancellation token mirrored by this control.
pub(crate) fn token(&self) -> ksp_job_api::JobCancellationToken {
fn token(&self) -> ksp_job_api::JobCancellationToken {
return self.token.clone();
}
/// Atomically accepts the first pre-terminal cancellation request.
pub(crate) fn request_cancellation(&self) -> bool {
fn request_cancellation(&self) -> bool {
let accepted = self
.state
.compare_exchange(CONTROL_ACTIVE, CONTROL_CANCELLATION_REQUESTED, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire)
@@ -442,12 +442,12 @@ impl BackfillRuntimeControl {
}
/// Returns whether cooperative cancellation was accepted.
pub(crate) fn is_cancellation_requested(&self) -> bool {
fn is_cancellation_requested(&self) -> bool {
return self.token.is_cancellation_requested();
}
/// Atomically resolves the completion-versus-cancellation terminal race.
pub(crate) fn claim_normal_terminal(&self) -> TerminalClaim {
fn claim_normal_terminal(&self) -> TerminalClaim {
let completed =
self.state.compare_exchange(CONTROL_ACTIVE, CONTROL_COMPLETED, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire);
if completed.is_ok() {
@@ -466,7 +466,7 @@ impl BackfillRuntimeControl {
}
/// Marks a non-terminal control as failed, overriding a pending cancellation request.
pub(crate) fn claim_failed(&self) {
fn claim_failed(&self) {
loop {
let state = self.state.load(std::sync::atomic::Ordering::Acquire);
if matches!(state, CONTROL_COMPLETED | CONTROL_CANCELLED | CONTROL_FAILED) {
@@ -487,7 +487,7 @@ pub(crate) struct BackfillCancellationSignal {
receiver: tokio::sync::watch::Receiver<bool>,
}
impl BackfillCancellationSignal {
impl crate::BackfillCancellationSignal {
/// Creates one signal from the runtime-neutral token and Tokio wake receiver.
pub(crate) fn new(token: ksp_job_api::JobCancellationToken, receiver: tokio::sync::watch::Receiver<bool>) -> Self {
return Self { token, receiver };
@@ -529,7 +529,7 @@ async fn wait_for_cancellation(token: &ksp_job_api::JobCancellationToken, receiv
/// Internal result of atomically claiming a normal terminal state.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum TerminalClaim {
enum TerminalClaim {
Completed,
Cancelled,
Failed,
@@ -538,23 +538,23 @@ pub(crate) enum TerminalClaim {
/// Internal latest-value publisher owning the concrete Backfill notification stream.
#[derive(Clone)]
pub(crate) struct BackfillRuntimePublisher {
sender: tokio::sync::watch::Sender<ksp_job_api::JobNotification<BackfillJobSnapshot>>,
sender: tokio::sync::watch::Sender<ksp_job_api::JobNotification<crate::BackfillJobSnapshot>>,
id: ksp_job_api::JobId,
kind: ksp_job_api::JobKindCode,
}
impl BackfillRuntimePublisher {
impl crate::BackfillRuntimePublisher {
/// Publishes one non-terminal running phase.
pub(crate) fn publish_running(&self, phase: BackfillJobPhase) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_running(&self, phase: crate::BackfillJobPhase) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(ksp_job_api::JobState::Running, |snapshot| {
snapshot.phase = phase;
});
}
/// Publishes the complete bounded discovery result as the current execution snapshot.
pub(crate) fn publish_discovery(&self, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_discovery(&self, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(ksp_job_api::JobState::Running, |snapshot| {
snapshot.phase = BackfillJobPhase::Executing;
snapshot.phase = crate::BackfillJobPhase::Executing;
snapshot.discovery_boundary = std::option::Option::Some(discovery.boundary());
snapshot.candidates_selected = discovery.candidates().len();
});
@@ -566,25 +566,25 @@ impl BackfillRuntimePublisher {
progress: &crate::BackfillExecutionProgress,
cancelling: bool,
draining: bool,
) -> ksp_core_lib::Result<BackfillJobSnapshot> {
) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
let state = if cancelling { ksp_job_api::JobState::Cancelling } else { ksp_job_api::JobState::Running };
return self.publish_with(state, |snapshot| {
snapshot.phase = if draining { BackfillJobPhase::Draining } else { BackfillJobPhase::Executing };
snapshot.phase = if draining { crate::BackfillJobPhase::Draining } else { crate::BackfillJobPhase::Executing };
apply_progress(snapshot, progress);
});
}
/// Publishes cancellation observation before terminal cancellation.
pub(crate) fn publish_cancelling(&self) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_cancelling(&self) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(ksp_job_api::JobState::Cancelling, |snapshot| {
snapshot.phase = BackfillJobPhase::Draining;
snapshot.phase = crate::BackfillJobPhase::Draining;
});
}
/// Publishes the drained batch state while cancellation is terminalizing.
pub(crate) fn publish_batch_cancelling(&self, batch: &crate::BackfillExecutionBatch) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_batch_cancelling(&self, batch: &crate::BackfillExecutionBatch) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(ksp_job_api::JobState::Cancelling, |snapshot| {
snapshot.phase = BackfillJobPhase::Draining;
snapshot.phase = crate::BackfillJobPhase::Draining;
apply_batch(snapshot, batch);
});
}
@@ -595,16 +595,16 @@ impl BackfillRuntimePublisher {
batch: &crate::BackfillExecutionBatch,
state: ksp_job_api::JobState,
failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
) -> ksp_core_lib::Result<BackfillJobSnapshot> {
) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(state, |snapshot| {
snapshot.phase = BackfillJobPhase::Finished;
snapshot.phase = crate::BackfillJobPhase::Finished;
snapshot.failure_code = failure_code;
apply_batch(snapshot, batch);
});
}
/// Publishes a terminal failure before a batch exists.
pub(crate) fn publish_failed(&self, code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_failed(&self, code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_terminal(ksp_job_api::JobState::Failed, std::option::Option::Some(code));
}
@@ -613,23 +613,23 @@ impl BackfillRuntimePublisher {
&self,
state: ksp_job_api::JobState,
failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
) -> ksp_core_lib::Result<BackfillJobSnapshot> {
) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(state, |snapshot| {
snapshot.phase = BackfillJobPhase::Finished;
snapshot.phase = crate::BackfillJobPhase::Finished;
snapshot.failure_code = failure_code;
});
}
/// Publishes direct Created-to-Cancelled termination before execution starts.
pub(crate) fn publish_cancelled_from_created(&self) -> ksp_core_lib::Result<BackfillJobSnapshot> {
pub(crate) fn publish_cancelled_from_created(&self) -> ksp_core_lib::Result<crate::BackfillJobSnapshot> {
return self.publish_with(ksp_job_api::JobState::Cancelled, |snapshot| {
snapshot.phase = BackfillJobPhase::Finished;
snapshot.phase = crate::BackfillJobPhase::Finished;
});
}
fn publish_with<F>(&self, state: ksp_job_api::JobState, update: F) -> ksp_core_lib::Result<BackfillJobSnapshot>
fn publish_with<F>(&self, state: ksp_job_api::JobState, update: F) -> ksp_core_lib::Result<crate::BackfillJobSnapshot>
where
F: FnOnce(&mut BackfillJobSnapshot),
F: FnOnce(&mut crate::BackfillJobSnapshot),
{
let current = self.sender.borrow().clone();
if current.state().is_terminal() {
@@ -668,7 +668,7 @@ fn valid_snapshot_transition(source: ksp_job_api::JobState, target: ksp_job_api:
);
}
fn apply_progress(snapshot: &mut BackfillJobSnapshot, progress: &crate::BackfillExecutionProgress) {
fn apply_progress(snapshot: &mut crate::BackfillJobSnapshot, progress: &crate::BackfillExecutionProgress) {
snapshot.candidates_admitted = progress.admitted_count();
snapshot.candidates_finished = progress.finished_count();
snapshot.entities_inserted = progress.inserted_count();
@@ -685,7 +685,7 @@ fn apply_progress(snapshot: &mut BackfillJobSnapshot, progress: &crate::Backfill
snapshot.checkpoint = std::option::Option::Some(progress.checkpoint().clone());
}
fn apply_batch(snapshot: &mut BackfillJobSnapshot, batch: &crate::BackfillExecutionBatch) {
fn apply_batch(snapshot: &mut crate::BackfillJobSnapshot, batch: &crate::BackfillExecutionBatch) {
snapshot.candidates_admitted = batch.admitted_count();
snapshot.candidates_finished = batch.finished_count();
snapshot.entities_inserted = batch.inserted_count();

6
crates/ksp-job-backfill-lib/tests/public_api.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/tests/public_api.rs
// version: 6
// version: 7
//! Public API canaries through the concrete Backfill cancellation and latest-value runtime tranche.
@@ -109,13 +109,13 @@ fn pre_008_checkpoint_and_bounded_execution_contracts_are_available_from_crate_r
#[test]
fn pre_009_concrete_runtime_snapshot_and_control_contracts_are_available_from_crate_root() {
fn assert_source<T>()
fn assert_source<T>(_: std::marker::PhantomData<T>)
where
T: ksp_job_api::JobSnapshotSource<Snapshot = ksp_job_backfill_lib::BackfillJobSnapshot>,
{
return;
}
assert_source::<ksp_job_backfill_lib::BackfillSnapshotSource>();
assert_source(std::marker::PhantomData::<ksp_job_backfill_lib::BackfillSnapshotSource>);
let _runtime_new = ksp_job_backfill_lib::BackfillJobRuntime::new;
let _handle: std::option::Option<ksp_job_backfill_lib::BackfillJobHandle> = std::option::Option::None;
let _phase = ksp_job_backfill_lib::BackfillJobPhase::Discovering;

9
crates/ksp-job-backfill-lib/unit_tests/discovery.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/unit_tests/discovery.rs
// version: 5
// version: 6
#[derive(Clone, Debug, Eq, PartialEq)]
struct PageCall {
@@ -392,13 +392,14 @@ async fn pre_009_discovery_rpc_wait_is_cancelled_cooperatively() {
std::option::Option::None => return,
};
let source = PendingSource::new();
let token = ksp_job_api::JobCancellationToken::new();
let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false);
let control = crate::BackfillRuntimeControl::new(cancel_sender);
let cancellation = crate::BackfillCancellationSignal::new(control.token(), cancel_receiver);
let cancellation = crate::BackfillCancellationSignal::new(token.clone(), cancel_receiver);
let discovery = super::discover_with_source(&source, &request, std::option::Option::Some(&cancellation));
let cancel = async {
tokio::task::yield_now().await;
assert!(control.request_cancellation());
assert!(token.cancel());
assert!(cancel_sender.send(true).is_ok());
};
let (result, ()) = tokio::join!(discovery, cancel);
let error = match result {

9
crates/ksp-job-backfill-lib/unit_tests/execution.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/unit_tests/execution.rs
// version: 2
// version: 3
#[derive(Clone, Copy)]
enum FakeDisposition {
@@ -291,9 +291,9 @@ async fn pre_009_cancellation_stops_admission_and_drains_already_admitted_candid
FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
]);
let token = ksp_job_api::JobCancellationToken::new();
let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false);
let control = crate::runtime::BackfillRuntimeControl::new(cancel_sender);
let signal = crate::runtime::BackfillCancellationSignal::new(control.token(), cancel_receiver);
let signal = crate::BackfillCancellationSignal::new(token.clone(), cancel_receiver);
let execution = super::execute_with_processor(&processor, &request, &discovery, std::option::Option::Some(&signal), std::option::Option::None);
let cancellation = async {
loop {
@@ -302,7 +302,8 @@ async fn pre_009_cancellation_stops_admission_and_drains_already_admitted_candid
}
tokio::task::yield_now().await;
}
assert!(control.request_cancellation());
assert!(token.cancel());
assert!(cancel_sender.send(true).is_ok());
};
let (result, ()) = tokio::join!(execution, cancellation);
let batch = match result {

18
crates/ksp-job-backfill-lib/unit_tests/runtime.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/unit_tests/runtime.rs
// version: 2
// version: 3
use ksp_job_api::JobSnapshotSource; // rust-rules: trait-import
@@ -78,7 +78,7 @@ async fn pre_009_terminal_snapshot_is_retained_and_late_cancellation_is_rejected
let listener = handle.snapshots();
let started = runtime.publisher.publish_running(crate::BackfillJobPhase::Discovering);
assert!(started.is_ok());
assert_eq!(runtime.control.claim_normal_terminal(), crate::TerminalClaim::Completed);
assert_eq!(runtime.control.claim_normal_terminal(), super::TerminalClaim::Completed);
let terminal = runtime.publisher.publish_terminal(ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Complete), std::option::Option::None);
assert!(terminal.is_ok());
assert!(!handle.cancel());
@@ -94,13 +94,13 @@ async fn pre_009_terminal_snapshot_is_retained_and_late_cancellation_is_rejected
#[test]
fn pre_009_terminal_race_is_first_decision_wins_for_cancellation_vs_completion() {
let (cancel_sender, _) = tokio::sync::watch::channel(false);
let cancellation_first = crate::BackfillRuntimeControl::new(cancel_sender);
let cancellation_first = super::BackfillRuntimeControl::new(cancel_sender);
assert!(cancellation_first.request_cancellation());
assert_eq!(cancellation_first.claim_normal_terminal(), crate::TerminalClaim::Cancelled);
assert_eq!(cancellation_first.claim_normal_terminal(), super::TerminalClaim::Cancelled);
assert!(!cancellation_first.request_cancellation());
let (cancel_sender, _) = tokio::sync::watch::channel(false);
let completion_first = crate::BackfillRuntimeControl::new(cancel_sender);
assert_eq!(completion_first.claim_normal_terminal(), crate::TerminalClaim::Completed);
let completion_first = super::BackfillRuntimeControl::new(cancel_sender);
assert_eq!(completion_first.claim_normal_terminal(), super::TerminalClaim::Completed);
assert!(!completion_first.request_cancellation());
return;
}
@@ -108,10 +108,10 @@ fn pre_009_terminal_race_is_first_decision_wins_for_cancellation_vs_completion()
#[test]
fn pre_009_fatal_failure_overrides_pending_cancellation_before_terminal_publication() {
let (cancel_sender, _) = tokio::sync::watch::channel(false);
let control = crate::BackfillRuntimeControl::new(cancel_sender);
let control = super::BackfillRuntimeControl::new(cancel_sender);
assert!(control.request_cancellation());
control.claim_failed();
assert_eq!(control.claim_normal_terminal(), crate::TerminalClaim::Failed);
assert_eq!(control.claim_normal_terminal(), super::TerminalClaim::Failed);
assert!(!control.request_cancellation());
return;
}
@@ -119,7 +119,7 @@ fn pre_009_fatal_failure_overrides_pending_cancellation_before_terminal_publicat
#[tokio::test]
async fn pre_009_long_running_pre_store_future_is_cancelled_cooperatively() {
let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false);
let control = crate::BackfillRuntimeControl::new(cancel_sender);
let control = super::BackfillRuntimeControl::new(cancel_sender);
let signal = crate::BackfillCancellationSignal::new(control.token(), cancel_receiver);
let operation = std::future::pending::<ksp_core_lib::Result<usize>>();
let wait = signal.run_cancellable(operation);