Add blockstore address-to-signature index (#9367)

automerge

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,

ledger/src/blockstore.rs

@@ -87,6 +87,7 @@ pub struct Blockstore {
     data_shred_cf: LedgerColumn<cf::ShredData>,
     code_shred_cf: LedgerColumn<cf::ShredCode>,
     transaction_status_cf: LedgerColumn<cf::TransactionStatus>,
+    address_signatures_cf: LedgerColumn<cf::AddressSignatures>,
     transaction_status_index_cf: LedgerColumn<cf::TransactionStatusIndex>,
     active_transaction_status_index: RwLock<u64>,
     rewards_cf: LedgerColumn<cf::Rewards>,
@@ -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::<cf::TransactionStatusIndex>(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,9 +395,13 @@ impl Blockstore {
             &mut w_active_transaction_status_index,
             to_slot,
         )? {
-            columns_empty &= &self
+            columns_empty &= self
                 .db
                 .delete_range_cf::<cf::TransactionStatus>(&mut write_batch, index, index + 1)
+                .unwrap_or(false)
+                & self
+                    .db
+                    .delete_range_cf::<cf::AddressSignatures>(&mut write_batch, index, index + 1)
                     .unwrap_or(false);
         }
         let mut write_timer = Measure::start("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)> {
+    ) -> Result<u64> {
-        let (signature, slot) = index;
-        if self.transaction_status_index_cf.get(0)?.is_none() {
-            self.initialize_transaction_status_index()?;
-        }
         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
+        // This write lock prevents interleaving issues with the transaction_status_index_cf by gating
-        // gating writes to that column
+        // writes to that column
         let mut w_active_transaction_status_index =
             self.active_transaction_status_index.write().unwrap();
-        let index =
+        let primary_index = self.get_primary_index(slot, &mut w_active_transaction_status_index)?;
-            self.make_transaction_status_index(index, &mut w_active_transaction_status_index)?;
+        self.transaction_status_cf
-        self.transaction_status_cf.put(index, status)
+            .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::<cf::TransactionStatus>(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::<cf::AddressSignatures>(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::<cf::TransactionStatusIndex>();
             let slot0 = 10;
 
-            assert!(transaction_status_index_cf.get(0).unwrap().is_none());
+            // Primary index column is initialized on Blockstore::open
-            assert!(transaction_status_index_cf.get(1).unwrap().is_none());
+            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<u8> = (0..64).map(|_| rand::random::<u8>()).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::<cf::AddressSignatures>(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<u8> = (0..64).map(|_| rand::random::<u8>()).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::<cf::TransactionStatus>(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::<cf::AddressSignatures>(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::<cf::TransactionStatus>(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::<cf::AddressSignatures>(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::<cf::TransactionStatus>(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::<cf::AddressSignatures>(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<u8> = (0..64).map(|_| rand::random::<u8>()).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::<cf::AddressSignatures>(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::<cf::AddressSignatures>(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::<cf::AddressSignatures>(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::<cf::AddressSignatures>(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::<cf::TransactionStatus>();
 
             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");
     }

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<Rocks> {
         use columns::{
-            DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards, Root, ShredCode,
+            AddressSignatures, DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards,
-            ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex,
+            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,
+            AddressSignatures, DeadSlots, DuplicateSlots, ErasureMeta, Index, Orphans, Rewards,
-            ShredData, SlotMeta, TransactionStatus, TransactionStatusIndex,
+            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<u8> {
-        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<u8> {
+        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;
 

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::*;