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

v0.3.14-pre.008

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-12 09:00:36 +02:00
parent 9fc30d3ab9
commit 505cc8c3f6
9 changed files with 1184 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs
// version: 7
// version: 8
/// Maximum number of slots admitted by one private continuity HTTP discovery window outside this module.
pub(crate) const MAX_RAW_TRANSACTION_INGEST_CONTINUITY_DISCOVERY_WINDOW_SLOTS: u64 = MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS;
@@ -89,6 +89,7 @@ struct RawTransactionIngestCoverageRequirement {
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RawTransactionIngestGapReason {
HttpProducedBlockUnavailable,
KnownReferenceMissing,
SourceFailure,
TransportOverflow,
WebSocketReconnect,
@@ -96,8 +97,14 @@ enum RawTransactionIngestGapReason {
}
impl RawTransactionIngestGapReason {
const ALL: [Self; 5] =
[Self::HttpProducedBlockUnavailable, Self::SourceFailure, Self::TransportOverflow, Self::WebSocketReconnect, Self::YellowstoneRetention];
const ALL: [Self; 6] = [
Self::HttpProducedBlockUnavailable,
Self::KnownReferenceMissing,
Self::SourceFailure,
Self::TransportOverflow,
Self::WebSocketReconnect,
Self::YellowstoneRetention,
];
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -116,6 +123,15 @@ impl RawTransactionIngestGapState {
}
}
/// Private supervisor decision for one terminal live-source loss after conservative continuity reconciliation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RawTransactionIngestSourceLossDecision {
/// Remaining active sources explicitly preserve all configured target coverage and no known gap blocks continuity.
Continue,
/// Coverage is incomplete, a known gap remains open, or the source-loss proof is otherwise insufficient.
Fault,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct RawTransactionIngestGapRange {
end_slot: u64,
@@ -267,7 +283,6 @@ impl crate::RawTransactionIngestWebSocketIncidentAnchor {
}
/// Returns the first source slot that safely anchors the incident, inclusively.
#[cfg(test)]
pub(crate) const fn start_slot(self) -> u64 {
return self.start_slot;
}
@@ -315,6 +330,84 @@ impl RawTransactionIngestGapLedger {
return Self { gaps: std::vec::Vec::new(), network, next_gap_id: 1 };
}
fn continuity_frontier(&self, processing_frontier_slot: std::option::Option<u64>) -> std::option::Option<u64> {
let processing_frontier_slot = match processing_frontier_slot {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::option::Option::None,
};
let earliest_open_gap_start = self.gaps.iter().filter(|gap| return gap.state.is_open()).map(|gap| return gap.range.start_slot()).min();
let earliest_open_gap_start = match earliest_open_gap_start {
std::option::Option::Some(value) if value <= processing_frontier_slot => value,
std::option::Option::Some(_) | std::option::Option::None => return std::option::Option::Some(processing_frontier_slot),
};
return earliest_open_gap_start.checked_sub(1);
}
fn has_open_gaps(&self) -> bool {
return self.gaps.iter().any(|gap| return gap.state.is_open());
}
fn record_gap(
&mut self,
source_key: [u8; 32],
commitment: ksp_onchain_transport_lib::SolanaCommitment,
coverage_requirement: crate::RawTransactionIngestCoverageScope,
reason: RawTransactionIngestGapReason,
range: RawTransactionIngestGapRange,
) -> ksp_core_lib::Result<()> {
for gap in &mut self.gaps {
if !gap.state.is_open()
|| gap.source_key != source_key
|| gap.commitment != commitment
|| gap.coverage_requirement != coverage_requirement
|| gap.reason != reason
{
continue;
}
let merged = match gap.range.try_merge(range) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if let std::option::Option::Some(merged) = merged {
gap.range = merged;
return self.validate_invariants();
}
}
let open_gap_count = self.gaps.iter().filter(|gap| return gap.state.is_open()).count();
if open_gap_count >= MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS {
return std::result::Result::Err(crate::runtime_error("continuity.open_gap_limit_exceeded"));
}
let gap_id = self.next_gap_id;
self.next_gap_id = match self.next_gap_id.checked_add(1) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.gap_id_exhausted")),
};
self.gaps.push(RawTransactionIngestGap {
commitment,
coverage_requirement,
gap_id: RawTransactionIngestGapId(gap_id),
network: self.network.clone(),
range,
reason,
source_key,
state: RawTransactionIngestGapState::Pending,
});
return self.validate_invariants();
}
fn reconcile_with_coverage_epochs(&mut self, coverage_epochs: &RawTransactionIngestCoverageEpochLedger) -> ksp_core_lib::Result<()> {
for gap in &mut self.gaps {
if !gap.state.is_open() {
continue;
}
let requirement = RawTransactionIngestCoverageRequirement { commitment: gap.commitment, scope: gap.coverage_requirement.clone() };
if coverage_epochs.redundant_relation_for_gap(gap.source_key, &requirement, gap.range).is_some() {
gap.state = RawTransactionIngestGapState::Repaired;
}
}
return self.validate_invariants();
}
fn validate_invariants(&self) -> ksp_core_lib::Result<()> {
if self.next_gap_id == 0 {
return std::result::Result::Err(crate::runtime_error("continuity.gap_id_exhausted"));
@@ -382,7 +475,7 @@ impl RawTransactionIngestGapLedger {
}
}
}
if RawTransactionIngestGapReason::ALL.len() != 5 || RawTransactionIngestGapState::ALL.len() != 4 {
if RawTransactionIngestGapReason::ALL.len() != 6 || RawTransactionIngestGapState::ALL.len() != 4 {
return std::result::Result::Err(crate::runtime_error("continuity.gap_catalog_invalid"));
}
return std::result::Result::Ok(());
@@ -477,6 +570,21 @@ impl RawTransactionIngestTargetCoverage {
return std::result::Result::Ok(value);
}
fn is_covered_by_active_sources(
&self,
capabilities: &[crate::RawTransactionIngestContinuityCapabilityDescriptor],
active_source_keys: &std::collections::BTreeSet<[u8; 32]>,
) -> bool {
return self.requirements.iter().all(|requirement| {
return capabilities.iter().any(|capability| {
if !active_source_keys.contains(&capability.source_key) || capability.commitment != requirement.commitment {
return false;
}
return capability.source_scope.relation_to(&requirement.scope).is_some();
});
});
}
fn include(
&mut self,
commitment: ksp_onchain_transport_lib::SolanaCommitment,
@@ -518,6 +626,16 @@ impl RawTransactionIngestCoverageRange {
const fn contains_gap(self, gap: RawTransactionIngestGapRange) -> bool {
return self.start_slot <= gap.start_slot() && self.end_slot >= gap.end_slot();
}
fn try_merge(self, other: Self) -> std::option::Option<Self> {
let overlaps = self.start_slot <= other.end_slot && other.start_slot <= self.end_slot;
let adjacent = self.end_slot.checked_add(1) == std::option::Option::Some(other.start_slot)
|| other.end_slot.checked_add(1) == std::option::Option::Some(self.start_slot);
if !overlaps && !adjacent {
return std::option::Option::None;
}
return std::option::Option::Some(Self { end_slot: self.end_slot.max(other.end_slot), start_slot: self.start_slot.min(other.start_slot) });
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -569,6 +687,47 @@ impl RawTransactionIngestCoverageEpochLedger {
return Self { epochs: std::vec::Vec::new(), next_epoch_id: 1 };
}
fn record(
&mut self,
source_key: [u8; 32],
commitment: ksp_onchain_transport_lib::SolanaCommitment,
scope: crate::RawTransactionIngestCoverageScope,
range: RawTransactionIngestCoverageRange,
capabilities: &[crate::RawTransactionIngestContinuityCapabilityDescriptor],
) -> ksp_core_lib::Result<()> {
let matching_capability = capabilities.iter().find(|capability| return capability.source_key == 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 != commitment || matching_capability.source_scope != scope {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_capability_mismatch"));
}
for epoch in &mut self.epochs {
if epoch.source_key != source_key || epoch.commitment != commitment || epoch.scope != scope {
continue;
}
if let std::option::Option::Some(merged) = epoch.range.try_merge(range) {
epoch.range = merged;
return self.validate_invariants(capabilities);
}
}
if self.epochs.len() >= MAX_RAW_TRANSACTION_INGEST_COVERAGE_EPOCHS {
return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_inventory_invalid"));
}
let epoch_id = self.next_epoch_id;
self.next_epoch_id = match self.next_epoch_id.checked_add(1) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_inventory_invalid")),
};
let epoch = match RawTransactionIngestCoverageEpoch::new(epoch_id, source_key, commitment, scope, range) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
self.epochs.push(epoch);
return self.validate_invariants(capabilities);
}
fn redundant_relation_for_gap(
&self,
target_source_key: [u8; 32],
@@ -675,6 +834,111 @@ impl crate::RawTransactionIngestContinuityContracts {
return std::result::Result::Ok(Self { capabilities, coverage_epochs, gap_ledger, target_coverage });
}
/// Records one interval that a configured source has actually proven covered during this run.
pub(crate) fn record_coverage_epoch(&mut self, source_key: [u8; 32], start_slot: u64, end_slot: u64) -> ksp_core_lib::Result<()> {
let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key);
let capability = match capability {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_source_unknown")),
};
let range = match RawTransactionIngestCoverageRange::new(start_slot, end_slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let commitment = capability.commitment;
let scope = capability.source_scope.clone();
if let std::result::Result::Err(error) = self.coverage_epochs.record(source_key, commitment, scope, range, self.capabilities.as_slice()) {
return std::result::Result::Err(error);
}
return self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs);
}
/// Records one unresolved known-reference obligation as a single-slot continuity gap without blocking the processing frontier.
pub(crate) fn record_known_reference_gap(&mut self, source_key: [u8; 32], slot: u64) -> ksp_core_lib::Result<()> {
let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key);
let capability = match capability {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.known_reference_source_unknown")),
};
let range = match RawTransactionIngestGapRange::new(slot, slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if let std::result::Result::Err(error) = self.gap_ledger.record_gap(
source_key,
capability.commitment,
capability.source_scope.clone(),
RawTransactionIngestGapReason::KnownReferenceMissing,
range,
) {
return std::result::Result::Err(error);
}
return self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs);
}
/// Records one bounded run-local source-loss gap using the exact configured capability as its required coverage.
pub(crate) fn record_source_loss_gap(&mut self, source_key: [u8; 32], start_slot: u64, end_slot: u64) -> ksp_core_lib::Result<()> {
let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key);
let capability = match capability {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_unknown_source")),
};
let range = match RawTransactionIngestGapRange::new(start_slot, end_slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return self.gap_ledger.record_gap(
source_key,
capability.commitment,
capability.source_scope.clone(),
RawTransactionIngestGapReason::SourceFailure,
range,
);
}
/// Returns the gap-aware continuity frontier without conflating it with the processing frontier.
///
/// The processing frontier is only an upper bound: any known open continuity gap at or below it clamps the returned frontier to the slot immediately
/// before that gap. A gap beginning at slot zero yields no continuity frontier.
pub(crate) fn continuity_frontier(&self, processing_frontier_slot: std::option::Option<u64>) -> std::option::Option<u64> {
return self.gap_ledger.continuity_frontier(processing_frontier_slot);
}
/// Reconciles known gaps against already-proven coverage and decides whether one lost source may remain absent without stopping sibling sources.
///
/// Continuation requires all configured `TargetCoverage` requirements to remain covered by distinct currently active sources and requires the known
/// gap ledger to be fully reconciled. The method never treats configuration alone as historical coverage evidence and never respawns a source.
pub(crate) fn source_loss_decision(
&mut self,
lost_source_key: [u8; 32],
active_source_keys: &[[u8; 32]],
processing_frontier_slot: std::option::Option<u64>,
) -> ksp_core_lib::Result<crate::RawTransactionIngestSourceLossDecision> {
if !self.capabilities.iter().any(|capability| return capability.source_key == lost_source_key) {
return std::result::Result::Err(crate::runtime_error("continuity.source_loss_unknown_source"));
}
let mut active = std::collections::BTreeSet::new();
for source_key in active_source_keys {
if *source_key == lost_source_key || !active.insert(*source_key) {
return std::result::Result::Err(crate::runtime_error("continuity.source_loss_active_set_invalid"));
}
if !self.capabilities.iter().any(|capability| return capability.source_key == *source_key) {
return std::result::Result::Err(crate::runtime_error("continuity.source_loss_active_source_unknown"));
}
}
if let std::result::Result::Err(error) = self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs) {
return std::result::Result::Err(error);
}
let continuity_frontier = self.continuity_frontier(processing_frontier_slot);
if self.gap_ledger.has_open_gaps() || continuity_frontier != processing_frontier_slot {
return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault);
}
if !self.target_coverage.is_covered_by_active_sources(self.capabilities.as_slice(), &active) {
return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault);
}
return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Continue);
}
/// Revalidates that the run-local contracts still correspond exactly to the caller-composed source count before tasks are spawned.
pub(crate) fn validate_for_source_count(&self, expected_source_count: usize) -> ksp_core_lib::Result<()> {
if self.capabilities.len() != expected_source_count || self.target_coverage.requirements.is_empty() {

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

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
// version: 31
// version: 32
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -18,7 +18,8 @@
//! Transaction/TransactionStatus/Block and standard logs notifications become signature/slot references; BlockMeta/Slot remain continuity-only signals.
//! Hydration is coalesced by network/signature/commitment under bounded in-flight and pending budgets. A bounded run-local processing frontier projects
//! hydration pending, oldest pending slot and highest unblocked actually observed slot. The productive source also projects safe Transport reconnect/replay
//! state and faults conservatively when Transport proves a replay-retention continuity gap; history remediation remains outside this crate.
//! state. Run-local continuity contracts reconcile bounded source-loss and known-reference gaps only from explicit coverage epochs; a lost source may remain
//! absent only while currently active siblings preserve the configured TargetCoverage. Transport source respawn remains forbidden.
mod admission;
mod continuity;
@@ -121,6 +122,8 @@ pub(crate) use self::continuity::RawTransactionIngestContinuityContracts;
pub(crate) use self::continuity::RawTransactionIngestCoverageScope;
/// Private run-local obligation for one transaction reference already observed by a live source.
pub(crate) use self::continuity::RawTransactionIngestKnownReferenceObligation;
/// Private supervisor decision after one live source becomes terminal.
pub(crate) use self::continuity::RawTransactionIngestSourceLossDecision;
/// Private run-local WebSocket incident anchor built only from observed source slots and safe Transport counters.
pub(crate) use self::continuity::RawTransactionIngestWebSocketIncidentAnchor;
/// Creates one terminal content-conflict error without copying conflicting material into diagnostics.

353
crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
// version: 34
// version: 35
use sha2::Digest; // rust-rules: trait-import
@@ -88,6 +88,14 @@ enum RawTransactionIngestLiveSource {
Yellowstone(crate::RawTransactionIngestYellowstoneSource),
}
#[derive(Clone)]
struct RawTransactionIngestSourceRuntimeShared {
continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
global_hydration_registry: std::sync::Arc<RawTransactionIngestGlobalHydrationRegistry>,
hydration_in_flight_limit: usize,
hydration_pending_limit: usize,
}
struct RawTransactionIngestSourceInventory {
source_keys: std::vec::Vec<[u8; 32]>,
source_projections: std::vec::Vec<crate::RawTransactionIngestProcessingFrontierProjection>,
@@ -211,56 +219,20 @@ impl RawTransactionIngestLiveSource {
settings: crate::RawTransactionIngestSettings,
stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
global_hydration_registry: std::sync::Arc<RawTransactionIngestGlobalHydrationRegistry>,
hydration_pending_limit: usize,
hydration_in_flight_limit: usize,
inventory_publisher: RawTransactionIngestSourceInventoryPublisher,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let (source_frontier_sender, mut source_frontier_receiver) =
tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
let mut source_future = std::boxed::Box::pin(async move {
return match self {
Self::HeliusTransaction(source) => {
source
.run(
settings,
stop_receiver,
admission_sender,
source_frontier_sender,
global_hydration_registry,
hydration_pending_limit,
hydration_in_flight_limit,
)
.await
Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
Self::HttpBlockPolling(source) => {
source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts).await
},
Self::HttpBlockPolling(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await,
Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await,
Self::StandardLogs(source) => {
source
.run(
settings,
stop_receiver,
admission_sender,
source_frontier_sender,
global_hydration_registry,
hydration_pending_limit,
hydration_in_flight_limit,
)
.await
},
Self::Yellowstone(source) => {
source
.run(
settings,
stop_receiver,
admission_sender,
source_frontier_sender,
global_hydration_registry,
hydration_pending_limit,
hydration_in_flight_limit,
)
.await
},
Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
};
});
loop {
@@ -389,6 +361,23 @@ impl RawTransactionIngestSourceInventory {
return Self { source_keys, source_projections };
}
fn supervisor_state(&self) -> ksp_core_lib::Result<(std::vec::Vec<[u8; 32]>, std::option::Option<u64>)> {
if self.source_keys.len() != self.source_projections.len() {
return std::result::Result::Err(crate::runtime_error("source.inventory_shape_mismatch"));
}
let mut active_source_keys = std::vec::Vec::new();
for (source_key, projection) in self.source_keys.iter().zip(self.source_projections.iter()) {
if projection.source_state() == std::option::Option::Some(crate::RawTransactionIngestSourceState::Active) {
active_source_keys.push(*source_key);
}
}
let aggregate = match self.aggregate() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return std::result::Result::Ok((active_source_keys, aggregate.processing_frontier_slot()));
}
fn update(
&mut self,
entry_index: usize,
@@ -725,6 +714,7 @@ struct RawTransactionIngestHydrationContext {
protocol: &'static str,
route: RawTransactionIngestSourceRoute,
route_prefix: &'static str,
source_key: [u8; 32],
source_key_domain: &'static [u8],
}
@@ -809,6 +799,7 @@ impl crate::RawTransactionIngestYellowstoneSource {
protocol: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL,
route: self.route.clone(),
route_prefix: "ys",
source_key: self.source_key,
source_key_domain: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN,
};
}
@@ -820,10 +811,10 @@ impl crate::RawTransactionIngestYellowstoneSource {
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
global_hydration_registry: std::sync::Arc<RawTransactionIngestGlobalHydrationRegistry>,
hydration_pending_limit: usize,
hydration_in_flight_limit: usize,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let opened = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -894,6 +885,7 @@ impl crate::RawTransactionIngestYellowstoneSource {
&admission_sender,
&mut stop_receiver,
&mut processing_frontier,
&continuity_contracts,
)
.await;
match handled {
@@ -1049,6 +1041,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
protocol: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_PROTOCOL,
route: self.route.clone(),
route_prefix: "hx",
source_key: self.source_key,
source_key_domain: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_SOURCE_KEY_DOMAIN,
};
}
@@ -1060,10 +1053,10 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
global_hydration_registry: std::sync::Arc<RawTransactionIngestGlobalHydrationRegistry>,
hydration_pending_limit: usize,
hydration_in_flight_limit: usize,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let connected = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -1101,7 +1094,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
return std::result::Result::Err(error);
}
let mut fault = std::option::Option::None;
let mut bounded_websocket_incident = false;
let mut bounded_websocket_incident = std::option::Option::None;
loop {
if *stop_receiver.borrow() {
break;
@@ -1110,11 +1103,14 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
fault = std::option::Option::Some(error);
break;
}
if bounded_websocket_incident && coordinator.pending_signal_count == 0 && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven"));
if let std::option::Option::Some((start_slot, end_slot)) = bounded_websocket_incident
&& coordinator.pending_signal_count == 0
&& coordinator.tasks.is_empty()
{
fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot));
break;
}
let can_receive = coordinator.can_receive() && !bounded_websocket_incident;
let can_receive = coordinator.can_receive() && bounded_websocket_incident.is_none();
if !can_receive && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
break;
@@ -1153,6 +1149,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
&admission_sender,
&mut stop_receiver,
&mut processing_frontier,
&continuity_contracts,
)
.await;
match handled {
@@ -1212,7 +1209,9 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
fault = std::option::Option::Some(error);
break;
}
bounded_websocket_incident = bounded_websocket_incident || incident_bounded;
if bounded_websocket_incident.is_none() {
bounded_websocket_incident = incident_bounded;
}
}
}
}
@@ -1333,6 +1332,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
) -> ksp_core_lib::Result<()> {
let context_config = ksp_onchain_transport_lib::SolanaContextConfig::new(std::option::Option::Some(self.commitment), std::option::Option::None);
let get_block_config = ksp_onchain_transport_lib::SolanaGetBlockConfig::new(
@@ -1390,6 +1390,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
std::option::Option::None => u64::MAX,
};
let window_end = current_tip.min(candidate_end);
let window_start = next_scan_slot;
let discovered = discover_http_block_window(
&self.http_pool,
&self.polling_role,
@@ -1469,6 +1470,17 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
}
}
if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot {
let coverage_result = {
let mut contracts = match continuity_contracts.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
contracts.record_coverage_epoch(self.source_key, window_start, proven_end_slot)
};
if let std::result::Result::Err(error) = coverage_result {
fault = std::option::Option::Some(error);
break;
}
next_scan_slot = match proven_end_slot.checked_add(1) {
std::option::Option::Some(value) => value,
std::option::Option::None => {
@@ -1680,8 +1692,8 @@ impl crate::RawTransactionIngestStandardBlockSource {
fault = std::option::Option::Some(error);
break;
}
if incident_bounded {
fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven"));
if let std::option::Option::Some((start_slot, end_slot)) = incident_bounded {
fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot));
break;
}
continue;
@@ -1706,8 +1718,8 @@ impl crate::RawTransactionIngestStandardBlockSource {
fault = std::option::Option::Some(error);
break;
}
if incident_bounded {
fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven"));
if let std::option::Option::Some((start_slot, end_slot)) = incident_bounded {
fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot));
break;
}
}
@@ -1834,6 +1846,7 @@ impl crate::RawTransactionIngestStandardLogsSource {
protocol: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL,
route: self.route.clone(),
route_prefix: "ws",
source_key: self.source_key,
source_key_domain: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN,
};
}
@@ -1845,10 +1858,10 @@ impl crate::RawTransactionIngestStandardLogsSource {
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
global_hydration_registry: std::sync::Arc<RawTransactionIngestGlobalHydrationRegistry>,
hydration_pending_limit: usize,
hydration_in_flight_limit: usize,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let connected = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -1886,7 +1899,7 @@ impl crate::RawTransactionIngestStandardLogsSource {
return std::result::Result::Err(error);
}
let mut fault = std::option::Option::None;
let mut bounded_websocket_incident = false;
let mut bounded_websocket_incident = std::option::Option::None;
loop {
if *stop_receiver.borrow() {
break;
@@ -1895,11 +1908,14 @@ impl crate::RawTransactionIngestStandardLogsSource {
fault = std::option::Option::Some(error);
break;
}
if bounded_websocket_incident && coordinator.pending_signal_count == 0 && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven"));
if let std::option::Option::Some((start_slot, end_slot)) = bounded_websocket_incident
&& coordinator.pending_signal_count == 0
&& coordinator.tasks.is_empty()
{
fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot));
break;
}
let can_receive = coordinator.can_receive() && !bounded_websocket_incident;
let can_receive = coordinator.can_receive() && bounded_websocket_incident.is_none();
if !can_receive && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
break;
@@ -1938,6 +1954,7 @@ impl crate::RawTransactionIngestStandardLogsSource {
&admission_sender,
&mut stop_receiver,
&mut processing_frontier,
&continuity_contracts,
)
.await;
match handled {
@@ -1990,7 +2007,9 @@ impl crate::RawTransactionIngestStandardLogsSource {
fault = std::option::Option::Some(error);
break;
}
bounded_websocket_incident = bounded_websocket_incident || incident_bounded;
if bounded_websocket_incident.is_none() {
bounded_websocket_incident = incident_bounded;
}
}
}
}
@@ -2233,15 +2252,15 @@ impl crate::RawTransactionIngestRuntimeResources {
if self.sources.is_empty() || self.sources.len() > crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_invalid"));
}
let mut repair_capabilities = std::vec::Vec::with_capacity(self.sources.len());
let mut continuity_capabilities = std::vec::Vec::with_capacity(self.sources.len());
for source in &self.sources {
let capability = match source.repair_capability_descriptor() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
repair_capabilities.push(capability);
continuity_capabilities.push(capability);
}
let continuity_contracts = match crate::RawTransactionIngestContinuityContracts::new(repair_capabilities) {
let continuity_contracts = match crate::RawTransactionIngestContinuityContracts::new(continuity_capabilities) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
@@ -2249,6 +2268,7 @@ impl crate::RawTransactionIngestRuntimeResources {
if let std::result::Result::Err(error) = continuity_contracts.validate_for_source_count(continuity_source_count) {
return std::result::Result::Err(error);
}
let continuity_contracts = std::sync::Arc::new(std::sync::Mutex::new(continuity_contracts));
let source_keys = self.sources.iter().map(RawTransactionIngestLiveSource::source_key).collect::<std::vec::Vec<_>>();
let hydration_source_count = self.sources.iter().filter(|source| return source.uses_hydration()).count();
if let std::result::Result::Err(error) =
@@ -2282,24 +2302,35 @@ impl crate::RawTransactionIngestRuntimeResources {
let source_admission_sender = admission_sender.clone();
let source_settings = settings.clone();
let source_stop_receiver = source_stop_receiver.clone();
let source_global_hydration_registry = std::sync::Arc::clone(&global_hydration_registry);
let source_shared = RawTransactionIngestSourceRuntimeShared {
continuity_contracts: std::sync::Arc::clone(&continuity_contracts),
global_hydration_registry: std::sync::Arc::clone(&global_hydration_registry),
hydration_in_flight_limit,
hydration_pending_limit,
};
let source_key = source.source_key();
let _abort_handle = children.spawn(async move {
return source
.run(
source_settings,
source_stop_receiver,
source_admission_sender,
source_global_hydration_registry,
hydration_pending_limit,
hydration_in_flight_limit,
publisher,
)
.await;
let result = source.run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared).await;
return (source_key, result);
});
}
std::mem::drop(admission_sender);
let result = supervise_live_source_tasks(stop_receiver, source_stop_sender, children).await;
if let std::result::Result::Err(error) = continuity_contracts.validate_for_source_count(continuity_source_count) {
let result = supervise_live_source_tasks(
stop_receiver,
source_stop_sender,
children,
std::sync::Arc::clone(&continuity_contracts),
std::sync::Arc::clone(&inventory),
)
.await;
let validation = {
let contracts = match continuity_contracts.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
contracts.validate_for_source_count(continuity_source_count)
};
if let std::result::Result::Err(error) = validation {
return std::result::Result::Err(error);
}
return result;
@@ -2384,7 +2415,9 @@ fn source_projection_with_state(
async fn supervise_live_source_tasks(
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
source_stop_sender: tokio::sync::watch::Sender<bool>,
mut children: tokio::task::JoinSet<ksp_core_lib::Result<()>>,
mut children: tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>,
continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
inventory: std::sync::Arc<std::sync::Mutex<RawTransactionIngestSourceInventory>>,
) -> ksp_core_lib::Result<()> {
loop {
if *stop_receiver.borrow() {
@@ -2402,21 +2435,131 @@ async fn supervise_live_source_tasks(
}
value = children.join_next(), if !children.is_empty() => value,
};
let first_fault = match joined {
std::option::Option::Some(std::result::Result::Ok(std::result::Result::Ok(()))) => {
std::option::Option::Some(crate::runtime_error("source.configured_source_closed"))
let (source_key, source_result) = match joined {
std::option::Option::Some(std::result::Result::Ok(value)) => value,
std::option::Option::Some(std::result::Result::Err(_)) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_join_failed"))).await;
},
std::option::Option::None => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_set_empty"))).await;
},
std::option::Option::Some(std::result::Result::Ok(std::result::Result::Err(error))) => std::option::Option::Some(error),
std::option::Option::Some(std::result::Result::Err(_)) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")),
std::option::Option::None => std::option::Option::Some(crate::runtime_error("source.task_set_empty")),
};
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, first_fault).await;
let source_fault = match source_result {
std::result::Result::Ok(()) => crate::runtime_error("source.configured_source_closed"),
std::result::Result::Err(error) => error,
};
if !source_loss_is_reconcilable(&source_fault) {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
}
let (active_source_keys, processing_frontier_slot) = {
let inventory = match inventory.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
match inventory.supervisor_state() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
},
}
};
let continuity_range = match source_loss_continuity_range(&source_fault) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
},
};
let continuity_range = match continuity_range {
std::option::Option::Some(value) => value,
std::option::Option::None => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
},
};
let decision = {
let mut contracts = match continuity_contracts.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
if let std::result::Result::Err(error) = contracts.record_source_loss_gap(source_key, continuity_range.0, continuity_range.1) {
std::result::Result::Err(error)
} else {
contracts.source_loss_decision(source_key, active_source_keys.as_slice(), processing_frontier_slot)
}
};
let decision = match decision {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
},
};
match decision {
crate::RawTransactionIngestSourceLossDecision::Continue => {
continue;
},
crate::RawTransactionIngestSourceLossDecision::Fault => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
},
}
}
}
fn continuity_range_error(condition: &'static str, start_slot: u64, end_slot: u64) -> ksp_core_lib::Error {
return crate::runtime_error(condition)
.with_context("continuity_start_slot", start_slot.to_string())
.with_context("continuity_end_slot", end_slot.to_string());
}
fn source_loss_continuity_range(error: &ksp_core_lib::Error) -> ksp_core_lib::Result<std::option::Option<(u64, u64)>> {
let mut start_slot = std::option::Option::None;
let mut end_slot = std::option::Option::None;
for context in error.context() {
if context.key() == "continuity_start_slot" {
start_slot = match context.value().parse::<u64>() {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")),
};
} else if context.key() == "continuity_end_slot" {
end_slot = match context.value().parse::<u64>() {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")),
};
}
}
return match (start_slot, end_slot) {
(std::option::Option::None, std::option::Option::None) => std::result::Result::Ok(std::option::Option::None),
(std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => std::result::Result::Ok(std::option::Option::Some((start, end))),
_ => std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")),
};
}
fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool {
if error.code() == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED {
return true;
}
if error.code() != crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID {
return false;
}
return error.context().iter().any(|context| {
if context.key() != "condition" {
return false;
}
return matches!(
context.value(),
"source.configured_source_closed" | "source.continuity_gap_proven" | "source.replay_coverage_unproven" | "source.websocket_incident_unbounded"
);
});
}
async fn drain_live_source_tasks(
children: &mut tokio::task::JoinSet<ksp_core_lib::Result<()>>,
children: &mut tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>,
mut first_fault: std::option::Option<ksp_core_lib::Error>,
) -> ksp_core_lib::Result<()> {
while let std::option::Option::Some(joined) = children.join_next().await {
@@ -2424,8 +2567,8 @@ async fn drain_live_source_tasks(
continue;
}
first_fault = match joined {
std::result::Result::Ok(std::result::Result::Ok(())) => std::option::Option::None,
std::result::Result::Ok(std::result::Result::Err(error)) => std::option::Option::Some(error),
std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None,
std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error),
std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")),
};
}
@@ -3930,19 +4073,20 @@ impl RawTransactionIngestProcessingFrontierReporter {
return std::result::Result::Ok(());
}
fn observe_websocket_post_incident_slot(&mut self, slot: u64) -> ksp_core_lib::Result<bool> {
fn observe_websocket_post_incident_slot(&mut self, slot: u64) -> ksp_core_lib::Result<std::option::Option<(u64, u64)>> {
let anchor = match self.websocket_incident_anchor.as_mut() {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Ok(false),
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
};
if anchor.end_slot().is_some() {
return std::result::Result::Ok(false);
return std::result::Result::Ok(std::option::Option::None);
}
if let std::result::Result::Err(error) = anchor.close_at(slot) {
return std::result::Result::Err(error);
}
let range = (anchor.start_slot(), slot);
self.publish();
return std::result::Result::Ok(true);
return std::result::Result::Ok(std::option::Option::Some(range));
}
fn observe_source_continuity(
@@ -4217,6 +4361,7 @@ impl RawTransactionIngestHydrationCoordinator {
admission_sender: &tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter,
continuity_contracts: &std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
) -> ksp_core_lib::Result<bool> {
let fetched = match joined {
std::result::Result::Ok(std::result::Result::Ok(value)) => value,
@@ -4251,6 +4396,16 @@ impl RawTransactionIngestHydrationCoordinator {
RawTransactionIngestKnownReferenceMissingDisposition::AwaitCoverage
| RawTransactionIngestKnownReferenceMissingDisposition::PreferBlockSlot => {},
}
let continuity_result = {
let mut contracts = match continuity_contracts.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
contracts.record_known_reference_gap(hydration.source_key, signal_slot)
};
if let std::result::Result::Err(error) = continuity_result {
return std::result::Result::Err(error);
}
if let std::result::Result::Err(error) = processing_frontier.settle_pending(signal_slot) {
return std::result::Result::Err(error);
}