447 lines
18 KiB
Rust
447 lines
18 KiB
Rust
// file: crates/ksp-onchain-transport-lib/src/ws_cluster.rs
|
|
// version: 5
|
|
|
|
/// Slot relationship reported by the standard Solana `slotNotification` WebSocket method.
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub struct SolanaSlotNotification {
|
|
slot: u64,
|
|
parent: u64,
|
|
root: u64,
|
|
}
|
|
|
|
impl SolanaSlotNotification {
|
|
/// Returns the newly processed slot.
|
|
#[must_use]
|
|
pub const fn slot(&self) -> u64 {
|
|
return self.slot;
|
|
}
|
|
|
|
/// Returns the parent slot reported by the validator.
|
|
#[must_use]
|
|
pub const fn parent(&self) -> u64 {
|
|
return self.parent;
|
|
}
|
|
|
|
/// Returns the current root slot reported alongside this slot update.
|
|
#[must_use]
|
|
pub const fn root(&self) -> u64 {
|
|
return self.root;
|
|
}
|
|
}
|
|
|
|
/// Execution statistics attached to unstable Solana `slotsUpdatesNotification` frozen updates.
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub struct SolanaSlotUpdateStats {
|
|
max_transactions_per_entry: u64,
|
|
num_failed_transactions: u64,
|
|
num_successful_transactions: u64,
|
|
num_transaction_entries: u64,
|
|
}
|
|
|
|
impl SolanaSlotUpdateStats {
|
|
/// Returns the maximum transactions per entry observed for the frozen bank.
|
|
#[must_use]
|
|
pub const fn max_transactions_per_entry(&self) -> u64 {
|
|
return self.max_transactions_per_entry;
|
|
}
|
|
|
|
/// Returns the failed transaction count.
|
|
#[must_use]
|
|
pub const fn num_failed_transactions(&self) -> u64 {
|
|
return self.num_failed_transactions;
|
|
}
|
|
|
|
/// Returns the successful transaction count.
|
|
#[must_use]
|
|
pub const fn num_successful_transactions(&self) -> u64 {
|
|
return self.num_successful_transactions;
|
|
}
|
|
|
|
/// Returns the transaction-entry count.
|
|
#[must_use]
|
|
pub const fn num_transaction_entries(&self) -> u64 {
|
|
return self.num_transaction_entries;
|
|
}
|
|
}
|
|
|
|
/// Typed unstable Solana slot-lifecycle update with an unknown-variant fallback.
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
pub enum SolanaSlotUpdate {
|
|
/// The first shred for a slot was received.
|
|
FirstShredReceived {
|
|
/// Slot whose first shred was observed.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
},
|
|
/// All shreds for a slot were received.
|
|
Completed {
|
|
/// Slot whose shred set completed.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
},
|
|
/// A bank was created for the slot.
|
|
CreatedBank {
|
|
/// Slot whose bank was created.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
/// Parent slot used to create the bank.
|
|
parent: u64,
|
|
},
|
|
/// A bank was frozen and execution statistics are available.
|
|
Frozen {
|
|
/// Slot whose bank was frozen.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
/// Execution statistics reported for the frozen bank.
|
|
stats: crate::SolanaSlotUpdateStats,
|
|
},
|
|
/// The slot was marked dead.
|
|
Dead {
|
|
/// Slot marked dead by the validator.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
/// Upstream diagnostic string explaining why the slot was marked dead.
|
|
error: std::string::String,
|
|
},
|
|
/// The slot reached the current unstable optimistic-confirmation marker.
|
|
OptimisticConfirmation {
|
|
/// Slot that reached optimistic confirmation.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
},
|
|
/// The slot became root.
|
|
Root {
|
|
/// Slot that became root.
|
|
slot: u64,
|
|
/// Millisecond Unix timestamp reported by the validator.
|
|
timestamp: i64,
|
|
},
|
|
/// A future upstream variant that KSP does not yet interpret.
|
|
///
|
|
/// `raw` is bounded by the physical session's configured inbound WebSocket message limit before JSON decoding.
|
|
Unknown {
|
|
/// Upstream `type` discriminator that KSP does not yet recognize.
|
|
update_type: std::string::String,
|
|
/// Complete bounded JSON object preserved for forward-compatible inspection.
|
|
raw: serde_json::Value,
|
|
},
|
|
}
|
|
|
|
impl SolanaSlotUpdate {
|
|
/// Returns the slot for known variants, or the optional slot found in an unknown raw variant.
|
|
#[must_use]
|
|
pub fn slot(&self) -> std::option::Option<u64> {
|
|
return match self {
|
|
Self::FirstShredReceived { slot, .. }
|
|
| Self::Completed { slot, .. }
|
|
| Self::CreatedBank { slot, .. }
|
|
| Self::Frozen { slot, .. }
|
|
| Self::Dead { slot, .. }
|
|
| Self::OptimisticConfirmation { slot, .. }
|
|
| Self::Root { slot, .. } => std::option::Option::Some(*slot),
|
|
Self::Unknown { raw, .. } => raw.get("slot").and_then(serde_json::Value::as_u64),
|
|
};
|
|
}
|
|
|
|
/// Returns the millisecond Unix timestamp for known variants, or an optional timestamp from an unknown raw variant.
|
|
#[must_use]
|
|
pub fn timestamp(&self) -> std::option::Option<i64> {
|
|
return match self {
|
|
Self::FirstShredReceived { timestamp, .. }
|
|
| Self::Completed { timestamp, .. }
|
|
| Self::CreatedBank { timestamp, .. }
|
|
| Self::Frozen { timestamp, .. }
|
|
| Self::Dead { timestamp, .. }
|
|
| Self::OptimisticConfirmation { timestamp, .. }
|
|
| Self::Root { timestamp, .. } => std::option::Option::Some(*timestamp),
|
|
Self::Unknown { raw, .. } => raw.get("timestamp").and_then(serde_json::Value::as_i64),
|
|
};
|
|
}
|
|
|
|
/// Returns the upstream `type` string, including unknown future values.
|
|
#[must_use]
|
|
pub fn update_type(&self) -> &str {
|
|
return match self {
|
|
Self::FirstShredReceived { .. } => "firstShredReceived",
|
|
Self::Completed { .. } => "completed",
|
|
Self::CreatedBank { .. } => "createdBank",
|
|
Self::Frozen { .. } => "frozen",
|
|
Self::Dead { .. } => "dead",
|
|
Self::OptimisticConfirmation { .. } => "optimisticConfirmation",
|
|
Self::Root { .. } => "root",
|
|
Self::Unknown { update_type, .. } => update_type.as_str(),
|
|
};
|
|
}
|
|
|
|
/// Returns the bounded raw object only for an unknown future upstream variant.
|
|
#[must_use]
|
|
pub const fn unknown_raw(&self) -> std::option::Option<&serde_json::Value> {
|
|
return match self {
|
|
Self::Unknown { raw, .. } => std::option::Option::Some(raw),
|
|
_ => std::option::Option::None,
|
|
};
|
|
}
|
|
}
|
|
|
|
/// Typed unstable gossip-vote notification delivered by standard Solana `voteSubscribe`.
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
pub struct SolanaVoteNotification {
|
|
vote_pubkey: ksp_core_lib::Pubkey,
|
|
slots: std::vec::Vec<u64>,
|
|
hash: std::string::String,
|
|
timestamp: std::option::Option<i64>,
|
|
signature: std::string::String,
|
|
}
|
|
|
|
impl SolanaVoteNotification {
|
|
/// Returns the vote-account public key.
|
|
#[must_use]
|
|
pub const fn vote_pubkey(&self) -> &ksp_core_lib::Pubkey {
|
|
return &self.vote_pubkey;
|
|
}
|
|
|
|
/// Returns the ordered slots covered by the observed vote.
|
|
#[must_use]
|
|
pub fn slots(&self) -> &[u64] {
|
|
return self.slots.as_slice();
|
|
}
|
|
|
|
/// Returns the vote hash exactly as reported by the unstable upstream wire.
|
|
#[must_use]
|
|
pub fn hash(&self) -> &str {
|
|
return self.hash.as_str();
|
|
}
|
|
|
|
/// Returns the optional vote timestamp, preserving omitted and explicit-null wire forms as `None`.
|
|
#[must_use]
|
|
pub const fn timestamp(&self) -> std::option::Option<i64> {
|
|
return self.timestamp;
|
|
}
|
|
|
|
/// Returns the vote transaction signature exactly as reported by the unstable upstream wire.
|
|
#[must_use]
|
|
pub fn signature(&self) -> &str {
|
|
return self.signature.as_str();
|
|
}
|
|
}
|
|
|
|
impl crate::WsSession {
|
|
/// Subscribes to standard Solana slot-processing notifications through `slotSubscribe`.
|
|
pub async fn slot_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotNotification>> {
|
|
return self
|
|
.subscribe_typed(crate::WsSubscriptionKind::Slot, std::vec::Vec::new(), |value| {
|
|
return decode_slot_notification("slotSubscribe", value);
|
|
})
|
|
.await;
|
|
}
|
|
|
|
/// Subscribes to standard Solana root-slot notifications through `rootSubscribe`.
|
|
pub async fn root_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<u64>> {
|
|
return self
|
|
.subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
|
|
return crate::decode_wire_json::<u64>("rootSubscribe", value);
|
|
})
|
|
.await;
|
|
}
|
|
|
|
/// Subscribes to unstable standard Solana slot-lifecycle notifications through `slotsUpdatesSubscribe`.
|
|
pub async fn slots_updates_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotUpdate>> {
|
|
return self
|
|
.subscribe_typed(crate::WsSubscriptionKind::SlotsUpdates, std::vec::Vec::new(), |value| {
|
|
return decode_slots_update_notification("slotsUpdatesSubscribe", value);
|
|
})
|
|
.await;
|
|
}
|
|
|
|
/// Subscribes to unstable pre-consensus gossip vote notifications through `voteSubscribe`.
|
|
pub async fn vote_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaVoteNotification>> {
|
|
return self
|
|
.subscribe_typed(crate::WsSubscriptionKind::Vote, std::vec::Vec::new(), |value| {
|
|
return decode_vote_notification("voteSubscribe", value);
|
|
})
|
|
.await;
|
|
}
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
struct WireSlotNotification {
|
|
slot: u64,
|
|
parent: u64,
|
|
root: u64,
|
|
}
|
|
|
|
fn decode_slot_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaSlotNotification> {
|
|
let decoded = crate::decode_wire_json::<WireSlotNotification>(method, value);
|
|
let wire = match decoded {
|
|
std::result::Result::Ok(wire) => wire,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(crate::SolanaSlotNotification { slot: wire.slot, parent: wire.parent, root: wire.root });
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct WireSlotUpdateStats {
|
|
max_transactions_per_entry: u64,
|
|
num_failed_transactions: u64,
|
|
num_successful_transactions: u64,
|
|
num_transaction_entries: u64,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
struct WireVoteNotification {
|
|
#[serde(rename = "votePubkey")]
|
|
vote_pubkey: std::string::String,
|
|
slots: std::vec::Vec<u64>,
|
|
hash: std::string::String,
|
|
#[serde(default)]
|
|
timestamp: std::option::Option<i64>,
|
|
signature: std::string::String,
|
|
}
|
|
|
|
fn decode_slots_update_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaSlotUpdate> {
|
|
let object = match value.as_object() {
|
|
std::option::Option::Some(object) => object,
|
|
std::option::Option::None => {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "slotsUpdatesSubscribe notification must be an object")
|
|
.with_context("rpc_method", method),
|
|
);
|
|
},
|
|
};
|
|
let update_type = match object.get("type").and_then(serde_json::Value::as_str) {
|
|
std::option::Option::Some(update_type) => update_type,
|
|
std::option::Option::None => {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "slotsUpdatesSubscribe notification is missing a string type")
|
|
.with_context("rpc_method", method),
|
|
);
|
|
},
|
|
};
|
|
if !matches!(update_type, "firstShredReceived" | "completed" | "createdBank" | "frozen" | "dead" | "optimisticConfirmation" | "root") {
|
|
return std::result::Result::Ok(crate::SolanaSlotUpdate::Unknown { update_type: update_type.to_owned(), raw: value.clone() });
|
|
}
|
|
let slot = match object.get("slot").and_then(serde_json::Value::as_u64) {
|
|
std::option::Option::Some(slot) => slot,
|
|
std::option::Option::None => return invalid_slots_update(method, "known slots update is missing numeric slot"),
|
|
};
|
|
let timestamp = match object.get("timestamp").and_then(serde_json::Value::as_i64) {
|
|
std::option::Option::Some(timestamp) => timestamp,
|
|
std::option::Option::None => return invalid_slots_update(method, "known slots update is missing numeric timestamp"),
|
|
};
|
|
return match update_type {
|
|
"firstShredReceived" => std::result::Result::Ok(crate::SolanaSlotUpdate::FirstShredReceived { slot, timestamp }),
|
|
"completed" => std::result::Result::Ok(crate::SolanaSlotUpdate::Completed { slot, timestamp }),
|
|
"createdBank" => match object.get("parent").and_then(serde_json::Value::as_u64) {
|
|
std::option::Option::Some(parent) => std::result::Result::Ok(crate::SolanaSlotUpdate::CreatedBank { slot, timestamp, parent }),
|
|
std::option::Option::None => invalid_slots_update(method, "createdBank update is missing numeric parent"),
|
|
},
|
|
"frozen" => {
|
|
let stats = match object.get("stats") {
|
|
std::option::Option::Some(stats) => crate::decode_wire_json::<WireSlotUpdateStats>(method, stats.clone()),
|
|
std::option::Option::None => return invalid_slots_update(method, "frozen update is missing stats"),
|
|
};
|
|
let stats = match stats {
|
|
std::result::Result::Ok(stats) => stats,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
std::result::Result::Ok(crate::SolanaSlotUpdate::Frozen {
|
|
slot,
|
|
timestamp,
|
|
stats: crate::SolanaSlotUpdateStats {
|
|
max_transactions_per_entry: stats.max_transactions_per_entry,
|
|
num_failed_transactions: stats.num_failed_transactions,
|
|
num_successful_transactions: stats.num_successful_transactions,
|
|
num_transaction_entries: stats.num_transaction_entries,
|
|
},
|
|
})
|
|
},
|
|
"dead" => match object.get("err").and_then(serde_json::Value::as_str) {
|
|
std::option::Option::Some(error) => std::result::Result::Ok(crate::SolanaSlotUpdate::Dead { slot, timestamp, error: error.to_owned() }),
|
|
std::option::Option::None => invalid_slots_update(method, "dead update is missing string err"),
|
|
},
|
|
"optimisticConfirmation" => std::result::Result::Ok(crate::SolanaSlotUpdate::OptimisticConfirmation { slot, timestamp }),
|
|
"root" => std::result::Result::Ok(crate::SolanaSlotUpdate::Root { slot, timestamp }),
|
|
_ => invalid_slots_update(method, "known slots update type dispatch failed"),
|
|
};
|
|
}
|
|
|
|
fn invalid_slots_update(method: &str, message: &'static str) -> ksp_core_lib::Result<crate::SolanaSlotUpdate> {
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, message).with_context("rpc_method", method));
|
|
}
|
|
|
|
fn decode_vote_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaVoteNotification> {
|
|
let decoded = crate::decode_wire_json::<WireVoteNotification>(method, value);
|
|
let wire = match decoded {
|
|
std::result::Result::Ok(wire) => wire,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let vote_pubkey = wire.vote_pubkey.parse::<ksp_core_lib::Pubkey>();
|
|
let vote_pubkey = match vote_pubkey {
|
|
std::result::Result::Ok(vote_pubkey) => vote_pubkey,
|
|
std::result::Result::Err(_) => {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "voteSubscribe notification contains an invalid votePubkey")
|
|
.with_context("rpc_method", method),
|
|
);
|
|
},
|
|
};
|
|
return std::result::Result::Ok(crate::SolanaVoteNotification {
|
|
vote_pubkey,
|
|
slots: wire.slots,
|
|
hash: wire.hash,
|
|
timestamp: wire.timestamp,
|
|
signature: wire.signature,
|
|
});
|
|
}
|
|
|
|
impl crate::SolanaStandardWsSession {
|
|
/// Subscribes to standard Solana slot-processing notifications through `slotSubscribe`.
|
|
pub async fn slot_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotNotification>> {
|
|
return self.physical_session().slot_subscribe().await;
|
|
}
|
|
|
|
/// Subscribes to standard Solana root-slot notifications through `rootSubscribe`.
|
|
pub async fn root_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<u64>> {
|
|
return self.physical_session().root_subscribe().await;
|
|
}
|
|
|
|
/// Subscribes to unstable standard Solana slot-lifecycle notifications through `slotsUpdatesSubscribe`.
|
|
pub async fn slots_updates_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotUpdate>> {
|
|
return self.physical_session().slots_updates_subscribe().await;
|
|
}
|
|
|
|
/// Subscribes to unstable pre-consensus gossip vote notifications through `voteSubscribe`.
|
|
pub async fn vote_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaVoteNotification>> {
|
|
return self.physical_session().vote_subscribe().await;
|
|
}
|
|
}
|
|
|
|
impl crate::HeliusLaserStreamWsSession {
|
|
/// Subscribes to slot-processing notifications through the standard `slotSubscribe` wire supported by Helius LaserStream WebSocket.
|
|
pub async fn slot_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotNotification>> {
|
|
return self.physical_session().slot_subscribe().await;
|
|
}
|
|
|
|
/// Subscribes to root-slot notifications through the standard `rootSubscribe` wire supported by Helius LaserStream WebSocket.
|
|
pub async fn root_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<u64>> {
|
|
return self.physical_session().root_subscribe().await;
|
|
}
|
|
|
|
/// Subscribes to unstable slot-lifecycle notifications through the standard `slotsUpdatesSubscribe` wire currently documented by Helius LaserStream
|
|
/// WebSocket.
|
|
pub async fn slots_updates_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotUpdate>> {
|
|
return self.physical_session().slots_updates_subscribe().await;
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/ws_cluster.rs"]
|
|
mod tests;
|