Feature/block priofees

.github/workflows/fly-deploy-staging.yml

@@ -3,6 +3,7 @@ name: Deploy to Fly Staging
 on:
   push:
       branches: [main]
+      tags: ['experimental/*']
 
 env:
   FLY_API_TOKEN: ${{ secrets.FLY_API_TOKEN }}

Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "quic-forward-proxy-integration-test",
     "cluster-endpoints",
     "history",
+    "block_priofees",
     "bench"
 ]
 
@@ -69,6 +70,7 @@ solana-lite-rpc-core = {path = "core", version="0.2.3"}
 solana-lite-rpc-cluster-endpoints = {path = "cluster-endpoints", version="0.2.3"}
 solana-lite-rpc-history = {path = "history", version="0.2.3"}
 solana-lite-rpc-stakevote = {path = "stake_vote", version="0.2.3"}
+solana-lite-rpc-block-priofees = {path = "block_priofees", version="0.2.3"}
 
 async-trait = "0.1.68"
 yellowstone-grpc-client = { git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v1.12.0+solana.1.17.15" }

block_priofees/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "solana-lite-rpc-block-priofees"
+version = "0.2.3"
+edition = "2021"
+description = "Expose priority fees stats per block via RPC and WebSocket"
+
+[dependencies]
+solana-lite-rpc-core = {workspace = true}
+
+solana-sdk = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+log = { workspace = true }
+itertools = { workspace = true }
+dashmap = { workspace = true }
+jsonrpsee = { workspace = true }
+tracing-subscriber = { workspace = true }
+prometheus = { workspace = true }
+lazy_static = { workspace = true }
+tokio = { version = "1.28.2", features = ["full"]}

block_priofees/src/block_priofees.rs

@@ -0,0 +1,157 @@
+use crate::rpc_data::{PrioFeesStats, PrioFeesUpdateMessage, TxAggregateStats};
+use crate::stats_calculation::calculate_supp_percentiles;
+use log::{error, info, trace, warn};
+use solana_lite_rpc_core::types::BlockStream;
+use solana_sdk::clock::Slot;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use tokio::sync::broadcast::error::RecvError::{Closed, Lagged};
+use tokio::sync::broadcast::Sender;
+use tokio::sync::RwLock;
+use tokio::task::JoinHandle;
+
+// note: ATM only the latest slot (highest key) is used
+const SLOTS_TO_RETAIN: u64 = 100;
+
+/// put everything required to serve sync data calls here
+#[derive(Clone)]
+pub struct PrioFeeStore {
+    // store priofees stats for recently processed blocks up to CLEANUP_SLOTS_AFTER
+    recent: Arc<RwLock<BTreeMap<Slot, PrioFeesStats>>>,
+}
+
+pub struct PrioFeesService {
+    pub block_fees_store: PrioFeeStore,
+    // use .subscribe() to get a receiver
+    pub block_fees_stream: Sender<PrioFeesUpdateMessage>,
+}
+
+impl PrioFeesService {
+    pub async fn get_latest_priofees(&self) -> Option<(Slot, PrioFeesStats)> {
+        let lock = self.block_fees_store.recent.read().await;
+        let latest_in_store = lock.last_key_value();
+        latest_in_store.map(|x| (*x.0, x.1.clone()))
+    }
+}
+
+pub async fn start_block_priofees_task(
+    mut block_stream: BlockStream,
+) -> (JoinHandle<()>, PrioFeesService) {
+    let recent_data = Arc::new(RwLock::new(BTreeMap::new()));
+    let store = PrioFeeStore {
+        recent: recent_data.clone(),
+    };
+    let (priofees_update_sender, _priofees_update_receiver) = tokio::sync::broadcast::channel(64);
+    let sender_to_return = priofees_update_sender.clone();
+
+    let jh_priofees_task = tokio::spawn(async move {
+        let sender = priofees_update_sender.clone();
+        'recv_loop: loop {
+            let block = block_stream.recv().await;
+            match block {
+                Ok(block) => {
+                    if !block.commitment_config.is_processed() {
+                        continue;
+                    }
+                    let processed_slot = block.slot;
+                    {
+                        // first do some cleanup
+                        let mut lock = recent_data.write().await;
+                        lock.retain(|slot, _| {
+                            *slot > processed_slot.saturating_sub(SLOTS_TO_RETAIN)
+                        });
+                    }
+
+                    let block_priofees = block
+                        .transactions
+                        .iter()
+                        .filter(|tx| !tx.is_vote)
+                        .map(|tx| {
+                            (
+                                tx.prioritization_fees.unwrap_or_default(),
+                                tx.cu_consumed.unwrap_or_default(),
+                            )
+                        })
+                        .collect::<Vec<(u64, u64)>>();
+
+                    let priofees_percentiles = calculate_supp_percentiles(&block_priofees);
+
+                    let total_tx_count = block.transactions.len() as u64;
+
+                    let nonvote_tx_count =
+                        block.transactions.iter().filter(|tx| !tx.is_vote).count() as u64;
+
+                    let total_cu_consumed = block
+                        .transactions
+                        .iter()
+                        .map(|tx| tx.cu_consumed.unwrap_or(0))
+                        .sum::<u64>();
+
+                    let nonvote_cu_consumed = block
+                        .transactions
+                        .iter()
+                        .filter(|tx| !tx.is_vote)
+                        .map(|tx| tx.cu_consumed.unwrap_or(0))
+                        .sum::<u64>();
+
+                    trace!("Got prio fees stats for processed block {}", processed_slot);
+
+                    let priofees_stats = PrioFeesStats {
+                        by_tx: priofees_percentiles.by_tx,
+                        by_tx_percentiles: priofees_percentiles.by_tx_percentiles,
+                        by_cu: priofees_percentiles.by_cu,
+                        by_cu_percentiles: priofees_percentiles.by_cu_percentiles,
+                        tx_count: TxAggregateStats {
+                            total: total_tx_count,
+                            nonvote: nonvote_tx_count,
+                        },
+                        cu_consumed: TxAggregateStats {
+                            total: total_cu_consumed,
+                            nonvote: nonvote_cu_consumed,
+                        },
+                    };
+
+                    {
+                        // first do some cleanup
+                        let mut lock = recent_data.write().await;
+                        lock.insert(processed_slot, priofees_stats.clone());
+                    }
+                    let msg = PrioFeesUpdateMessage {
+                        slot: processed_slot,
+                        priofees_stats,
+                    };
+                    let send_result = sender.send(msg);
+                    match send_result {
+                        Ok(n_subscribers) => {
+                            trace!(
+                                "sent priofees update message to {} subscribers (buffer={})",
+                                n_subscribers,
+                                sender.len()
+                            );
+                        }
+                        Err(_) => {
+                            trace!("no subscribers for priofees update message");
+                        }
+                    }
+                }
+                Err(Lagged(_lagged)) => {
+                    warn!("channel error receiving block for priofees calculation - continue");
+                    continue 'recv_loop;
+                }
+                Err(Closed) => {
+                    error!("failed to receive block, sender closed - aborting");
+                    break 'recv_loop;
+                }
+            }
+        }
+        info!("priofees task shutting down");
+    });
+
+    (
+        jh_priofees_task,
+        PrioFeesService {
+            block_fees_store: store,
+            block_fees_stream: sender_to_return,
+        },
+    )
+}

block_priofees/src/lib.rs

@@ -0,0 +1,6 @@
+mod block_priofees;
+
+pub mod rpc_data;
+mod stats_calculation;
+
+pub use block_priofees::{start_block_priofees_task, PrioFeesService};

block_priofees/src/rpc_data.rs

@@ -0,0 +1,39 @@
+use jsonrpsee::core::Serialize;
+use solana_sdk::clock::Slot;
+use std::fmt::Display;
+
+#[derive(Clone, Serialize, Debug)]
+pub struct TxAggregateStats {
+    pub total: u64,
+    pub nonvote: u64,
+}
+
+#[derive(Clone, Serialize, Debug)]
+pub struct PrioFeesStats {
+    pub by_tx: Vec<u64>,
+    pub by_tx_percentiles: Vec<f32>,
+    pub by_cu: Vec<u64>,
+    pub by_cu_percentiles: Vec<f32>,
+    pub tx_count: TxAggregateStats,
+    pub cu_consumed: TxAggregateStats,
+}
+
+#[derive(Clone, Serialize, Debug, Eq, PartialEq, Hash)]
+pub struct FeePoint {
+    // percentile
+    pub p: u32,
+    // value of fees in lamports
+    pub v: u64,
+}
+
+impl Display for FeePoint {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "(p{}, {})", self.p, self.v)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct PrioFeesUpdateMessage {
+    pub slot: Slot,
+    pub priofees_stats: PrioFeesStats,
+}