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

v0.3.14-pre.004

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

61
crates/ksp-onchain-transport-lib/src/grpc_stream.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs
// version: 4
// version: 5
use tonic_prost::prost::Message; // rust-rules: trait-import
@@ -34,6 +34,8 @@ pub struct YellowstoneGrpcSubscribeSnapshot {
continuity_gap_count: u64,
duplicate_update_count: u64,
replay_attempt_count: u64,
replay_delivery_count: u64,
replay_coverage_unproven_count: u64,
last_requested_from_slot: std::option::Option<u64>,
last_observed_slot: std::option::Option<u64>,
terminal_error_code: std::option::Option<ksp_core_lib::ErrorCode>,
@@ -84,6 +86,8 @@ impl YellowstoneGrpcSubscribeSnapshot {
continuity_gap_count: 0,
duplicate_update_count: 0,
replay_attempt_count: 0,
replay_delivery_count: 0,
replay_coverage_unproven_count: 0,
last_requested_from_slot: initial_from_slot,
last_observed_slot: std::option::Option::None,
terminal_error_code: std::option::Option::None,
@@ -122,6 +126,25 @@ impl YellowstoneGrpcSubscribeSnapshot {
return self.replay_attempt_count;
}
/// Returns the number of replay-bearing reconnects that delivered the requested replay boundary slot again.
///
/// This is conservative delivery evidence only. It does not prove that every matching update in the replay interval was delivered and must never be
/// interpreted as `replay_covered`.
#[must_use]
pub const fn replay_delivery_count(self) -> u64 {
return self.replay_delivery_count;
}
/// Returns the number of successful replay-bearing reconnects whose target coverage remains unproven.
///
/// The counter advances when the first post-reconnect slot-bearing update reaches or passes the requested replay boundary, because generic Transport
/// cannot prove from that delivery alone that every matching update in the replay interval was delivered. It also advances when another reconnect starts
/// before any slot-bearing replay material arrives. This is an explicit lack of coverage proof, not proof that a filtered event actually existed or was lost.
#[must_use]
pub const fn replay_coverage_unproven_count(self) -> u64 {
return self.replay_coverage_unproven_count;
}
/// Returns the most recent effective `from_slot` sent by KSP, including any clamp to `SubscribeReplayInfo.first_available`.
#[must_use]
pub const fn last_requested_from_slot(self) -> std::option::Option<u64> {
@@ -506,17 +529,44 @@ enum UpdateIdentity {
}
struct ContinuityTracker {
pending_replay_from_slot: std::option::Option<u64>,
recent_order: std::collections::VecDeque<UpdateIdentity>,
recent_set: std::collections::HashSet<UpdateIdentity>,
}
impl ContinuityTracker {
fn new() -> Self {
return Self { recent_order: std::collections::VecDeque::new(), recent_set: std::collections::HashSet::new() };
return Self {
pending_replay_from_slot: std::option::Option::None,
recent_order: std::collections::VecDeque::new(),
recent_set: std::collections::HashSet::new(),
};
}
fn begin_replay(&mut self, from_slot: std::option::Option<u64>) {
self.pending_replay_from_slot = from_slot;
return;
}
fn abandon_pending_replay(&mut self, snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot) -> bool {
if self.pending_replay_from_slot.take().is_none() {
return false;
}
snapshot.replay_coverage_unproven_count = snapshot.replay_coverage_unproven_count.saturating_add(1);
return true;
}
fn observe(&mut self, update: &crate::YellowstoneSubscribeUpdate, snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot) {
if let std::option::Option::Some(slot) = update_slot(update) {
if let std::option::Option::Some(requested) = self.pending_replay_from_slot
&& slot >= requested
{
if slot == requested {
snapshot.replay_delivery_count = snapshot.replay_delivery_count.saturating_add(1);
}
snapshot.replay_coverage_unproven_count = snapshot.replay_coverage_unproven_count.saturating_add(1);
self.pending_replay_from_slot = std::option::Option::None;
}
snapshot.last_observed_slot = std::option::Option::Some(match snapshot.last_observed_slot {
std::option::Option::Some(previous) => std::cmp::max(previous, slot),
std::option::Option::None => slot,
@@ -620,6 +670,7 @@ async fn run_subscribe_actor(
&mut shutdown_rx,
&mut snapshot,
&snapshot_tx,
&mut tracker,
)
.await
{
@@ -662,6 +713,7 @@ async fn run_subscribe_actor(
&mut shutdown_rx,
&mut snapshot,
&snapshot_tx,
&mut tracker,
)
.await
{
@@ -701,8 +753,12 @@ async fn reconnect_subscribe_stream(
shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot,
snapshot_tx: &tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
tracker: &mut ContinuityTracker,
) -> ReconnectOutcome {
clear_request_sender(request_state);
if tracker.abandon_pending_replay(snapshot) {
snapshot_tx.send_replace(*snapshot);
}
snapshot.state = crate::YellowstoneGrpcSubscribeState::Reconnecting;
snapshot.terminal_error_code = std::option::Option::None;
snapshot_tx.send_replace(*snapshot);
@@ -785,6 +841,7 @@ async fn reconnect_subscribe_stream(
return ReconnectOutcome::Exhausted(error);
}
snapshot.reconnect_count = snapshot.reconnect_count.saturating_add(1);
tracker.begin_replay(effective_from_slot);
snapshot.state = crate::YellowstoneGrpcSubscribeState::Active;
snapshot.terminal_error_code = std::option::Option::None;
snapshot_tx.send_replace(*snapshot);