Closing QUIC connection correctly, changing some QUIC params

services/src/quic_connection.rs

@@ -4,7 +4,7 @@ use crate::quic_connection_utils::{
 use futures::FutureExt;
 use log::warn;
 use prometheus::{core::GenericGauge, opts, register_int_gauge};
-use quinn::{Connection, Endpoint};
+use quinn::{Connection, Endpoint, VarInt};
 use solana_lite_rpc_core::structures::rotating_queue::RotatingQueue;
 use solana_sdk::pubkey::Pubkey;
 use std::{
@@ -225,6 +225,13 @@ impl QuicConnection {
             None => false,
         }
     }
+
+    pub async fn close(&self) {
+        let lk = self.connection.read().await;
+        if let Some(connection) = lk.as_ref() {
+            connection.close(VarInt::from_u32(0), b"Not needed");
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -319,4 +326,10 @@ impl QuicConnectionPool {
     pub fn is_empty(&self) -> bool {
         self.connections.is_empty()
     }
+
+    pub async fn close_all(&self) {
+        for connection in &self.connections {
+            connection.close().await;
+        }
+    }
 }

services/src/quic_connection_utils.rs

@@ -92,7 +92,7 @@ impl Default for QuicConnectionParameters {
             connection_timeout: Duration::from_millis(60000),
             unistream_timeout: Duration::from_millis(10000),
             write_timeout: Duration::from_millis(10000),
-            finalize_timeout: Duration::from_millis(10000),
+            finalize_timeout: Duration::from_millis(20000),
             connection_retry_count: 20,
             max_number_of_connections: 8,
             number_of_transactions_per_unistream: 1,

services/src/tpu_utils/tpu_connection_manager.rs

@@ -223,6 +223,7 @@ impl ActiveConnection {
         let elements_removed = priorization_heap.clear().await;
         TRANSACTIONS_IN_HEAP.sub(elements_removed as i64);
         NB_QUIC_ACTIVE_CONNECTIONS.dec();
+        connection_pool.close_all().await;
     }
 
     pub fn start_listening(

services/src/tpu_utils/tpu_service.rs

@@ -120,6 +120,7 @@ impl TpuService {
     ) -> anyhow::Result<()> {
         let fanout = self.config.fanout_slots;
         let last_slot = estimated_slot + fanout;
+        let current_slot = current_slot.saturating_sub(4);
 
         let cluster_nodes = self.data_cache.cluster_info.cluster_nodes.clone();