grpc multiplexing integrate

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

@@ -2,7 +2,9 @@ name: Deploy to Fly Staging
 
 on:
   push:
-      branches: [main]
+      # TODO remove feature/grpc-multiplexing-integrate when it's merged
+      # restore "main"
+      branches: [feature/grpc-multiplexing-integrate]
 
 env:
   FLY_API_TOKEN: ${{ secrets.FLY_API_TOKEN }}

cd/lite-rpc-staging.toml

@@ -13,8 +13,10 @@ primary_region = "nrt"
 [env]
   PORT_HTTP = "8890"
   PORT_WS = "8891"
-  RUST_LOG = "info"
+  # remove after grpc multipex got merged
+  RUST_LOG = "info,solana_lite_rpc_cluster_endpoints=debug,geyser_grpc_connector=debug"
 
  [metrics]
    path = "/metrics"
    port = 9091
+

cluster-endpoints/Cargo.toml

@@ -8,6 +8,10 @@ repository = "https://github.com/blockworks-foundation/lite-rpc"
 license = "AGPL"
 
 [dependencies]
+#geyser-grpc-connector = { path = "../../geyser-grpc-connector" }
+#geyser-grpc-connector = { tag = "v0.5.0+yellowstone.1.11+solana.1.16.17", git = "https://github.com/blockworks-foundation/geyser-grpc-connector.git" }
+geyser-grpc-connector = "0.5.0+yellowstone.1.11"
+
 solana-sdk = { workspace = true }
 solana-rpc-client-api = { workspace = true }
 solana-transaction-status = { workspace = true }
@@ -24,6 +28,7 @@ bs58 = { workspace = true }
 base64 = { workspace = true }
 thiserror = { workspace = true }
 futures = { workspace = true }
+merge-streams = "0.1.2"
 bytes = { workspace = true }
 anyhow = { workspace = true }
 log = { workspace = true }

cluster-endpoints/src/grpc_inspect.rs

@@ -0,0 +1,66 @@
+use log::{debug, warn};
+use solana_lite_rpc_core::types::BlockStream;
+use solana_sdk::commitment_config::CommitmentConfig;
+use tokio::sync::broadcast::error::RecvError;
+use tokio::task::JoinHandle;
+
+pub fn block_debug_listen(
+    mut block_notifier: BlockStream,
+    commitment_config: CommitmentConfig,
+) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut last_highest_slot_number = 0;
+
+        loop {
+            match block_notifier.recv().await {
+                Ok(block) => {
+                    if block.commitment_config != commitment_config {
+                        continue;
+                    }
+
+                    debug!(
+                        "Saw block: {} @ {} with {} txs",
+                        block.slot,
+                        block.commitment_config.commitment,
+                        block.transactions.len()
+                    );
+
+                    if last_highest_slot_number != 0 {
+                        if block.parent_slot == last_highest_slot_number {
+                            debug!(
+                                "parent slot is correct ({} -> {})",
+                                block.slot, block.parent_slot
+                            );
+                        } else {
+                            warn!(
+                                "parent slot not correct ({} -> {})",
+                                block.slot, block.parent_slot
+                            );
+                        }
+                    }
+
+                    if block.slot > last_highest_slot_number {
+                        last_highest_slot_number = block.slot;
+                    } else {
+                        // note: ATM this fails very often (using the RPC poller)
+                        warn!(
+                            "Monotonic check failed - block {} is out of order, last highest was {}",
+                            block.slot, last_highest_slot_number
+                        );
+                    }
+                } // -- Ok
+                Err(RecvError::Lagged(missed_blocks)) => {
+                    warn!(
+                        "Could not keep up with producer - missed {} blocks",
+                        missed_blocks
+                    );
+                }
+                Err(other_err) => {
+                    panic!("Error receiving block: {:?}", other_err);
+                }
+            }
+
+            // ...
+        }
+    })
+}