api: fix context slot (#21)

Author: fanatid

.github/workflows/be-release.yml

@@ -34,9 +34,9 @@ jobs:
 
       - name: rename binaries
         run: |
-          mv target/release/solfees-be target/release/solfees-be-${{ matrix.is }}
+          mv target/release/solfees-be target/release/solfees-be-${{ matrix.os }}
           mv target/release/solfees-grpc2redis target/release/solfees-grpc2redis-${{ matrix.os }}
-          mv target/release/solfees-ws-client target/release/solfees-ws-client-${{ matrix.is }}
+          mv target/release/solfees-ws-client target/release/solfees-ws-client-${{ matrix.os }}
 
       - name: upload artifact
         if: startsWith(github.ref, 'refs/tags/') != true

CHANGELOG.md

@@ -14,6 +14,10 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Features
 
+### Breaking
+
+## [2.0.0] - 2024-11-18
+
 - api: add getLeaderSchedule ([#1](https://github.com/solana-stream-solutions/solfees/pull/1))
 - rpc: run multiple handlers ([#2](https://github.com/solana-stream-solutions/solfees/pull/2))
 - api: fix getLeaderSchedule for frontend ([#3](https://github.com/solana-stream-solutions/solfees/pull/3))
@@ -30,5 +34,4 @@ The minor version will be incremented upon a breaking change and the patch versi
 - backend: add metrics of used resources ([#18](https://github.com/solana-stream-solutions/solfees/pull/18))
 - backend: fix ws stream shutdown ([#19](https://github.com/solana-stream-solutions/solfees/pull/19))
 - api: add `skipZeros` to `gRPF` solfees ([#20](https://github.com/solana-stream-solutions/solfees/pull/20))
-
-### Breaking
+- api: fix context slot ([#21](https://github.com/solana-stream-solutions/solfees/pull/21))

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "solfees-be"
-version = "1.0.0"
+version = "2.0.0"
 authors = { workspace = true }
 edition = { workspace = true }
 description = "Backend of solfees.io"

solfees-be/Cargo.toml

@@ -45,7 +45,7 @@ use {
     },
 };
 
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum CommitmentLevel {
     #[default]

solfees-be/src/grpc_geyser.rs

@@ -33,7 +33,7 @@ use {
     },
     std::{
         borrow::Cow,
-        collections::{BTreeMap, HashMap},
+        collections::{BTreeMap, BTreeSet, HashMap},
         sync::{
             atomic::{AtomicBool, Ordering},
             Arc,
@@ -783,12 +783,19 @@ impl SolanaRpc {
                         hash,
                         time,
                         height,
-                        parent_slot: _parent_slot,
+                        parent_slot,
                         parent_hash: _parent_hash,
                         transactions,
                     } => {
-                        let info =
-                            StreamsSlotInfo::new(leader, slot, hash, time, height, transactions);
+                        let info = StreamsSlotInfo::new(
+                            leader,
+                            slot,
+                            parent_slot,
+                            hash,
+                            time,
+                            height,
+                            transactions,
+                        );
                         let _ = streams_tx.send(Arc::new(StreamsUpdateMessage::Slot { info }));
                     }
                 },
@@ -832,13 +839,13 @@ impl SolanaRpc {
                             hash,
                             time,
                             height,
-                            parent_slot: _parent_slot,
+                            parent_slot,
                             parent_hash: _parent_hash,
                             transactions,
                         } => {
-                            latest_blockhash_storage.push_block(slot, height, hash);
+                            latest_blockhash_storage.push_block(slot, parent_slot, height, hash);
 
-                            let info = StreamsSlotInfo::new(leader, slot, hash, time, height, transactions);
+                            let info = StreamsSlotInfo::new(leader, slot, parent_slot, hash, time, height, transactions);
                             slots_info.insert(slot, info.clone());
                             while slots_info.len() > MAX_NUM_RECENT_SLOT_INFO {
                                 slots_info.pop_first();
@@ -923,37 +930,17 @@ impl SolanaRpc {
                         SolanaRpc::internal_error_with_data("no slot"),
                     );
                 };
-                if rollback > 0 {
-                    let Some(first_available_slot) =
-                        latest_blockhash_storage.slots.keys().next().copied()
+                for _ in 0..rollback {
+                    let Some(parent_value) = latest_blockhash_storage.slots.get(&value.parent)
                     else {
                         return Self::create_failure(
                             jsonrpc,
                             id,
-                            JsonrpcError::invalid_params("empty slots storage"),
+                            JsonrpcError::invalid_params("not enought slots in the storage"),
                         );
                     };
-
-                    for _ in 0..rollback {
-                        loop {
-                            slot -= 1;
-
-                            if let Some(prev_value) = latest_blockhash_storage.slots.get(&slot) {
-                                if prev_value.height + 1 == value.height {
-                                    value = prev_value;
-                                    break;
-                                }
-                            } else if slot < first_available_slot {
-                                return Self::create_failure(
-                                    jsonrpc,
-                                    id,
-                                    JsonrpcError::invalid_params(
-                                        "not enought slots in the storage",
-                                    ),
-                                );
-                            }
-                        }
-                    }
+                    slot = value.parent;
+                    value = parent_value;
                 }
 
                 Self::create_success2(
@@ -1154,60 +1141,138 @@ enum RpcRequest {
 
 #[derive(Debug, Default)]
 struct LatestBlockhashStorage {
-    slots: BTreeMap<Slot, LatestBlockhashSlot>,
-    finalized_total: usize,
     slot_processed: Slot,
     slot_confirmed: Slot,
     slot_finalized: Slot,
+    confirmed: BTreeSet<Slot>,
+    finalized: BTreeSet<Slot>,
+    slots: BTreeMap<Slot, LatestBlockhashSlot>,
 }
 
 impl LatestBlockhashStorage {
-    fn push_block(&mut self, slot: Slot, height: Slot, hash: Hash) {
+    fn update_slots(&mut self) {
+        for (slot, entry) in self.slots.iter().rev() {
+            match entry.commitment {
+                CommitmentLevel::Processed => {
+                    self.slot_processed = self.slot_processed.max(*slot);
+                }
+                CommitmentLevel::Confirmed => {
+                    self.slot_confirmed = self.slot_confirmed.max(*slot);
+                }
+                CommitmentLevel::Finalized => {
+                    self.slot_finalized = self.slot_finalized.max(*slot);
+                    break;
+                }
+            }
+        }
+
+        // in case of epoch change with a lot of forks we can receive confirmed directly
+        // and processed would be behind, so we take `max(processed, confirmed)`
+        self.slot_processed = self.slot_processed.max(self.slot_confirmed);
+    }
+
+    fn push_block(&mut self, slot: Slot, parent: Slot, height: Slot, hash: Hash) {
+        // in case if slot status was received before block
+        let mut commitment = CommitmentLevel::Processed;
+        if self.confirmed.contains(&slot) {
+            commitment = CommitmentLevel::Confirmed;
+        }
+        if self.finalized.contains(&slot) {
+            commitment = CommitmentLevel::Finalized;
+        }
         self.slots.insert(
             slot,
             LatestBlockhashSlot {
-                hash,
+                commitment,
+                parent,
                 height,
-                commitment: CommitmentLevel::Processed,
+                hash,
             },
         );
+
+        // keep slots under the limit (based only on finalized count)
+        let slots_to_remove = self
+            .slots
+            .values()
+            .filter(|entry| entry.commitment == CommitmentLevel::Finalized)
+            .count()
+            .checked_sub(MAX_PROCESSING_AGE + 10)
+            .unwrap_or_default();
+        for _ in 0..slots_to_remove {
+            while let Some((_slot, value)) = self.slots.pop_first() {
+                if value.commitment == CommitmentLevel::Finalized {
+                    break;
+                }
+            }
+        }
+
+        // update tips
+        self.update_slots();
     }
 
     fn update_commitment(&mut self, slot: Slot, commitment: CommitmentLevel) {
-        if let Some(value) = self.slots.get_mut(&slot) {
-            value.commitment = commitment;
-
-            if commitment == CommitmentLevel::Processed && slot > self.slot_processed {
-                self.slot_processed = slot;
-            } else if commitment == CommitmentLevel::Confirmed {
-                self.slot_confirmed = slot;
-            } else if commitment == CommitmentLevel::Finalized {
-                self.finalized_total += 1;
-                self.slot_finalized = slot;
-            }
+        // save commitment
+        if commitment == CommitmentLevel::Confirmed {
+            self.confirmed.insert(slot);
+        } else if commitment == CommitmentLevel::Finalized {
+            self.confirmed.insert(slot);
+            self.finalized.insert(slot);
+        }
+        while self.confirmed.len() > MAX_PROCESSING_AGE {
+            self.confirmed.pop_first();
+        }
+        while self.finalized.len() > MAX_PROCESSING_AGE {
+            self.finalized.pop_first();
         }
 
-        while self.finalized_total > MAX_PROCESSING_AGE + 10 {
-            if let Some((_slot, value)) = self.slots.pop_first() {
-                if value.commitment == CommitmentLevel::Finalized {
-                    self.finalized_total -= 1;
+        // update current and all slots before
+        if let Some(value) = self.slots.get_mut(&slot) {
+            value.commitment = value.commitment.max(commitment);
+
+            let mut parent_slot = value.parent;
+            if commitment == CommitmentLevel::Confirmed {
+                loop {
+                    if let Some(value) = self.slots.get_mut(&parent_slot) {
+                        if value.commitment == CommitmentLevel::Processed {
+                            value.commitment = CommitmentLevel::Confirmed;
+                            parent_slot = value.parent;
+                            continue;
+                        }
+                    }
+                    break;
+                }
+            } else if commitment == CommitmentLevel::Finalized {
+                loop {
+                    if let Some(value) = self.slots.get_mut(&parent_slot) {
+                        if value.commitment != CommitmentLevel::Finalized {
+                            value.commitment = CommitmentLevel::Finalized;
+                            parent_slot = value.parent;
+                            continue;
+                        }
+                    }
+                    break;
                 }
             }
         }
+
+        // update tips
+        self.update_slots();
     }
 }
 
 #[derive(Debug)]
 struct LatestBlockhashSlot {
-    hash: Hash,
-    height: Slot,
     commitment: CommitmentLevel,
+    parent: Slot,
+    height: Slot,
+    hash: Hash,
 }
 
 #[derive(Debug, Clone)]
 struct StreamsSlotInfo {
     leader: Option<Pubkey>,
     slot: Slot,
+    parent_slot: Slot,
     commitment: CommitmentLevel,
     hash: Hash,
     time: UnixTimestamp,
@@ -1223,6 +1288,7 @@ impl StreamsSlotInfo {
     fn new(
         leader: Option<Pubkey>,
         slot: Slot,
+        parent_slot: Slot,
         hash: Hash,
         time: UnixTimestamp,
         height: Slot,
@@ -1242,6 +1308,7 @@ impl StreamsSlotInfo {
         Self {
             leader,
             slot,
+            parent_slot,
             commitment: CommitmentLevel::Processed,
             hash,
             time,
@@ -1292,6 +1359,7 @@ impl StreamsSlotInfo {
         SlotsSubscribeOutput::Slot {
             leader: self.leader.map(|pk| pk.to_string()).unwrap_or_default(),
             slot: self.slot,
+            parent_slot: self.parent_slot,
             commitment: self.commitment,
             hash: self.hash.to_string(),
             time: self.time,
@@ -1379,14 +1447,14 @@ enum StreamsUpdateMessage {
     },
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Default, Deserialize)]
+#[serde(default)]
 struct ReqParamsSlotsSubscribe {
-    #[serde(default)]
     config: Option<ReqParamsSlotsSubscribeConfig>,
 }
 
 #[derive(Debug, Default, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(default, rename_all = "camelCase")]
 struct ReqParamsSlotsSubscribeConfig {
     read_write: Vec<String>,
     read_only: Vec<String>,
@@ -1467,6 +1535,7 @@ enum SlotsSubscribeOutput {
     Slot {
         leader: String,
         slot: Slot,
+        parent_slot: Slot,
         commitment: CommitmentLevel,
         hash: String,
         time: UnixTimestamp,