[TieredStorage] Write owners block for HotAccountStorage (solana-labs…

…#34927)

#### Problem
So far the current HotStorageWriter::write_accounts() only writes
accounts blocks and index block.

#### Summary of Changes
The PR further writes owners block in HotStorageWriter::write_accounts().

#### Test Plan
Extended existing test for HotStorageWriter to cover the owners block.

accounts-db/src/tiered_storage/hot.rs

@@ -14,7 +14,7 @@ use {
             index::{AccountIndexWriterEntry, AccountOffset, IndexBlockFormat, IndexOffset},
             meta::{AccountMetaFlags, AccountMetaOptionalFields, TieredAccountMeta},
             mmap_utils::{get_pod, get_slice},
-            owners::{OwnerOffset, OwnersBlockFormat},
+            owners::{OwnerOffset, OwnersBlockFormat, OwnersTable, OWNER_NO_OWNER},
             readable::TieredReadableAccount,
             StorableAccounts, StorableAccountsWithHashesAndWriteVersions, TieredStorageError,
             TieredStorageFormat, TieredStorageResult,
@@ -488,6 +488,7 @@ impl HotStorageWriter {
     fn write_account(
         &self,
         lamports: u64,
+        owner_offset: OwnerOffset,
         account_data: &[u8],
         executable: bool,
         rent_epoch: Option<Epoch>,
@@ -504,6 +505,7 @@ impl HotStorageWriter {
         let padding_len = padding_bytes(account_data.len());
         let meta = HotAccountMeta::new()
             .with_lamports(lamports)
+            .with_owner_offset(owner_offset)
             .with_account_data_size(account_data.len() as u64)
             .with_account_data_padding(padding_len)
             .with_flags(&flags);
@@ -520,8 +522,9 @@ impl HotStorageWriter {
         Ok(stored_size)
     }
 
-    /// A work-in-progress function that will eventually implements
-    /// AccountsFile::appends_account()
+    /// Persists `accounts` into the underlying hot accounts file associated
+    /// with this HotStorageWriter.  The first `skip` number of accounts are
+    /// *not* persisted.
     pub fn write_accounts<
         'a,
         'b,
@@ -535,6 +538,7 @@ impl HotStorageWriter {
     ) -> TieredStorageResult<()> {
         let mut footer = new_hot_footer();
         let mut index = vec![];
+        let mut owners_table = OwnersTable::default();
         let mut cursor = 0;
 
         // writing accounts blocks
@@ -548,20 +552,28 @@ impl HotStorageWriter {
 
             // Obtain necessary fields from the account, or default fields
             // for a zero-lamport account in the None case.
-            let (lamports, data, executable, rent_epoch, account_hash) = account
+            let (lamports, owner, data, executable, rent_epoch, account_hash) = account
                 .map(|acc| {
                     (
                         acc.lamports(),
+                        acc.owner(),
                         acc.data(),
                         acc.executable(),
                         // only persist rent_epoch for those non-rent-exempt accounts
                         (acc.rent_epoch() != RENT_EXEMPT_RENT_EPOCH).then_some(acc.rent_epoch()),
                         Some(*account_hash),
                     )
                 })
-                .unwrap_or((0, &[], false, None, None));
-
-            cursor += self.write_account(lamports, data, executable, rent_epoch, account_hash)?;
+                .unwrap_or((0, &OWNER_NO_OWNER, &[], false, None, None));
+            let owner_offset = owners_table.insert(owner);
+            cursor += self.write_account(
+                lamports,
+                owner_offset,
+                data,
+                executable,
+                rent_epoch,
+                account_hash,
+            )?;
             index.push(index_entry);
         }
         footer.account_entry_count = (len - skip) as u32;
@@ -581,11 +593,13 @@ impl HotStorageWriter {
             cursor += self.storage.write_pod(&0u32)?;
         }
 
-        // TODO: owner block will be implemented in the follow-up PRs
-        // expect the offset of each block aligned.
+        // writing owners block
         assert!(cursor % HOT_BLOCK_ALIGNMENT == 0);
         footer.owners_block_offset = cursor as u64;
-        footer.owner_count = 0;
+        footer.owner_count = owners_table.len() as u32;
+        footer
+            .owners_block_format
+            .write_owners_block(&self.storage, &owners_table)?;
 
         footer.write_footer_block(&self.storage)?;
 
@@ -1230,12 +1244,17 @@ pub mod tests {
     /// Create a test account based on the specified seed.
     /// The created test account might have default rent_epoch
     /// and write_version.
+    ///
+    /// When the seed is zero, then a zero-lamport test account will be
+    /// created.
     fn create_test_account(seed: u64) -> (StoredMeta, AccountSharedData) {
         let data_byte = seed as u8;
+        let owner_byte = u8::MAX - data_byte;
         let account = Account {
-            lamports: seed + 1,
+            lamports: seed,
             data: std::iter::repeat(data_byte).take(seed as usize).collect(),
-            owner: Pubkey::new_unique(),
+            // this will allow some test account sharing the same owner.
+            owner: [owner_byte; 32].into(),
             executable: seed % 2 > 0,
             rent_epoch: if seed % 3 > 0 {
                 seed
@@ -1304,15 +1323,30 @@ pub mod tests {
                 .unwrap()
                 .unwrap();
 
-            let (account, address, hash, _write_version) = storable_accounts.get(i);
-            let account = account.unwrap();
+            let (account, address, account_hash, _write_version) = storable_accounts.get(i);
+            let (lamports, owner, data, executable, account_hash) = account
+                .map(|acc| {
+                    (
+                        acc.lamports(),
+                        acc.owner(),
+                        acc.data(),
+                        acc.executable(),
+                        // only persist rent_epoch for those non-rent-exempt accounts
+                        Some(*account_hash),
+                    )
+                })
+                .unwrap_or((0, &OWNER_NO_OWNER, &[], false, None));
 
-            assert_eq!(stored_meta.lamports(), account.lamports());
-            assert_eq!(stored_meta.data().len(), account.data().len());
-            assert_eq!(stored_meta.data(), account.data());
-            assert_eq!(stored_meta.executable(), account.executable());
+            assert_eq!(stored_meta.lamports(), lamports);
+            assert_eq!(stored_meta.data().len(), data.len());
+            assert_eq!(stored_meta.data(), data);
+            assert_eq!(stored_meta.executable(), executable);
+            assert_eq!(stored_meta.owner(), owner);
             assert_eq!(stored_meta.pubkey(), address);
-            assert_eq!(stored_meta.hash(), hash);
+            assert_eq!(
+                *stored_meta.hash(),
+                account_hash.unwrap_or(AccountHash(Hash::default()))
+            );
 
             assert_eq!(i + 1, next);
         }

accounts-db/src/tiered_storage/owners.rs

@@ -16,6 +16,10 @@ use {
 #[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd)]
 pub struct OwnerOffset(pub u32);
 
+lazy_static! {
+    pub static ref OWNER_NO_OWNER: Pubkey = Pubkey::default();
+}
+
 /// Owner block holds a set of unique addresses of account owners,
 /// and an account meta has a owner_offset field for accessing
 /// it's owner address.
@@ -97,6 +101,16 @@ impl<'a> OwnersTable<'a> {
 
         OwnerOffset(offset as u32)
     }
+
+    /// Returns the number of unique owner addresses in the table.
+    pub fn len(&self) -> usize {
+        self.owners_set.len()
+    }
+
+    /// Returns true if the OwnersTable is empty
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
 }
 
 #[cfg(test)]