From 2d3e38f55a1704f53a62e0c214073a713a066732 Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Thu, 8 May 2025 20:08:03 +0000 Subject: [PATCH 1/7] fix more tests in blockstore.rs --- ledger/src/blockstore.rs | 40 ++++++++++++++++++++++++++++++++-------- 1 file changed, 32 insertions(+), 8 deletions(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index 8794dc19e0c..f63e2e0cb51 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -10775,19 +10775,43 @@ pub mod tests { #[test] fn test_duplicate_last_index() { - let num_shreds = 2; - let num_entries = max_ticks_per_n_shreds(num_shreds, None); let slot = 1; - let (mut shreds, _) = - make_slot_entries(slot, 0, num_entries, /*merkle_variant:*/ false); + let entries = make_slot_entries_with_transactions(1); + let leader_keypair = Arc::new(Keypair::new()); + let reed_solomon_cache = ReedSolomonCache::default(); + let shredder = Shredder::new(slot, 0, 0, 0).unwrap(); + let (shreds1, code1) = shredder.entries_to_shreds( + &leader_keypair, + &entries, + true, // is_last_in_slot + Some(Hash::new_unique()), + 0, // next_shred_index + 0, // next_code_index, + true, // merkle_variant + &reed_solomon_cache, + &mut ProcessShredsStats::default(), + ); + let last_data1 = shreds1.last().unwrap(); + let last_code1 = code1.last().unwrap(); + let (shreds2, _) = shredder.entries_to_shreds( + &leader_keypair, + &entries, + true, // is_last_in_slot + Some(Hash::new_unique()), + last_data1.index(), // next_shred_index + last_code1.index(), // next_code_index, + true, // merkle_variant + &reed_solomon_cache, + &mut ProcessShredsStats::default(), + ); + let last_data2 = shreds2.last().unwrap(); - // Mark both as last shred - shreds[0].set_last_in_slot(); - shreds[1].set_last_in_slot(); let ledger_path = get_tmp_ledger_path_auto_delete!(); let blockstore = Blockstore::open(ledger_path.path()).unwrap(); - blockstore.insert_shreds(shreds, None, false).unwrap(); + blockstore + .insert_shreds(vec![last_data1.clone(), last_data2.clone()], None, false) + .unwrap(); assert!(blockstore.get_duplicate_slot(slot).is_some()); } From 7c7fd806a314791a691e313ee8e2747bc1d8c815 Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Wed, 18 Jun 2025 16:36:45 +0000 Subject: [PATCH 2/7] fix more tests --- ledger/src/blockstore.rs | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index f63e2e0cb51..e90ea7b81a6 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -10773,9 +10773,9 @@ pub mod tests { assert!(!blockstore.is_dead(0)); } - #[test] - fn test_duplicate_last_index() { - let slot = 1; + fn setup_duplicate_last_in_slot( + slot: Slot, + ) -> ((Vec, Vec), (Vec, Vec)) { let entries = make_slot_entries_with_transactions(1); let leader_keypair = Arc::new(Keypair::new()); let reed_solomon_cache = ReedSolomonCache::default(); @@ -10793,7 +10793,7 @@ pub mod tests { ); let last_data1 = shreds1.last().unwrap(); let last_code1 = code1.last().unwrap(); - let (shreds2, _) = shredder.entries_to_shreds( + let (shreds2, code2) = shredder.entries_to_shreds( &leader_keypair, &entries, true, // is_last_in_slot @@ -10804,6 +10804,13 @@ pub mod tests { &reed_solomon_cache, &mut ProcessShredsStats::default(), ); + ((shreds1, code1), (shreds2, code2)) + } + + #[test] + fn test_duplicate_last_index() { + let slot = 1; + ((shreds1, code1), (shreds2, code2)) = setup_duplicate_last_in_slot(slot); let last_data2 = shreds2.last().unwrap(); let ledger_path = get_tmp_ledger_path_auto_delete!(); @@ -10819,16 +10826,13 @@ pub mod tests { #[test] fn test_duplicate_last_index_mark_dead() { let num_shreds = 10; - let smaller_last_shred_index = 5; + let smaller_last_shred_index = 32; let larger_last_shred_index = 8; let setup_test_shreds = |slot: Slot| -> Vec { - let num_entries = max_ticks_per_n_shreds(num_shreds, Some(LEGACY_SHRED_DATA_CAPACITY)); - let (mut shreds, _) = - make_slot_entries(slot, 0, num_entries, /*merkle_variant:*/ false); - shreds[smaller_last_shred_index].set_last_in_slot(); - shreds[larger_last_shred_index].set_last_in_slot(); - shreds + let ((mut shreds1, code1), (mut shreds2, code2)) = setup_duplicate_last_in_slot(slot); + shreds1.append(&mut shreds2); + shreds1 }; let get_expected_slot_meta_and_index_meta = @@ -10991,6 +10995,7 @@ pub mod tests { let setup_test_shreds = move |slot: Slot| -> Vec { let num_shreds = 10; let middle_shred_index = 5; + let num_entries = max_ticks_per_n_shreds(num_shreds, None); let (shreds, _) = make_slot_entries(slot, 0, num_entries, /*merkle_variant:*/ false); From 6f79d1c64b1b3d2c9e097222527cbf7a7113aa3c Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Mon, 23 Jun 2025 11:39:13 +0000 Subject: [PATCH 3/7] problem --- ledger/src/blockstore.rs | 76 +++++++++------------------------------- 1 file changed, 16 insertions(+), 60 deletions(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index e90ea7b81a6..cea1ce0eec9 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -5339,7 +5339,7 @@ pub mod tests { crate::{ genesis_utils::{create_genesis_config, GenesisConfigInfo}, leader_schedule::{FixedSchedule, IdentityKeyedLeaderSchedule}, - shred::{max_ticks_per_n_shreds, ShredFlags, LEGACY_SHRED_DATA_CAPACITY}, + shred::{max_ticks_per_n_shreds, ShredFlags}, }, assert_matches::assert_matches, bincode::{serialize, Options}, @@ -10773,6 +10773,9 @@ pub mod tests { assert!(!blockstore.is_dead(0)); } + /// Prepare two FEC sets of shreds for the same slot index + /// with reasonable shred indices, but in such a way that + /// both FEC sets include a shred with LAST_IN_SLOT flag set. fn setup_duplicate_last_in_slot( slot: Slot, ) -> ((Vec, Vec), (Vec, Vec)) { @@ -10797,7 +10800,7 @@ pub mod tests { &leader_keypair, &entries, true, // is_last_in_slot - Some(Hash::new_unique()), + Some(last_data1.chained_merkle_root().unwrap()), last_data1.index(), // next_shred_index last_code1.index(), // next_code_index, true, // merkle_variant @@ -10810,9 +10813,10 @@ pub mod tests { #[test] fn test_duplicate_last_index() { let slot = 1; - ((shreds1, code1), (shreds2, code2)) = setup_duplicate_last_in_slot(slot); - let last_data2 = shreds2.last().unwrap(); + let ((shreds1, _code1), (shreds2, _code2)) = setup_duplicate_last_in_slot(slot); + let last_data1 = shreds1.last().unwrap(); + let last_data2 = shreds2.last().unwrap(); let ledger_path = get_tmp_ledger_path_auto_delete!(); let blockstore = Blockstore::open(ledger_path.path()).unwrap(); @@ -10826,11 +10830,11 @@ pub mod tests { #[test] fn test_duplicate_last_index_mark_dead() { let num_shreds = 10; - let smaller_last_shred_index = 32; + let smaller_last_shred_index = 31; let larger_last_shred_index = 8; let setup_test_shreds = |slot: Slot| -> Vec { - let ((mut shreds1, code1), (mut shreds2, code2)) = setup_duplicate_last_in_slot(slot); + let ((mut shreds1, _code1), (mut shreds2, _code2)) = setup_duplicate_last_in_slot(slot); shreds1.append(&mut shreds2); shreds1 }; @@ -10891,38 +10895,6 @@ pub mod tests { assert_eq!(meta, expected_slot_meta); assert_eq!(blockstore.get_index(slot).unwrap().unwrap(), expected_index); - // Case 2: Inserting a duplicate with an even smaller last shred index should not - // mark the slot as dead since the Slotmeta is full. - let even_smaller_last_shred_duplicate = { - let mut payload = shreds[smaller_last_shred_index - 1].payload().clone(); - // Flip a byte to create a duplicate shred - payload[0] = u8::MAX - payload[0]; - let mut shred = Shred::new_from_serialized_shred(payload).unwrap(); - shred.set_last_in_slot(); - shred - }; - assert!(blockstore - .is_shred_duplicate(&even_smaller_last_shred_duplicate) - .is_some()); - blockstore - .insert_shreds(vec![even_smaller_last_shred_duplicate], None, false) - .unwrap(); - assert!(!blockstore.is_dead(slot)); - for i in 0..num_shreds { - if i <= smaller_last_shred_index as u64 { - assert_eq!( - blockstore.get_data_shred(slot, i).unwrap().unwrap(), - shreds[i as usize].payload().as_ref(), - ); - } else { - assert!(blockstore.get_data_shred(slot, i).unwrap().is_none()); - } - } - let mut meta = blockstore.meta(slot).unwrap().unwrap(); - meta.first_shred_timestamp = expected_slot_meta.first_shred_timestamp; - assert_eq!(meta, expected_slot_meta); - assert_eq!(blockstore.get_index(slot).unwrap().unwrap(), expected_index); - // Case 3: Insert shreds in reverse so that consumed will not be updated. Now on insert, the // the slot should be marked as dead slot += 1; @@ -10931,6 +10903,9 @@ pub mod tests { blockstore .insert_shreds(shreds.clone(), None, false) .unwrap(); + for s in shreds.iter() { + dbg!(s.last_in_slot()); + } assert!(blockstore.is_dead(slot)); // All the shreds other than the two last index shreds because those two // are marked as last, but less than the first received index == 10. @@ -10992,25 +10967,6 @@ pub mod tests { #[test] fn test_get_slot_entries_dead_slot_race() { - let setup_test_shreds = move |slot: Slot| -> Vec { - let num_shreds = 10; - let middle_shred_index = 5; - - let num_entries = max_ticks_per_n_shreds(num_shreds, None); - let (shreds, _) = - make_slot_entries(slot, 0, num_entries, /*merkle_variant:*/ false); - - // Reverse shreds so that last shred gets inserted first and sets meta.received - let mut shreds: Vec = shreds.into_iter().rev().collect(); - - // Push the real middle shred to the end of the shreds list - shreds.push(shreds[middle_shred_index].clone()); - - // Set the middle shred as a last shred to cause the slot to be marked dead - shreds[middle_shred_index].set_last_in_slot(); - shreds - }; - let ledger_path = get_tmp_ledger_path_auto_delete!(); { let blockstore = Arc::new(Blockstore::open(ledger_path.path()).unwrap()); @@ -11069,11 +11025,11 @@ pub mod tests { }; for slot in 0..100 { - let shreds = setup_test_shreds(slot); - + let ((mut shreds1, _), (mut shreds2, _)) = setup_duplicate_last_in_slot(slot); + shreds1.append(&mut shreds2); // Start a task on each thread to trigger a race condition slot_sender.send(slot).unwrap(); - shred_sender.send(shreds).unwrap(); + shred_sender.send(shreds1).unwrap(); // Check that each thread processed their task before continuing for _ in 1..=2 { From 2f2b6f188801b9eae7074176264b1de8ec9be61e Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Wed, 23 Jul 2025 13:25:39 +0000 Subject: [PATCH 4/7] stop using near-deprecated function --- ledger/src/blockstore.rs | 50 ++++++++++++++++++++-------------------- 1 file changed, 25 insertions(+), 25 deletions(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index cea1ce0eec9..aa479425455 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -10783,30 +10783,33 @@ pub mod tests { let leader_keypair = Arc::new(Keypair::new()); let reed_solomon_cache = ReedSolomonCache::default(); let shredder = Shredder::new(slot, 0, 0, 0).unwrap(); - let (shreds1, code1) = shredder.entries_to_shreds( - &leader_keypair, - &entries, - true, // is_last_in_slot - Some(Hash::new_unique()), - 0, // next_shred_index - 0, // next_code_index, - true, // merkle_variant - &reed_solomon_cache, - &mut ProcessShredsStats::default(), - ); + let (shreds1, code1): (Vec, Vec) = shredder + .make_merkle_shreds_from_entries( + &leader_keypair, + &entries, + true, // is_last_in_slot + Some(Hash::new_unique()), + 0, // next_shred_index + 0, // next_code_index, + &reed_solomon_cache, + &mut ProcessShredsStats::default(), + ) + .partition(Shred::is_data); let last_data1 = shreds1.last().unwrap(); let last_code1 = code1.last().unwrap(); - let (shreds2, code2) = shredder.entries_to_shreds( - &leader_keypair, - &entries, - true, // is_last_in_slot - Some(last_data1.chained_merkle_root().unwrap()), - last_data1.index(), // next_shred_index - last_code1.index(), // next_code_index, - true, // merkle_variant - &reed_solomon_cache, - &mut ProcessShredsStats::default(), - ); + + let (shreds2, code2) = shredder + .make_merkle_shreds_from_entries( + &leader_keypair, + &entries, + true, // is_last_in_slot + Some(last_data1.chained_merkle_root().unwrap()), + last_data1.index() + 1, // next_shred_index + last_code1.index() + 1, // next_code_index, + &reed_solomon_cache, + &mut ProcessShredsStats::default(), + ) + .partition(Shred::is_data); ((shreds1, code1), (shreds2, code2)) } @@ -10903,9 +10906,6 @@ pub mod tests { blockstore .insert_shreds(shreds.clone(), None, false) .unwrap(); - for s in shreds.iter() { - dbg!(s.last_in_slot()); - } assert!(blockstore.is_dead(slot)); // All the shreds other than the two last index shreds because those two // are marked as last, but less than the first received index == 10. From 1a54e9aae35aee992857d42382c6669d5ad41efb Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Wed, 23 Jul 2025 13:33:21 +0000 Subject: [PATCH 5/7] keep clippy happy --- ledger/src/blockstore.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index aa479425455..da89b034793 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -10776,6 +10776,7 @@ pub mod tests { /// Prepare two FEC sets of shreds for the same slot index /// with reasonable shred indices, but in such a way that /// both FEC sets include a shred with LAST_IN_SLOT flag set. + #[allow(clippy::type_complexity)] fn setup_duplicate_last_in_slot( slot: Slot, ) -> ((Vec, Vec), (Vec, Vec)) { From 2b9df4ac1f4a8251de0b3e6ccc56a8ced9a27dc0 Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Wed, 23 Jul 2025 14:01:24 +0000 Subject: [PATCH 6/7] fix the assert --- ledger/src/blockstore.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index da89b034793..c46e2d1b53d 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -11017,7 +11017,7 @@ pub mod tests { blockstore.lowest_cleanup_slot.write().unwrap(); blockstore.insert_shreds(shreds, None, false).unwrap(); assert!(blockstore.get_duplicate_slot(slot).is_some()); - assert!(blockstore.is_dead(slot)); + assert!(!blockstore.is_dead(slot)); assert!(blockstore.meta(slot).unwrap().unwrap().is_full()); signal_sender.send(Ok(())).unwrap(); } From ecaf48ed56d5bd4c812f5416fe8e3f8a560bdf69 Mon Sep 17 00:00:00 2001 From: Alex Pyattaev Date: Wed, 23 Jul 2025 21:13:13 +0000 Subject: [PATCH 7/7] fix test_get_slot_entries_dead_slot_race --- ledger/src/blockstore.rs | 118 +++++++++++++++++---------------------- 1 file changed, 51 insertions(+), 67 deletions(-) diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index c46e2d1b53d..c11d4c80cab 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -5366,7 +5366,7 @@ pub mod tests { solana_transaction_status::{ InnerInstruction, InnerInstructions, Reward, Rewards, TransactionTokenBalance, }, - std::{cmp::Ordering, thread::Builder, time::Duration}, + std::{cmp::Ordering, time::Duration}, test_case::test_case, }; @@ -10975,79 +10975,63 @@ pub mod tests { let (shred_sender, shred_receiver) = unbounded::>(); let (signal_sender, signal_receiver) = unbounded(); - let t_entry_getter = { - let blockstore = blockstore.clone(); - let signal_sender = signal_sender.clone(); - Builder::new() - .spawn(move || { - while let Ok(slot) = slot_receiver.recv() { - match blockstore.get_slot_entries_with_shred_info(slot, 0, false) { - Ok((_entries, _num_shreds, is_full)) => { - if is_full { - signal_sender - .send(Err(IoError::other( - "got full slot entries for dead slot", - ))) - .unwrap(); - } - } - Err(err) => { - assert_matches!(err, BlockstoreError::DeadSlot); + std::thread::scope(|scope| { + scope.spawn(|| { + while let Ok(slot) = slot_receiver.recv() { + match blockstore.get_slot_entries_with_shred_info(slot, 0, false) { + Ok((_entries, _num_shreds, is_full)) => { + if is_full { + signal_sender + .send(Err(IoError::other( + "got full slot entries for dead slot", + ))) + .unwrap(); } } - signal_sender.send(Ok(())).unwrap(); + Err(err) => { + assert_matches!(err, BlockstoreError::DeadSlot); + } } - }) - .unwrap() - }; + signal_sender.send(Ok(())).unwrap(); + } + }); - let t_shred_inserter = { - let blockstore = blockstore.clone(); - Builder::new() - .spawn(move || { - while let Ok(shreds) = shred_receiver.recv() { - let slot = shreds[0].slot(); - // Grab this lock to block `get_slot_entries` before it fetches completed datasets - // and then mark the slot as dead, but full, by inserting carefully crafted shreds. - - #[allow(clippy::readonly_write_lock)] - // Possible clippy bug, the lock is unused so clippy shouldn't care - // about read vs. write lock - let _lowest_cleanup_slot = - blockstore.lowest_cleanup_slot.write().unwrap(); - blockstore.insert_shreds(shreds, None, false).unwrap(); - assert!(blockstore.get_duplicate_slot(slot).is_some()); - assert!(!blockstore.is_dead(slot)); - assert!(blockstore.meta(slot).unwrap().unwrap().is_full()); - signal_sender.send(Ok(())).unwrap(); - } - }) - .unwrap() - }; + scope.spawn(|| { + while let Ok(shreds) = shred_receiver.recv() { + let slot = shreds[0].slot(); + // Grab this lock to block `get_slot_entries` before it fetches completed datasets + // and then mark the slot as dead, but full, by inserting carefully crafted shreds. + + #[allow(clippy::readonly_write_lock)] + // Possible clippy bug, the lock is unused so clippy shouldn't care + // about read vs. write lock + let _lowest_cleanup_slot = blockstore.lowest_cleanup_slot.write().unwrap(); + blockstore.insert_shreds(shreds, None, false).unwrap(); + assert!(blockstore.get_duplicate_slot(slot).is_some()); + assert!(blockstore.is_dead(slot)); + signal_sender.send(Ok(())).unwrap(); + } + }); - for slot in 0..100 { - let ((mut shreds1, _), (mut shreds2, _)) = setup_duplicate_last_in_slot(slot); - shreds1.append(&mut shreds2); - // Start a task on each thread to trigger a race condition - slot_sender.send(slot).unwrap(); - shred_sender.send(shreds1).unwrap(); - - // Check that each thread processed their task before continuing - for _ in 1..=2 { - let res = signal_receiver.recv().unwrap(); - assert!(res.is_ok(), "race condition: {res:?}"); + for slot in 0..100 { + let ((mut shreds1, _), (mut shreds2, _)) = setup_duplicate_last_in_slot(slot); + // compose shreds in reverse order of FEC sets to + // make sure slot is marked dead + shreds2.append(&mut shreds1); + // Start a task on each thread to trigger a race condition + slot_sender.send(slot).unwrap(); + shred_sender.send(shreds2).unwrap(); + + // Check that each thread processed their task before continuing + for _ in 1..=2 { + let res = signal_receiver.recv().unwrap(); + assert!(res.is_ok(), "race condition: {res:?}"); + } } - } - drop(slot_sender); - drop(shred_sender); - - let handles = vec![t_entry_getter, t_shred_inserter]; - for handle in handles { - assert!(handle.join().is_ok()); - } - - assert!(Arc::strong_count(&blockstore) == 1); + drop(slot_sender); + drop(shred_sender); + }); } }