send duplicate shred proofs for conflicting shred scenarios

core/src/window_service.rs

@@ -19,7 +19,7 @@ use {
     rayon::{prelude::*, ThreadPool},
     solana_gossip::cluster_info::ClusterInfo,
     solana_ledger::{
-        blockstore::{Blockstore, BlockstoreInsertionMetrics},
+        blockstore::{Blockstore, BlockstoreInsertionMetrics, PossibleDuplicateShred},
         leader_schedule_cache::LeaderScheduleCache,
         shred::{self, Nonce, ReedSolomonCache, Shred},
     },
@@ -138,23 +138,37 @@ impl WindowServiceMetrics {
 fn run_check_duplicate(
     cluster_info: &ClusterInfo,
     blockstore: &Blockstore,
-    shred_receiver: &Receiver<Shred>,
+    shred_receiver: &Receiver<PossibleDuplicateShred>,
     duplicate_slots_sender: &DuplicateSlotSender,
 ) -> Result<()> {
-    let check_duplicate = |shred: Shred| -> Result<()> {
+    let check_duplicate = |shred: PossibleDuplicateShred| -> Result<()> {
         let shred_slot = shred.slot();
-        if !blockstore.has_duplicate_shreds_in_slot(shred_slot) {
-            if let Some(existing_shred_payload) = blockstore.is_shred_duplicate(&shred) {
-                cluster_info.push_duplicate_shred(&shred, &existing_shred_payload)?;
+        let (shred1, shred2) = match shred {
+            PossibleDuplicateShred::LastIndexConflict(shred, conflict) => (shred, conflict),
+            PossibleDuplicateShred::ErasureConflict(shred, conflict) => (shred, conflict),
+            PossibleDuplicateShred::Exists(shred) => {
+                // Unlike the other cases we have to wait until here to decide to handle the duplicate and store
+                // in blockstore. This is because the duplicate could have been part of the same insert batch,
+                // so we wait until the batch has been written.
+                if blockstore.has_duplicate_shreds_in_slot(shred_slot) {
+                    return Ok(()); // A duplicate is already recorded
+                }
+                let Some(existing_shred_payload) = blockstore.is_shred_duplicate(&shred) else {
+                    return Ok(()); // Not a duplicate
+                };
                 blockstore.store_duplicate_slot(
                     shred_slot,
-                    existing_shred_payload,
-                    shred.into_payload(),
+                    existing_shred_payload.clone(),
+                    shred.clone().into_payload(),
                 )?;
-
-                duplicate_slots_sender.send(shred_slot)?;
+                (shred, existing_shred_payload)
             }
-        }
+        };
+
+        // Propagate duplicate proof through gossip
+        cluster_info.push_duplicate_shred(&shred1, &shred2)?;
+        // Notify duplicate consensus state machine
+        duplicate_slots_sender.send(shred_slot)?;
 
         Ok(())
     };
@@ -226,7 +240,7 @@ fn run_insert<F>(
     reed_solomon_cache: &ReedSolomonCache,
 ) -> Result<()>
 where
-    F: Fn(Shred),
+    F: Fn(PossibleDuplicateShred),
 {
     const RECV_TIMEOUT: Duration = Duration::from_millis(200);
     let mut shred_receiver_elapsed = Measure::start("shred_receiver_elapsed");
@@ -370,7 +384,7 @@ impl WindowService {
         cluster_info: Arc<ClusterInfo>,
         exit: Arc<AtomicBool>,
         blockstore: Arc<Blockstore>,
-        duplicate_receiver: Receiver<Shred>,
+        duplicate_receiver: Receiver<PossibleDuplicateShred>,
         duplicate_slots_sender: DuplicateSlotSender,
     ) -> JoinHandle<()> {
         let handle_error = || {
@@ -400,7 +414,7 @@ impl WindowService {
         blockstore: Arc<Blockstore>,
         leader_schedule_cache: Arc<LeaderScheduleCache>,
         verified_receiver: Receiver<Vec<PacketBatch>>,
-        check_duplicate_sender: Sender<Shred>,
+        check_duplicate_sender: Sender<PossibleDuplicateShred>,
         completed_data_sets_sender: CompletedDataSetsSender,
         retransmit_sender: Sender<Vec<ShredPayload>>,
         outstanding_requests: Arc<RwLock<OutstandingShredRepairs>>,
@@ -417,8 +431,8 @@ impl WindowService {
         Builder::new()
             .name("solWinInsert".to_string())
             .spawn(move || {
-                let handle_duplicate = |shred| {
-                    let _ = check_duplicate_sender.send(shred);
+                let handle_duplicate = |possible_duplicate_shred| {
+                    let _ = check_duplicate_sender.send(possible_duplicate_shred);
                 };
                 let mut metrics = BlockstoreInsertionMetrics::default();
                 let mut ws_metrics = WindowServiceMetrics::default();
@@ -551,7 +565,9 @@ mod test {
         };
         assert_eq!(duplicate_shred.slot(), shreds[0].slot());
         let duplicate_shred_slot = duplicate_shred.slot();
-        sender.send(duplicate_shred.clone()).unwrap();
+        sender
+            .send(PossibleDuplicateShred::Exists(duplicate_shred.clone()))
+            .unwrap();
         assert!(!blockstore.has_duplicate_shreds_in_slot(duplicate_shred_slot));
         let keypair = Keypair::new();
         let contact_info = ContactInfo::new_localhost(&keypair.pubkey(), timestamp());

gossip/Cargo.toml

@@ -53,6 +53,7 @@ thiserror = { workspace = true }
 [dev-dependencies]
 num_cpus = { workspace = true }
 serial_test = { workspace = true }
+test-case = { workspace = true }
 
 [build-dependencies]
 rustc_version = { workspace = true }

gossip/src/cluster_info.rs

@@ -271,7 +271,7 @@ pub fn make_accounts_hashes_message(
 pub(crate) type Ping = ping_pong::Ping<[u8; GOSSIP_PING_TOKEN_SIZE]>;
 
 // TODO These messages should go through the gpu pipeline for spam filtering
-#[frozen_abi(digest = "6T2sn92PMrTijsgncH3bBZL4K5GUowb442cCw4y4DuwV")]
+#[frozen_abi(digest = "EnbW8mYTsPMndq9NkHLTkHJgduXvWSfSD6bBdmqQ8TiF")]
 #[derive(Serialize, Deserialize, Debug, AbiEnumVisitor, AbiExample)]
 #[allow(clippy::large_enum_variant)]
 pub(crate) enum Protocol {

gossip/src/crds_gossip.rs

@@ -149,7 +149,7 @@ impl CrdsGossip {
         let now = timestamp();
         for entry in entries {
             if let Err(err) = crds.insert(entry, now, GossipRoute::LocalMessage) {
-                error!("push_duplicate_shred faild: {:?}", err);
+                error!("push_duplicate_shred failed: {:?}", err);
             }
         }
         Ok(())