resets packet flags obtained from QUIC datagrams (#32673)

Packets obtained from recycler have dirty meta information and need to
re-initialized.

core/src/shred_fetch_stage.rs

@@ -12,7 +12,7 @@ use {
     solana_sdk::{
         clock::{Slot, DEFAULT_MS_PER_SLOT},
         feature_set,
-        packet::PACKET_DATA_SIZE,
+        packet::{Meta, PACKET_DATA_SIZE},
         pubkey::Pubkey,
     },
     solana_streamer::streamer::{self, PacketBatchReceiver, StreamerReceiveStats},
@@ -276,10 +276,13 @@ fn receive_quic_datagrams(
             .filter(|(_, _, bytes)| bytes.len() <= PACKET_DATA_SIZE)
             .zip(packet_batch.iter_mut())
             .map(|((_pubkey, addr, bytes), packet)| {
+                *packet.meta_mut() = Meta {
+                    size: bytes.len(),
+                    addr: addr.ip(),
+                    port: addr.port(),
+                    flags: PacketFlags::empty(),
+                };
                 packet.buffer_mut()[..bytes.len()].copy_from_slice(&bytes);
-                let meta = packet.meta_mut();
-                meta.size = bytes.len();
-                meta.set_socket_addr(&addr);
             })
             .count();
         if size > 0 {

core/src/validator.rs

@@ -1332,7 +1332,6 @@ impl Validator {
     pub fn join(self) {
         drop(self.bank_forks);
         drop(self.cluster_info);
-        solana_turbine::quic_endpoint::close_quic_endpoint(&self.turbine_quic_endpoint);
 
         self.poh_service.join().expect("poh_service");
         drop(self.poh_recorder);
@@ -1413,6 +1412,7 @@ impl Validator {
         self.accounts_hash_verifier
             .join()
             .expect("accounts_hash_verifier");
+        solana_turbine::quic_endpoint::close_quic_endpoint(&self.turbine_quic_endpoint);
         self.tpu.join().expect("tpu");
         self.tvu.join().expect("tvu");
         self.turbine_quic_endpoint_runtime