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

v0.3.14-pre.005

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-11 23:15:00 +02:00
parent 1006079654
commit a3df8f6ef5
6 changed files with 784 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs
View File

@@ -1,6 +1,8 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs
// version: 4
// version: 5
/// Maximum number of proven redundant coverage epochs retained by one run-local continuity contract.
const MAX_RAW_TRANSACTION_INGEST_COVERAGE_EPOCHS: usize = 256;
/// Maximum number of simultaneously retained non-repaired run-local gaps.
const MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS: usize = 64;
/// Maximum number of run-local gaps that may actively repair at once.
@@ -23,6 +25,12 @@ pub(crate) enum RawTransactionIngestCoverageScope {
KnownReferences([u8; 32]),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RawTransactionIngestCoverageRelation {
Exact,
Superset,
}
impl crate::RawTransactionIngestCoverageScope {
/// Creates one provider-neutral exact source scope from a stable family code and semantic fingerprint.
pub(crate) const fn exact_source_scope(family_code: &'static str, fingerprint: [u8; 32]) -> Self {
@@ -54,6 +62,19 @@ impl crate::RawTransactionIngestCoverageScope {
fn is_known_references(&self) -> bool {
return matches!(self, Self::KnownReferences(_));
}
fn relation_to(&self, requirement: &Self) -> std::option::Option<RawTransactionIngestCoverageRelation> {
return match (self, requirement) {
(Self::FullLedgerTransactions, Self::FullLedgerTransactions) => std::option::Option::Some(RawTransactionIngestCoverageRelation::Exact),
(Self::FullLedgerTransactions, Self::ExactSourceScope(_, _)) => std::option::Option::Some(RawTransactionIngestCoverageRelation::Superset),
(Self::ExactSourceScope(family_code, fingerprint), Self::ExactSourceScope(required_family_code, required_fingerprint))
if family_code == required_family_code && fingerprint == required_fingerprint =>
{
std::option::Option::Some(RawTransactionIngestCoverageRelation::Exact)
},
_ => std::option::Option::None,
};
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -424,14 +445,14 @@ impl RawTransactionIngestTargetCoverage {
if scope.is_known_references() {
return std::result::Result::Err(crate::runtime_error("continuity.known_references_not_target_scope"));
}
if matches!(&scope, crate::RawTransactionIngestCoverageScope::FullLedgerTransactions) {
self.requirements.retain(|requirement| return requirement.commitment != commitment);
self.requirements.push(RawTransactionIngestCoverageRequirement { commitment, scope });
return std::result::Result::Ok(());
}
self.requirements.retain(|requirement| {
if requirement.commitment != commitment {
return true;
}
return scope.relation_to(&requirement.scope) != std::option::Option::Some(RawTransactionIngestCoverageRelation::Superset);
});
if self.requirements.iter().any(|requirement| {
return requirement.commitment == commitment
&& (requirement.scope == crate::RawTransactionIngestCoverageScope::FullLedgerTransactions || requirement.scope == scope);
return requirement.commitment == commitment && requirement.scope.relation_to(&scope).is_some();
}) {
return std::result::Result::Ok(());
}
@@ -440,9 +461,133 @@ impl RawTransactionIngestTargetCoverage {
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct RawTransactionIngestCoverageRange {
end_slot: u64,
start_slot: u64,
}
impl RawTransactionIngestCoverageRange {
fn new(start_slot: u64, end_slot: u64) -> ksp_core_lib::Result<Self> {
if end_slot < start_slot {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_range_reversed"));
}
return std::result::Result::Ok(Self { end_slot, start_slot });
}
const fn contains_gap(self, gap: RawTransactionIngestGapRange) -> bool {
return self.start_slot <= gap.start_slot() && self.end_slot >= gap.end_slot();
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct RawTransactionIngestCoverageEpoch {
commitment: ksp_onchain_transport_lib::SolanaCommitment,
epoch_id: u64,
range: RawTransactionIngestCoverageRange,
scope: crate::RawTransactionIngestCoverageScope,
source_key: [u8; 32],
}
impl RawTransactionIngestCoverageEpoch {
fn new(
epoch_id: u64,
source_key: [u8; 32],
commitment: ksp_onchain_transport_lib::SolanaCommitment,
scope: crate::RawTransactionIngestCoverageScope,
range: RawTransactionIngestCoverageRange,
) -> ksp_core_lib::Result<Self> {
if epoch_id == 0 {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_id_invalid"));
}
if let std::result::Result::Err(error) = scope.validate_configured_target_scope() {
return std::result::Result::Err(error);
}
return std::result::Result::Ok(Self { commitment, epoch_id, range, scope, source_key });
}
fn relation_to_requirement(
&self,
target_source_key: [u8; 32],
requirement: &RawTransactionIngestCoverageRequirement,
gap: RawTransactionIngestGapRange,
) -> std::option::Option<RawTransactionIngestCoverageRelation> {
if self.source_key == target_source_key || self.commitment != requirement.commitment || !self.range.contains_gap(gap) {
return std::option::Option::None;
}
return self.scope.relation_to(&requirement.scope);
}
}
struct RawTransactionIngestCoverageEpochLedger {
epochs: std::vec::Vec<RawTransactionIngestCoverageEpoch>,
next_epoch_id: u64,
}
impl RawTransactionIngestCoverageEpochLedger {
fn new() -> Self {
return Self { epochs: std::vec::Vec::new(), next_epoch_id: 1 };
}
fn redundant_relation_for_gap(
&self,
target_source_key: [u8; 32],
requirement: &RawTransactionIngestCoverageRequirement,
gap: RawTransactionIngestGapRange,
) -> std::option::Option<RawTransactionIngestCoverageRelation> {
let mut exact = false;
for epoch in &self.epochs {
match epoch.relation_to_requirement(target_source_key, requirement, gap) {
std::option::Option::Some(RawTransactionIngestCoverageRelation::Superset) => {
return std::option::Option::Some(RawTransactionIngestCoverageRelation::Superset);
},
std::option::Option::Some(RawTransactionIngestCoverageRelation::Exact) => exact = true,
std::option::Option::None => {},
}
}
if exact {
return std::option::Option::Some(RawTransactionIngestCoverageRelation::Exact);
}
return std::option::Option::None;
}
fn validate_invariants(&self, capabilities: &[crate::RawTransactionIngestContinuityCapabilityDescriptor]) -> ksp_core_lib::Result<()> {
if self.next_epoch_id == 0 || self.epochs.len() > MAX_RAW_TRANSACTION_INGEST_COVERAGE_EPOCHS {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_inventory_invalid"));
}
let mut ids = std::collections::BTreeSet::new();
let mut highest_epoch_id = 0_u64;
for epoch in &self.epochs {
if epoch.epoch_id == 0 || !ids.insert(epoch.epoch_id) {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_id_invalid"));
}
highest_epoch_id = highest_epoch_id.max(epoch.epoch_id);
if let std::result::Result::Err(error) = epoch.scope.validate_configured_target_scope() {
return std::result::Result::Err(error);
}
if RawTransactionIngestCoverageRange::new(epoch.range.start_slot, epoch.range.end_slot).is_err() {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_range_invalid"));
}
let matching_capability = capabilities.iter().find(|capability| return capability.source_key == epoch.source_key);
let matching_capability = match matching_capability {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_source_unknown")),
};
if matching_capability.commitment != epoch.commitment || matching_capability.source_scope != epoch.scope {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_capability_mismatch"));
}
}
if !self.epochs.is_empty() && self.next_epoch_id <= highest_epoch_id {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_next_id_invalid"));
}
return std::result::Result::Ok(());
}
}
/// Run-local continuity contracts prepared from the exact caller-composed live-source aggregate before productive source execution.
pub(crate) struct RawTransactionIngestContinuityContracts {
capabilities: std::vec::Vec<crate::RawTransactionIngestContinuityCapabilityDescriptor>,
coverage_epochs: RawTransactionIngestCoverageEpochLedger,
gap_ledger: RawTransactionIngestGapLedger,
target_coverage: RawTransactionIngestTargetCoverage,
}
@@ -470,11 +615,15 @@ impl crate::RawTransactionIngestContinuityContracts {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let coverage_epochs = RawTransactionIngestCoverageEpochLedger::new();
if let std::result::Result::Err(error) = coverage_epochs.validate_invariants(capabilities.as_slice()) {
return std::result::Result::Err(error);
}
let gap_ledger = RawTransactionIngestGapLedger::new(network);
if let std::result::Result::Err(error) = gap_ledger.validate_invariants() {
return std::result::Result::Err(error);
}
if !coverage_scope_catalog_is_complete() {
if !coverage_scope_catalog_is_complete() || !coverage_relation_catalog_is_complete() {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_scope_catalog_invalid"));
}
if MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT == 0
@@ -483,7 +632,7 @@ impl crate::RawTransactionIngestContinuityContracts {
{
return std::result::Result::Err(crate::runtime_error("continuity.repair_bounds_invalid"));
}
return std::result::Result::Ok(Self { capabilities, gap_ledger, target_coverage });
return std::result::Result::Ok(Self { capabilities, coverage_epochs, gap_ledger, target_coverage });
}
/// Revalidates that the run-local contracts still correspond exactly to the caller-composed source count before tasks are spawned.
@@ -491,6 +640,9 @@ impl crate::RawTransactionIngestContinuityContracts {
if self.capabilities.len() != expected_source_count || self.target_coverage.requirements.is_empty() {
return std::result::Result::Err(crate::runtime_error("continuity.source_inventory_mismatch"));
}
if let std::result::Result::Err(error) = self.coverage_epochs.validate_invariants(self.capabilities.as_slice()) {
return std::result::Result::Err(error);
}
if let std::result::Result::Err(error) = self.gap_ledger.validate_invariants() {
return std::result::Result::Err(error);
}
@@ -510,6 +662,68 @@ fn coverage_scope_catalog_is_complete() -> bool {
&& known.is_known_references();
}
fn coverage_relation_catalog_is_complete() -> bool {
let target_scope = crate::RawTransactionIngestCoverageScope::ExactSourceScope("standard_logs", [7_u8; 32]);
let target_requirement =
RawTransactionIngestCoverageRequirement { commitment: ksp_onchain_transport_lib::SolanaCommitment::Confirmed, scope: target_scope.clone() };
let gap = match RawTransactionIngestGapRange::new(100, 110) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let full_range = match RawTransactionIngestCoverageRange::new(90, 120) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let partial_range = match RawTransactionIngestCoverageRange::new(100, 109) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let exact_epoch =
match RawTransactionIngestCoverageEpoch::new(1, [2_u8; 32], ksp_onchain_transport_lib::SolanaCommitment::Confirmed, target_scope, full_range) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let full_epoch = match RawTransactionIngestCoverageEpoch::new(
2,
[3_u8; 32],
ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
crate::RawTransactionIngestCoverageScope::FullLedgerTransactions,
full_range,
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let cross_family_epoch = match RawTransactionIngestCoverageEpoch::new(
3,
[4_u8; 32],
ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
crate::RawTransactionIngestCoverageScope::ExactSourceScope("helius_transaction", [7_u8; 32]),
full_range,
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let partial_epoch = match RawTransactionIngestCoverageEpoch::new(
4,
[5_u8; 32],
ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
crate::RawTransactionIngestCoverageScope::FullLedgerTransactions,
partial_range,
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return false,
};
let ledger = RawTransactionIngestCoverageEpochLedger {
epochs: std::vec![exact_epoch.clone(), full_epoch.clone(), cross_family_epoch.clone(), partial_epoch.clone()],
next_epoch_id: 5,
};
return ledger.redundant_relation_for_gap([1_u8; 32], &target_requirement, gap)
== std::option::Option::Some(RawTransactionIngestCoverageRelation::Superset)
&& exact_epoch.relation_to_requirement([2_u8; 32], &target_requirement, gap).is_none()
&& cross_family_epoch.relation_to_requirement([1_u8; 32], &target_requirement, gap).is_none()
&& partial_epoch.relation_to_requirement([1_u8; 32], &target_requirement, gap).is_none();
}
#[cfg(test)]
#[path = "../unit_tests/continuity.rs"]
mod tests;