Merge branch 'production'

blockworks-foundation · Mar 27, 2024 · cc02f11 · cc02f11
2 parents bd9d15a + 26222f9
commit cc02f11
Show file tree

Hide file tree

Showing 8 changed files with 167 additions and 162 deletions.
diff --git a/.github/workflows/clippy_test.yml b/.github/workflows/clippy_test.yml
@@ -28,7 +28,7 @@ jobs:
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
           # use toolchain version from rust-toolchain.toml
-          toolchain: nightly-2024-01-05
+          toolchain: nightly-2023-10-05
           components: rustfmt, clippy
           cache: true
           # avoid the default "-D warnings" which thrashes cache
@@ -48,5 +48,5 @@ jobs:
 
       - name: Run fmt+clippy
         run: |
-          cargo +nightly-2024-01-05 fmt --all --check
-          cargo +nightly-2024-01-05 clippy --locked --workspace --all-targets -- -D warnings
+          cargo +nightly-2023-10-05 fmt --all --check
+          cargo +nightly-2023-10-05 clippy --locked --workspace --all-targets -- -D warnings
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/cluster-endpoints/src/grpc_subscription.rs b/cluster-endpoints/src/grpc_subscription.rs
@@ -270,12 +270,12 @@ fn map_compute_budget_instructions(message: &VersionedMessage) -> (Option<u32>,
     (cu_requested, prioritization_fees)
 }
 
-// not called
 pub fn create_block_processing_task(
     grpc_addr: String,
     grpc_x_token: Option<String>,
     block_sx: async_channel::Sender<SubscribeUpdateBlock>,
     commitment_level: CommitmentLevel,
+    exit_notfier: Arc<Notify>,
 ) -> AnyhowJoinHandle {
     tokio::spawn(async move {
         loop {
@@ -307,33 +307,44 @@ pub fn create_block_processing_task(
                 )
                 .await?;
 
-            while let Some(message) = stream.next().await {
-                let message = message?;
-
-                let Some(update) = message.update_oneof else {
-                    continue;
-                };
-
-                match update {
-                    UpdateOneof::Block(block) => {
-                        log::trace!(
-                            "received block, hash: {} slot: {}",
-                            block.blockhash,
-                            block.slot
-                        );
-                        block_sx
-                            .send(block)
-                            .await
-                            .context("Problem sending on block channel")?;
+            loop {
+                tokio::select! {
+                    message = stream.next() => {
+                        let Some(Ok(message)) = message else {
+                            break;
+                        };
+
+                        let Some(update) = message.update_oneof else {
+                            continue;
+                        };
+
+                        match update {
+                            UpdateOneof::Block(block) => {
+                                log::trace!(
+                                    "received block, hash: {} slot: {}",
+                                    block.blockhash,
+                                    block.slot
+                                );
+                                block_sx
+                                    .send(block)
+                                    .await
+                                    .context("Problem sending on block channel")?;
+                            }
+                            UpdateOneof::Ping(_) => {
+                                log::trace!("GRPC Ping");
+                            }
+                            _ => {
+                                log::trace!("unknown GRPC notification");
+                            }
+                        };
+                    },
+                    _ = exit_notfier.notified() => {
+                        break;
                     }
-                    UpdateOneof::Ping(_) => {
-                        log::trace!("GRPC Ping");
-                    }
-                    _ => {
-                        log::trace!("unknown GRPC notification");
-                    }
-                };
+                }
             }
+            drop(stream);
+            drop(client);
             log::error!("Grpc block subscription broken (resubscribing)");
             tokio::time::sleep(std::time::Duration::from_secs(1)).await;
         }

diff --git a/core/src/structures/prioritization_fee_heap.rs b/core/src/structures/prioritization_fee_heap.rs
@@ -85,7 +85,7 @@ impl PrioritizationFeesHeap {
         }
     }
 
-    pub async fn remove_expired_transactions(&self, current_blockheight: u64) {
+    pub async fn remove_expired_transactions(&self, current_blockheight: u64) -> usize {
         let mut write_lock = self.map.lock().await;
         let mut cells_to_remove = vec![];
         let mut signatures_to_remove = vec![];
@@ -104,9 +104,11 @@ impl PrioritizationFeesHeap {
         for p in cells_to_remove {
             write_lock.map.remove(&p);
         }
+        let signatures_len = signatures_to_remove.len();
         for sig in signatures_to_remove {
             write_lock.signatures.remove(&sig);
         }
+        signatures_len
     }
 
     pub async fn size(&self) -> usize {

diff --git a/run_clippy_fmt.sh b/run_clippy_fmt.sh
@@ -1,2 +1,2 @@
-cargo +nightly-2024-01-05 fmt --all
-cargo +nightly-2024-01-05 clippy --locked --workspace --all-targets -- -D warnings
+cargo +nightly-2023-10-05 fmt --all
+cargo +nightly-2023-10-05 clippy --locked --workspace --all-targets -- -D warnings