diff --git a/core/src/transaction_status_service.rs b/core/src/transaction_status_service.rs index 677bfa739b..92ea1de135 100644 --- a/core/src/transaction_status_service.rs +++ b/core/src/transaction_status_service.rs @@ -70,9 +70,14 @@ impl TransactionStatusService { } .expect("FeeCalculator must exist"); let fee = fee_calculator.calculate_fee(transaction.message()); + let (writable_keys, readonly_keys) = + transaction.message.get_account_keys_by_lock_type(); blockstore .write_transaction_status( - (transaction.signatures[0], slot), + slot, + transaction.signatures[0], + writable_keys, + readonly_keys, &TransactionStatusMeta { status, fee, diff --git a/ledger/src/blockstore.rs b/ledger/src/blockstore.rs index 9f05432e94..f8f2311ba5 100644 --- a/ledger/src/blockstore.rs +++ b/ledger/src/blockstore.rs @@ -87,6 +87,7 @@ pub struct Blockstore { data_shred_cf: LedgerColumn, code_shred_cf: LedgerColumn, transaction_status_cf: LedgerColumn, + address_signatures_cf: LedgerColumn, transaction_status_index_cf: LedgerColumn, active_transaction_status_index: RwLock, rewards_cf: LedgerColumn, @@ -202,6 +203,7 @@ impl Blockstore { let data_shred_cf = db.column(); let code_shred_cf = db.column(); let transaction_status_cf = db.column(); + let address_signatures_cf = db.column(); let transaction_status_index_cf = db.column(); let rewards_cf = db.column(); @@ -218,7 +220,9 @@ impl Blockstore { // Get active transaction-status index or 0 let active_transaction_status_index = db .iter::(IteratorMode::Start)? - .next() + .next(); + let initialize_transaction_status_index = active_transaction_status_index.is_none(); + let active_transaction_status_index = active_transaction_status_index .and_then(|(_, data)| { let index0: TransactionStatusIndexMeta = deserialize(&data).unwrap(); if index0.frozen { @@ -242,6 +246,7 @@ impl Blockstore { data_shred_cf, code_shred_cf, transaction_status_cf, + address_signatures_cf, transaction_status_index_cf, active_transaction_status_index: RwLock::new(active_transaction_status_index), rewards_cf, @@ -252,6 +257,9 @@ impl Blockstore { lowest_cleanup_slot: Arc::new(RwLock::new(0)), no_compaction: false, }; + if initialize_transaction_status_index { + blockstore.initialize_transaction_status_index()?; + } Ok(blockstore) } @@ -387,10 +395,14 @@ impl Blockstore { &mut w_active_transaction_status_index, to_slot, )? { - columns_empty &= &self + columns_empty &= self .db .delete_range_cf::(&mut write_batch, index, index + 1) - .unwrap_or(false); + .unwrap_or(false) + & self + .db + .delete_range_cf::(&mut write_batch, index, index + 1) + .unwrap_or(false); } let mut write_timer = Measure::start("write_batch"); if let Err(e) = self.db.write(write_batch) { @@ -450,6 +462,14 @@ impl Blockstore { .transaction_status_cf .compact_range(0, 2) .unwrap_or(false) + && self + .address_signatures_cf + .compact_range(0, 2) + .unwrap_or(false) + && self + .transaction_status_index_cf + .compact_range(0, 2) + .unwrap_or(false) && self .rewards_cf .compact_range(from_slot, to_slot) @@ -1546,8 +1566,12 @@ impl Blockstore { .put(1, &TransactionStatusIndexMeta::default())?; // This dummy status improves compaction performance self.transaction_status_cf.put( - (2, Signature::default(), 0), + cf::TransactionStatus::as_index(2), &TransactionStatusMeta::default(), + )?; + self.address_signatures_cf.put( + cf::AddressSignatures::as_index(2), + &AddressSignatureMeta::default(), ) } @@ -1598,15 +1622,11 @@ impl Blockstore { } } - fn make_transaction_status_index( + fn get_primary_index( &self, - index: (Signature, Slot), + slot: Slot, w_active_transaction_status_index: &mut u64, - ) -> Result<(u64, Signature, Slot)> { - let (signature, slot) = index; - if self.transaction_status_index_cf.get(0)?.is_none() { - self.initialize_transaction_status_index()?; - } + ) -> Result { let i = *w_active_transaction_status_index; let mut index_meta = self.transaction_status_index_cf.get(i)?.unwrap(); if slot > index_meta.max_slot { @@ -1614,7 +1634,7 @@ impl Blockstore { index_meta.max_slot = slot; self.transaction_status_index_cf.put(i, &index_meta)?; } - Ok((i, signature, slot)) + Ok(i) } pub fn read_transaction_status( @@ -1632,16 +1652,38 @@ impl Blockstore { pub fn write_transaction_status( &self, - index: (Signature, Slot), + slot: Slot, + signature: Signature, + writable_keys: Vec<&Pubkey>, + readonly_keys: Vec<&Pubkey>, status: &TransactionStatusMeta, ) -> Result<()> { - // This write lock prevents interleaving issues with the transactions_status_index_cf by - // gating writes to that column + // This write lock prevents interleaving issues with the transaction_status_index_cf by gating + // writes to that column let mut w_active_transaction_status_index = self.active_transaction_status_index.write().unwrap(); - let index = - self.make_transaction_status_index(index, &mut w_active_transaction_status_index)?; - self.transaction_status_cf.put(index, status) + let primary_index = self.get_primary_index(slot, &mut w_active_transaction_status_index)?; + self.transaction_status_cf + .put((primary_index, signature, slot), status)?; + for address in writable_keys { + self.address_signatures_cf.put( + (primary_index, *address, slot), + &AddressSignatureMeta { + signature, + writeable: true, + }, + )?; + } + for address in readonly_keys { + self.address_signatures_cf.put( + (primary_index, *address, slot), + &AddressSignatureMeta { + signature, + writeable: false, + }, + )?; + } + Ok(()) } // Returns a transaction status if it was processed in a root, as well as a loop counter for @@ -2805,7 +2847,18 @@ pub mod tests { .iter::(IteratorMode::Start) .unwrap() .next() - .map(|((_, _, slot), _)| slot >= min_slot) + .map(|((primary_index, _, slot), _)| { + slot >= min_slot || (primary_index == 2 && slot == 0) + }) + .unwrap_or(true) + & blockstore + .db + .iter::(IteratorMode::Start) + .unwrap() + .next() + .map(|((primary_index, _, slot), _)| { + slot >= min_slot || (primary_index == 2 && slot == 0) + }) .unwrap_or(true) & blockstore .db @@ -5378,14 +5431,18 @@ pub mod tests { let transaction_status_index_cf = blockstore.db.column::(); let slot0 = 10; - assert!(transaction_status_index_cf.get(0).unwrap().is_none()); - assert!(transaction_status_index_cf.get(1).unwrap().is_none()); + // Primary index column is initialized on Blockstore::open + assert!(transaction_status_index_cf.get(0).unwrap().is_some()); + assert!(transaction_status_index_cf.get(1).unwrap().is_some()); for _ in 0..5 { let random_bytes: Vec = (0..64).map(|_| rand::random::()).collect(); blockstore .write_transaction_status( - (Signature::new(&random_bytes), slot0), + slot0, + Signature::new(&random_bytes), + vec![&Pubkey::new(&random_bytes[0..32])], + vec![&Pubkey::new(&random_bytes[32..])], &TransactionStatusMeta::default(), ) .unwrap(); @@ -5416,6 +5473,18 @@ pub mod tests { .0; assert_eq!(first_status_entry.0, 0); assert_eq!(first_status_entry.2, slot0); + let first_address_entry = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap() + .next() + .unwrap() + .0; + assert_eq!(first_address_entry.0, 0); + assert_eq!(first_address_entry.2, slot0); blockstore.run_purge(0, 8).unwrap(); // First successful prune freezes index 0 @@ -5436,7 +5505,10 @@ pub mod tests { let random_bytes: Vec = (0..64).map(|_| rand::random::()).collect(); blockstore .write_transaction_status( - (Signature::new(&random_bytes), slot1), + slot1, + Signature::new(&random_bytes), + vec![&Pubkey::new(&random_bytes[0..32])], + vec![&Pubkey::new(&random_bytes[32..])], &TransactionStatusMeta::default(), ) .unwrap(); @@ -5458,7 +5530,7 @@ pub mod tests { } ); - // Index 0 statuses still exist + // Index 0 statuses and address records still exist let first_status_entry = blockstore .db .iter::(IteratorMode::From( @@ -5471,7 +5543,19 @@ pub mod tests { .0; assert_eq!(first_status_entry.0, 0); assert_eq!(first_status_entry.2, 10); - // New statuses are stored in index 1 + let first_address_entry = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap() + .next() + .unwrap() + .0; + assert_eq!(first_address_entry.0, 0); + assert_eq!(first_address_entry.2, slot0); + // New statuses and address records are stored in index 1 let index1_first_status_entry = blockstore .db .iter::(IteratorMode::From( @@ -5484,6 +5568,18 @@ pub mod tests { .0; assert_eq!(index1_first_status_entry.0, 1); assert_eq!(index1_first_status_entry.2, slot1); + let index1_first_address_entry = blockstore + .db + .iter::(IteratorMode::From( + (1, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap() + .next() + .unwrap() + .0; + assert_eq!(index1_first_address_entry.0, 1); + assert_eq!(index1_first_address_entry.2, slot1); blockstore.run_purge(0, 18).unwrap(); // Successful prune toggles TransactionStatusIndex @@ -5502,7 +5598,7 @@ pub mod tests { } ); - // Index 0 has been pruned, so first status entry is now index 1 + // Index 0 has been pruned, so first status and address entries are now index 1 let first_status_entry = blockstore .db .iter::(IteratorMode::From( @@ -5515,6 +5611,18 @@ pub mod tests { .0; assert_eq!(first_status_entry.0, 1); assert_eq!(first_status_entry.2, slot1); + let first_address_entry = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap() + .next() + .unwrap() + .0; + assert_eq!(first_address_entry.0, 1); + assert_eq!(first_address_entry.2, slot1); } Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction"); } @@ -5530,7 +5638,10 @@ pub mod tests { let random_bytes: Vec = (0..64).map(|_| rand::random::()).collect(); blockstore .write_transaction_status( - (Signature::new(&random_bytes), slot), + slot, + Signature::new(&random_bytes), + vec![&Pubkey::new(&random_bytes[0..32])], + vec![&Pubkey::new(&random_bytes[32..])], &TransactionStatusMeta::default(), ) .unwrap(); @@ -5549,6 +5660,18 @@ pub mod tests { assert_eq!(entry.0, 0); assert_eq!(entry.2, slot); } + let mut address_transactions_iterator = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap(); + for _ in 0..10 { + let entry = address_transactions_iterator.next().unwrap().0; + assert_eq!(entry.0, 0); + assert_eq!(entry.2, slot); + } assert_eq!( transaction_status_index_cf.get(0).unwrap().unwrap(), TransactionStatusIndexMeta { @@ -5571,6 +5694,18 @@ pub mod tests { assert_eq!(entry.0, 0); assert_eq!(entry.2, slot); } + let mut address_transactions_iterator = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap(); + for _ in 0..10 { + let entry = address_transactions_iterator.next().unwrap().0; + assert_eq!(entry.0, 0); + assert_eq!(entry.2, slot); + } assert_eq!( transaction_status_index_cf.get(0).unwrap().unwrap(), TransactionStatusIndexMeta { @@ -5593,6 +5728,18 @@ pub mod tests { assert_eq!(entry.0, 0); assert_eq!(entry.2, slot); } + let mut address_transactions_iterator = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap(); + for _ in 0..10 { + let entry = address_transactions_iterator.next().unwrap().0; + assert_eq!(entry.0, 0); + assert_eq!(entry.2, slot); + } assert_eq!( transaction_status_index_cf.get(0).unwrap().unwrap(), TransactionStatusIndexMeta { @@ -5613,6 +5760,17 @@ pub mod tests { assert_eq!(padding_entry.0, 2); assert_eq!(padding_entry.2, 0); assert!(status_entry_iterator.next().is_none()); + let mut address_transactions_iterator = blockstore + .db + .iter::(IteratorMode::From( + (0, Pubkey::default(), 0), + IteratorDirection::Forward, + )) + .unwrap(); + let padding_entry = address_transactions_iterator.next().unwrap().0; + assert_eq!(padding_entry.0, 2); + assert_eq!(padding_entry.2, 0); + assert!(address_transactions_iterator.next().is_none()); assert_eq!( transaction_status_index_cf.get(0).unwrap().unwrap(), TransactionStatusIndexMeta { @@ -5636,6 +5794,7 @@ pub mod tests { let blockstore_path = get_tmp_ledger_path!(); { let blockstore = Blockstore::open(&blockstore_path).unwrap(); + // TransactionStatus column opens initialized with one entry at index 2 let transaction_status_cf = blockstore.db.column::(); let pre_balances_vec = vec![1, 2, 3]; @@ -5726,7 +5885,7 @@ pub mod tests { .get_transaction_status_with_counter(signature5) .unwrap(); assert_eq!(status, None); - assert_eq!(counter, 5); + assert_eq!(counter, 6); // Signature does not exist, smaller than existing entries let (status, counter) = blockstore @@ -5747,7 +5906,7 @@ pub mod tests { .get_transaction_status_with_counter(signature6) .unwrap(); assert_eq!(status, None); - assert_eq!(counter, 1); + assert_eq!(counter, 2); } Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction"); } diff --git a/ledger/src/blockstore_db.rs b/ledger/src/blockstore_db.rs index bc825a8dcd..90bf4e7746 100644 --- a/ledger/src/blockstore_db.rs +++ b/ledger/src/blockstore_db.rs @@ -10,7 +10,7 @@ use rocksdb::{ }; use serde::de::DeserializeOwned; use serde::Serialize; -use solana_sdk::{clock::Slot, signature::Signature}; +use solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Signature}; use solana_transaction_status::{Rewards, TransactionStatusMeta}; use std::{collections::HashMap, fs, marker::PhantomData, path::Path, sync::Arc}; use thiserror::Error; @@ -38,7 +38,9 @@ const DATA_SHRED_CF: &str = "data_shred"; const CODE_SHRED_CF: &str = "code_shred"; /// Column family for Transaction Status const TRANSACTION_STATUS_CF: &str = "transaction_status"; -/// Column family for Transaction Status +/// Column family for Address Signatures +const ADDRESS_SIGNATURES_CF: &str = "address_signatures"; +/// Column family for Transaction Status Index const TRANSACTION_STATUS_INDEX_CF: &str = "transaction_status_index"; /// Column family for Rewards const REWARDS_CF: &str = "rewards"; @@ -110,6 +112,10 @@ pub mod columns { /// The transaction status column pub struct TransactionStatus; + #[derive(Debug)] + /// The address signatures column + pub struct AddressSignatures; + #[derive(Debug)] /// The transaction status index column pub struct TransactionStatusIndex; @@ -125,8 +131,8 @@ struct Rocks(rocksdb::DB); impl Rocks { fn open(path: &Path) -> Result { use columns::{ - DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards, Root, ShredCode, - ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex, + AddressSignatures, DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards, + Root, ShredCode, ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex, }; fs::create_dir_all(&path)?; @@ -151,6 +157,8 @@ impl Rocks { ColumnFamilyDescriptor::new(ShredCode::NAME, get_cf_options()); let transaction_status_cf_descriptor = ColumnFamilyDescriptor::new(TransactionStatus::NAME, get_cf_options()); + let address_signatures_cf_descriptor = + ColumnFamilyDescriptor::new(AddressSignatures::NAME, get_cf_options()); let transaction_status_index_cf_descriptor = ColumnFamilyDescriptor::new(TransactionStatusIndex::NAME, get_cf_options()); let rewards_cf_descriptor = ColumnFamilyDescriptor::new(Rewards::NAME, get_cf_options()); @@ -166,6 +174,7 @@ impl Rocks { shred_data_cf_descriptor, shred_code_cf_descriptor, transaction_status_cf_descriptor, + address_signatures_cf_descriptor, transaction_status_index_cf_descriptor, rewards_cf_descriptor, ]; @@ -178,8 +187,8 @@ impl Rocks { fn columns(&self) -> Vec<&'static str> { use columns::{ - DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards, Root, ShredCode, - ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex, + AddressSignatures, DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards, + Root, ShredCode, ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex, }; vec![ @@ -193,6 +202,7 @@ impl Rocks { ShredData::NAME, ShredCode::NAME, TransactionStatus::NAME, + AddressSignatures::NAME, TransactionStatusIndex::NAME, Rewards::NAME, ] @@ -282,6 +292,10 @@ impl TypedColumn for columns::TransactionStatus { type Type = TransactionStatusMeta; } +impl TypedColumn for columns::AddressSignatures { + type Type = blockstore_meta::AddressSignatureMeta; +} + impl TypedColumn for columns::TransactionStatusIndex { type Type = blockstore_meta::TransactionStatusIndexMeta; } @@ -314,7 +328,7 @@ impl Column for columns::TransactionStatus { type Index = (u64, Signature, Slot); fn key((index, signature, slot): (u64, Signature, Slot)) -> Vec { - let mut key = vec![0; 8 + 8 + 64]; + let mut key = vec![0; 8 + 64 + 8]; // size_of u64 + size_of Signature + size_of Slot BigEndian::write_u64(&mut key[0..8], index); key[8..72].clone_from_slice(&signature.as_ref()[0..64]); BigEndian::write_u64(&mut key[72..80], slot); @@ -341,6 +355,37 @@ impl ColumnName for columns::TransactionStatus { const NAME: &'static str = TRANSACTION_STATUS_CF; } +impl Column for columns::AddressSignatures { + type Index = (u64, Pubkey, Slot); + + fn key((index, pubkey, slot): (u64, Pubkey, Slot)) -> Vec { + let mut key = vec![0; 8 + 32 + 8]; // size_of u64 + size_of Pubkey + size_of Slot + BigEndian::write_u64(&mut key[0..8], index); + key[8..40].clone_from_slice(&pubkey.as_ref()[0..32]); + BigEndian::write_u64(&mut key[40..48], slot); + key + } + + fn index(key: &[u8]) -> (u64, Pubkey, Slot) { + let index = BigEndian::read_u64(&key[0..8]); + let pubkey = Pubkey::new(&key[8..40]); + let slot = BigEndian::read_u64(&key[40..48]); + (index, pubkey, slot) + } + + fn primary_index(index: Self::Index) -> u64 { + index.0 + } + + fn as_index(index: u64) -> Self::Index { + (index, Pubkey::default(), 0) + } +} + +impl ColumnName for columns::AddressSignatures { + const NAME: &'static str = ADDRESS_SIGNATURES_CF; +} + impl Column for columns::TransactionStatusIndex { type Index = u64; diff --git a/ledger/src/blockstore_meta.rs b/ledger/src/blockstore_meta.rs index 5a98beb003..eb6599dd11 100644 --- a/ledger/src/blockstore_meta.rs +++ b/ledger/src/blockstore_meta.rs @@ -1,6 +1,6 @@ use crate::erasure::ErasureConfig; use serde::{Deserialize, Serialize}; -use solana_sdk::clock::Slot; +use solana_sdk::{clock::Slot, signature::Signature}; use std::{collections::BTreeSet, ops::RangeBounds}; #[derive(Clone, Debug, Default, Deserialize, Serialize, Eq, PartialEq)] @@ -228,6 +228,12 @@ pub struct TransactionStatusIndexMeta { pub frozen: bool, } +#[derive(Debug, Default, Deserialize, Serialize, PartialEq)] +pub struct AddressSignatureMeta { + pub signature: Signature, + pub writeable: bool, +} + #[cfg(test)] mod test { use super::*;