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

v0.3.14-pre.003

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-11 21:54:19 +02:00
parent 7bf938c19f
commit 6721a4142a
15 changed files with 872 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
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: 29
// version: 30
use sha2::Digest; // rust-rules: trait-import
@@ -1055,12 +1055,17 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
return std::result::Result::Err(source_transport_error(error.code()));
},
};
let mut session_snapshot_source = session.snapshot_source();
let hydration = self.hydration_context();
let mut coordinator =
RawTransactionIngestHydrationCoordinator::with_global_registry(global_hydration_registry, hydration_pending_limit, hydration_in_flight_limit);
let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active);
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(session_snapshot_source.current()) {
let _closed = session.close().await;
return std::result::Result::Err(error);
}
let mut fault = std::option::Option::None;
let mut bounded_websocket_incident = false;
loop {
if *stop_receiver.borrow() {
break;
@@ -1069,7 +1074,11 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
fault = std::option::Option::Some(error);
break;
}
let can_receive = coordinator.can_receive();
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"));
break;
}
let can_receive = coordinator.can_receive() && !bounded_websocket_incident;
if !can_receive && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
break;
@@ -1079,6 +1088,19 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
_ = stop_receiver.changed() => {
break;
}
source_snapshot = session_snapshot_source.wait_for_change() => {
let source_snapshot = match source_snapshot {
std::option::Option::Some(value) => value,
std::option::Option::None => {
fault = std::option::Option::Some(crate::runtime_error("source.websocket_snapshot_closed"));
break;
},
};
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(source_snapshot) {
fault = std::option::Option::Some(error);
break;
}
}
joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => {
let joined = match joined {
std::option::Option::Some(value) => value,
@@ -1143,10 +1165,18 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
break;
},
};
let incident_bounded = match processing_frontier.observe_websocket_post_incident_slot(signal.slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
fault = std::option::Option::Some(error);
break;
},
};
if let std::result::Result::Err(error) = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier) {
fault = std::option::Option::Some(error);
break;
}
bounded_websocket_incident = bounded_websocket_incident || incident_bounded;
}
}
}
@@ -1541,8 +1571,12 @@ impl crate::RawTransactionIngestStandardBlockSource {
return std::result::Result::Err(source_transport_error(error.code()));
},
};
let mut session_snapshot_source = session.snapshot_source();
let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active);
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(session_snapshot_source.current()) {
let _closed = session.close().await;
return std::result::Result::Err(error);
}
let mut fault = std::option::Option::None;
'source: loop {
if *stop_receiver.borrow() {
@@ -1553,6 +1587,20 @@ impl crate::RawTransactionIngestStandardBlockSource {
_ = stop_receiver.changed() => {
break;
}
source_snapshot = session_snapshot_source.wait_for_change() => {
let source_snapshot = match source_snapshot {
std::option::Option::Some(value) => value,
std::option::Option::None => {
fault = std::option::Option::Some(crate::runtime_error("source.websocket_snapshot_closed"));
break;
},
};
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(source_snapshot) {
fault = std::option::Option::Some(error);
break;
}
continue;
}
value = subscription.recv() => value,
};
let notification = match notification {
@@ -1584,11 +1632,22 @@ impl crate::RawTransactionIngestStandardBlockSource {
break;
},
};
let incident_bounded = match processing_frontier.observe_websocket_post_incident_slot(slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
fault = std::option::Option::Some(error);
break;
},
};
if ingresses.is_empty() {
if let std::result::Result::Err(error) = processing_frontier.observe_settled(slot) {
fault = std::option::Option::Some(error);
break;
}
if incident_bounded {
fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven"));
break;
}
continue;
}
for ingress in ingresses {
@@ -1611,6 +1670,10 @@ 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"));
break;
}
}
processing_frontier.discard_all_pending();
processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing);
@@ -1777,12 +1840,17 @@ impl crate::RawTransactionIngestStandardLogsSource {
return std::result::Result::Err(source_transport_error(error.code()));
},
};
let mut session_snapshot_source = session.snapshot_source();
let hydration = self.hydration_context();
let mut coordinator =
RawTransactionIngestHydrationCoordinator::with_global_registry(global_hydration_registry, hydration_pending_limit, hydration_in_flight_limit);
let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active);
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(session_snapshot_source.current()) {
let _closed = session.close().await;
return std::result::Result::Err(error);
}
let mut fault = std::option::Option::None;
let mut bounded_websocket_incident = false;
loop {
if *stop_receiver.borrow() {
break;
@@ -1791,7 +1859,11 @@ impl crate::RawTransactionIngestStandardLogsSource {
fault = std::option::Option::Some(error);
break;
}
let can_receive = coordinator.can_receive();
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"));
break;
}
let can_receive = coordinator.can_receive() && !bounded_websocket_incident;
if !can_receive && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
break;
@@ -1801,6 +1873,19 @@ impl crate::RawTransactionIngestStandardLogsSource {
_ = stop_receiver.changed() => {
break;
}
source_snapshot = session_snapshot_source.wait_for_change() => {
let source_snapshot = match source_snapshot {
std::option::Option::Some(value) => value,
std::option::Option::None => {
fault = std::option::Option::Some(crate::runtime_error("source.websocket_snapshot_closed"));
break;
},
};
if let std::result::Result::Err(error) = processing_frontier.observe_websocket_session_snapshot(source_snapshot) {
fault = std::option::Option::Some(error);
break;
}
}
joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => {
let joined = match joined {
std::option::Option::Some(value) => value,
@@ -1858,10 +1943,18 @@ impl crate::RawTransactionIngestStandardLogsSource {
break;
},
};
let incident_bounded = match processing_frontier.observe_websocket_post_incident_slot(signal.slot) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
fault = std::option::Option::Some(error);
break;
},
};
if let std::result::Result::Err(error) = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier) {
fault = std::option::Option::Some(error);
break;
}
bounded_websocket_incident = bounded_websocket_incident || incident_bounded;
}
}
}
@@ -3364,6 +3457,18 @@ fn map_yellowstone_source_state(state: ksp_onchain_transport_lib::YellowstoneGrp
};
}
fn map_websocket_source_state(state: ksp_onchain_transport_lib::WsSessionState) -> crate::RawTransactionIngestSourceState {
return match state {
ksp_onchain_transport_lib::WsSessionState::Disconnected
| ksp_onchain_transport_lib::WsSessionState::Connecting
| ksp_onchain_transport_lib::WsSessionState::Reconnecting { .. } => crate::RawTransactionIngestSourceState::Reconnecting,
ksp_onchain_transport_lib::WsSessionState::Active => crate::RawTransactionIngestSourceState::Active,
ksp_onchain_transport_lib::WsSessionState::Closing => crate::RawTransactionIngestSourceState::Closing,
ksp_onchain_transport_lib::WsSessionState::Closed => crate::RawTransactionIngestSourceState::Closed,
ksp_onchain_transport_lib::WsSessionState::Failed => crate::RawTransactionIngestSourceState::Failed,
};
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct RawTransactionIngestProcessingSlotState {
pending: usize,
@@ -3447,6 +3552,10 @@ impl RawTransactionIngestProcessingFrontier {
return;
}
fn highest_observed_slot(&self) -> std::option::Option<u64> {
return self.slots.keys().next_back().copied();
}
fn projection(&self) -> crate::RawTransactionIngestProcessingFrontierProjection {
let oldest_pending_slot = self.slots.iter().find_map(|(slot, state)| {
if state.pending == 0 {
@@ -3527,6 +3636,8 @@ struct RawTransactionIngestProcessingFrontierReporter {
source_reconnect_total: u64,
source_replay_attempt_total: u64,
source_continuity_gap_total: u64,
source_overflow_total: u64,
websocket_incident_anchor: std::option::Option<crate::RawTransactionIngestWebSocketIncidentAnchor>,
}
impl RawTransactionIngestProcessingFrontierReporter {
@@ -3538,6 +3649,8 @@ impl RawTransactionIngestProcessingFrontierReporter {
source_reconnect_total: 0,
source_replay_attempt_total: 0,
source_continuity_gap_total: 0,
source_overflow_total: 0,
websocket_incident_anchor: std::option::Option::None,
};
}
@@ -3580,6 +3693,85 @@ impl RawTransactionIngestProcessingFrontierReporter {
);
}
fn observe_websocket_session_snapshot(&mut self, snapshot: ksp_onchain_transport_lib::WsSessionSnapshot) -> ksp_core_lib::Result<()> {
return self.observe_websocket_continuity(map_websocket_source_state(snapshot.state()), snapshot.continuity_gap_count(), snapshot.overflow_count());
}
fn observe_websocket_continuity(
&mut self,
state: crate::RawTransactionIngestSourceState,
reconnect_total: u64,
overflow_total: u64,
) -> ksp_core_lib::Result<()> {
if reconnect_total < self.source_reconnect_total || overflow_total < self.source_overflow_total {
return std::result::Result::Err(crate::runtime_error("source.websocket_continuity_counter_regression"));
}
let continuity_gap_total = match reconnect_total.checked_add(overflow_total) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::counter_exhausted_error("source.websocket_continuity_gap_total")),
};
if continuity_gap_total < self.source_continuity_gap_total {
return std::result::Result::Err(crate::runtime_error("source.continuity_counter_regression"));
}
let saw_reconnect = reconnect_total > self.source_reconnect_total;
let saw_overflow = overflow_total > self.source_overflow_total;
self.source_reconnect_total = reconnect_total;
self.source_replay_attempt_total = 0;
self.source_continuity_gap_total = continuity_gap_total;
self.source_overflow_total = overflow_total;
if saw_reconnect || saw_overflow {
let start_slot = match self.frontier.highest_observed_slot() {
std::option::Option::Some(value) => value,
std::option::Option::None => {
self.publish();
return std::result::Result::Err(crate::runtime_error("source.websocket_incident_unbounded"));
},
};
match self.websocket_incident_anchor.as_mut() {
std::option::Option::Some(anchor) => {
if let std::result::Result::Err(error) = anchor.extend(reconnect_total, overflow_total, saw_reconnect, saw_overflow) {
self.publish();
return std::result::Result::Err(error);
}
},
std::option::Option::None => {
let anchor =
match crate::RawTransactionIngestWebSocketIncidentAnchor::new(start_slot, reconnect_total, overflow_total, saw_reconnect, saw_overflow)
{
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
self.publish();
return std::result::Result::Err(error);
},
};
self.websocket_incident_anchor = std::option::Option::Some(anchor);
},
}
}
self.source_state = if self.websocket_incident_anchor.is_some() && state == crate::RawTransactionIngestSourceState::Active {
std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting)
} else {
std::option::Option::Some(state)
};
self.publish();
return std::result::Result::Ok(());
}
fn observe_websocket_post_incident_slot(&mut self, slot: u64) -> ksp_core_lib::Result<bool> {
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),
};
if anchor.end_slot().is_some() {
return std::result::Result::Ok(false);
}
if let std::result::Result::Err(error) = anchor.close_at(slot) {
return std::result::Result::Err(error);
}
self.publish();
return std::result::Result::Ok(true);
}
fn observe_source_continuity(
&mut self,
state: crate::RawTransactionIngestSourceState,