geyser: reconstruct blocks with zero entries (#245)

CHANGELOG.md

@@ -16,6 +16,14 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Breaking
 
+## 2023-11-24
+
+- yellowstone-grpc-geyser-1.11.1+solana.1.17.6
+
+### Fixes
+
+- geyser: reconstruct blocks with zero entries ([#245](https://github.com/rpcpool/yellowstone-grpc/pull/245))
+
 ## 2023-11-21
 
 - yellowstone-grpc-client-1.12.0+solana.1.17.6

Cargo.toml

@@ -3,7 +3,7 @@ resolver = "2"
 members = [
     "examples/rust", # 1.11.0+solana.1.17.6
     "yellowstone-grpc-client", # 1.12.0+solana.1.17.6
-    "yellowstone-grpc-geyser", # 1.11.0+solana.1.17.6
+    "yellowstone-grpc-geyser", # 1.11.1+solana.1.17.6
     "yellowstone-grpc-proto", # 1.11.0+solana.1.17.6
     "yellowstone-grpc-tools", # 1.0.0-rc.7+solana.1.17.6
 ]
@@ -20,14 +20,14 @@ keywords = ["solana"]
 anyhow = "1.0.62"
 async-trait = "0.1.73"
 atty = "0.2.14"
-backoff = { version = "0.4.0", features = ["tokio"] }
+backoff = "0.4.0"
 base64 = "0.21.0"
 bincode = "1.3.3"
 bs58 = "0.4.0"
 bytes = "1.3.0"
 cargo-lock = "9.0.0"
 chrono = "0.4.26"
-clap = { version = "4.3.0", features = ["cargo", "derive"] }
+clap = "4.3.0"
 const-hex = "1.6.2"
 crossbeam-channel = "0.5.8"
 env_logger = "0.10.0"
@@ -37,16 +37,16 @@ google-cloud-googleapis = "0.11.0"
 google-cloud-pubsub = "0.21.0"
 hex = "0.4.3"
 http = "0.2.8"
-hyper = { version = "0.14.27", features = ["server"] }
+hyper = "0.14.27"
 json5 = "0.4.1"
 lazy_static = "1.4.0"
 log = "0.4.17"
 maplit = "1.0.2"
 prometheus = "0.13.2"
 prost = "0.12.1"
 protobuf-src = "1.1.0"
-rdkafka = { version = "0.34.0", features = ["sasl"] }
-serde = { version = "1.0.145", features = ["derive"] }
+rdkafka = "0.34.0"
+serde = "1.0.145"
 serde_json = "1.0.86"
 serde_yaml = "0.9.25"
 sha2 = "0.10.7"
@@ -55,16 +55,16 @@ solana-geyser-plugin-interface = "=1.17.6"
 solana-logger = "=1.17.6"
 solana-sdk = "=1.17.6"
 solana-transaction-status = "=1.17.6"
-spl-token-2022 = { version = "0.9.0", features = ["no-entrypoint"] }
+spl-token-2022 = "0.9.0"
 thiserror = "1.0"
-tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros", "time", "fs", "signal"] }
+tokio = "1.21.2"
 tokio-stream = "0.1.11"
-tonic = { version = "0.10.2", features = ["gzip", "tls", "tls-roots"] }
+tonic = "0.10.2"
 tonic-build = "0.10.2"
 tonic-health = "0.10.2"
 tracing = "0.1.37"
-tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-vergen = { version = "8.2.1", features = ["build", "rustc"] }
+tracing-subscriber = "0.3.17"
+vergen = "8.2.1"
 yellowstone-grpc-client = { path = "yellowstone-grpc-client", version = "=1.12.0+solana.1.17.6" }
 yellowstone-grpc-proto = { path = "yellowstone-grpc-proto", version = "=1.11.0+solana.1.17.6" }
 

README.md

@@ -6,6 +6,10 @@ It provides the ability to get slots, blocks, transactions, and account update n
 
 For additional documentation,  please see: https://docs.triton.one/rpc-pool/grpc-subscriptions
 
+#### Known bugs
+
+Block reconstruction inside gRPC plugin based on information provided by BlockMeta, unfortunately number of entries for blocks generated on validators always equal to zero. These blocks always will have zero entries. See issue on GitHub: https://github.com/solana-labs/solana/issues/33823
+
 ### Validator
 
 Current plugin version (`+solana.1.16.x`) use validator with backported `ReplicaBlockInfoV3` to Geyser interface — https://github.com/solana-labs/solana/pull/33359. As result it's not compatible with original validator from Solana Labs and would not work. You need to compile validator from the source code and can find patched releases in `Triton One` Solana fork: https://github.com/rpcpool/solana-public/tree/v1.16.16-geyser-block-v3.

examples/rust/Cargo.toml

@@ -14,11 +14,11 @@ name = "client"
 
 [dependencies]
 anyhow = { workspace = true }
-backoff = { workspace = true }
+backoff = { workspace = true, features = ["tokio"] }
 bincode = { workspace = true }
 bs58 = { workspace = true }
 chrono = { workspace = true }
-clap = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
 env_logger = { workspace = true }
 futures = { workspace = true }
 hex = { workspace = true }

yellowstone-grpc-client/Cargo.toml

@@ -15,9 +15,9 @@ bytes = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
 thiserror ={ workspace = true }
-tonic = { workspace = true }
+tonic = { workspace = true, features = ["tls", "tls-roots"] }
 tonic-health = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 
 [dev-dependencies]
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

yellowstone-grpc-geyser/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "yellowstone-grpc-geyser"
-version = "1.11.0+solana.1.17.6"
+version = "1.11.1+solana.1.17.6"
 authors = { workspace = true }
 edition = { workspace = true }
 description = "Yellowstone gRPC Geyser Plugin"
@@ -21,7 +21,7 @@ anyhow = { workspace = true }
 base64 = { workspace = true }
 bincode = { workspace = true }
 bs58 = { workspace = true }
-clap = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
 crossbeam-channel = { workspace = true }
 futures = { workspace = true }
 hyper = { workspace = true }
@@ -34,15 +34,15 @@ solana-geyser-plugin-interface = { workspace = true }
 solana-logger = { workspace = true }
 solana-sdk = { workspace = true }
 solana-transaction-status = { workspace = true }
-spl-token-2022 = { workspace = true }
-tokio = { workspace = true, features = ["fs"] }
+spl-token-2022 = { workspace = true, features = ["no-entrypoint"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs"] }
 tokio-stream = { workspace = true }
-tonic = { workspace = true }
+tonic = { workspace = true, features = ["gzip", "tls", "tls-roots"] }
 tonic-health = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 
 [build-dependencies]
 anyhow = { workspace = true }
 cargo-lock = { workspace = true }
 git-version = { workspace = true }
-vergen = { workspace = true }
+vergen = { workspace = true, features = ["build", "rustc"] }

yellowstone-grpc-geyser/src/grpc.rs

@@ -636,6 +636,7 @@ struct SlotMessages {
     accounts_dedup: HashMap<Pubkey, (u64, usize)>, // (write_version, message_index)
     entries: Vec<MessageEntry>,
     sealed: bool,
+    entries_count: usize,
     confirmed_at: Option<usize>,
     finalized_at: Option<usize>,
 }
@@ -644,11 +645,19 @@ impl SlotMessages {
     pub fn try_seal(&mut self) -> Option<Message> {
         if !self.sealed {
             if let Some(block_meta) = &self.block_meta {
-                if self.transactions.len() == block_meta.executed_transaction_count as usize
-                    && self.entries.len() == block_meta.entries_count as usize
+                let executed_transaction_count = block_meta.executed_transaction_count as usize;
+                let entries_count = block_meta.entries_count as usize;
+
+                // Additional check `entries_count == 0` due to bug of zero entries on block produced by validator
+                // See GitHub issue: https://github.com/solana-labs/solana/issues/33823
+                if self.transactions.len() == executed_transaction_count
+                    && (entries_count == 0 || self.entries.len() == entries_count)
                 {
                     let transactions = std::mem::take(&mut self.transactions);
-                    let entries = std::mem::take(&mut self.entries);
+                    let mut entries = std::mem::take(&mut self.entries);
+                    if entries_count == 0 {
+                        entries.clear();
+                    }
 
                     let mut accounts = Vec::with_capacity(self.messages.len());
                     for item in self.messages.iter().flatten() {
@@ -663,6 +672,7 @@ impl SlotMessages {
                     self.messages.push(Some(message.clone()));
 
                     self.sealed = true;
+                    self.entries_count = entries_count;
                     return Some(message);
                 }
             }
@@ -865,7 +875,7 @@ impl GrpcService {
                         slot_messages.messages.push(Some(message.clone()));
 
                         // If we already build Block message, new message will be a problem
-                        if slot_messages.sealed {
+                        if slot_messages.sealed && !(matches!(message, Message::Entry(_)) && slot_messages.entries_count == 0) {
                             prom::update_invalid_blocks(format!("unexpected message {}", message.kind()));
                             match block_fail_action {
                                 ConfigBlockFailAction::Log => {

yellowstone-grpc-tools/Cargo.toml

@@ -22,7 +22,7 @@ required-features = ["kafka"]
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 atty = { workspace = true }
-clap = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
 const-hex = { workspace = true, optional = true }
 futures = { workspace = true }
 google-cloud-googleapis = { workspace = true, optional = true }
@@ -35,26 +35,26 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 sha2 = { workspace = true, optional = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["signal"] }
 tokio-stream = { workspace = true }
-tonic = { workspace = true }
+tonic = { workspace = true, features = ["gzip"] }
 tonic-health = { workspace = true }
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
 yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 
 [target.'cfg(not(all(target_os = "macos", target_arch = "aarch64")))'.dependencies]
-rdkafka = { workspace = true, features = ["ssl"], optional = true }
+rdkafka = { workspace = true, features = ["sasl", "ssl"], optional = true }
 
 [target.'cfg(all(target_os = "macos", target_arch = "aarch64"))'.dependencies]
-rdkafka = { workspace = true, features = ["ssl-vendored"], optional = true }
+rdkafka = { workspace = true, features = ["sasl", "ssl-vendored"], optional = true }
 
 [build-dependencies]
 anyhow = { workspace = true }
 cargo-lock = { workspace = true }
 git-version = { workspace = true }
-vergen = { workspace = true }
+vergen = { workspace = true, features = ["build", "rustc"] }
 
 [features]
 default = ["google-pubsub", "kafka"]