TieredStorage struct (2/N) -- write_accounts part 1 (#32541)

#### Summary of Changes
This PR initiates the implementation of TieredAccountsFile::write_accounts,
which will later support AccountsFile::append_accounts.

This PR also introduces TieredAccountsFileWriter -- the main writer struct for a
TieredAccountsFile.

#### Test Plan
Extended existing unit tests.

runtime/src/tiered_storage.rs

@@ -9,12 +9,23 @@ pub mod index;
 pub mod meta;
 pub mod mmap_utils;
 pub mod readable;
+pub mod writer;
 
 use {
+    crate::{
+        account_storage::meta::{StorableAccountsWithHashesAndWriteVersions, StoredAccountInfo},
+        storable_accounts::StorableAccounts,
+    },
     error::TieredStorageError,
     footer::{AccountBlockFormat, AccountMetaFormat, OwnersBlockFormat},
     index::AccountIndexFormat,
-    std::path::{Path, PathBuf},
+    solana_sdk::{account::ReadableAccount, hash::Hash},
+    std::{
+        borrow::Borrow,
+        fs::OpenOptions,
+        path::{Path, PathBuf},
+    },
+    writer::TieredStorageWriter,
 };
 
 pub type TieredStorageResult<T> = Result<T, TieredStorageError>;
@@ -40,7 +51,7 @@ impl TieredStorage {
     /// Creates a new writable instance of TieredStorage based on the
     /// specified path and TieredStorageFormat.
     ///
-    /// Note that the actual file will not be created until append_accounts
+    /// Note that the actual file will not be created until write_accounts
     /// is called.
     pub fn new_writable(path: impl Into<PathBuf>, format: TieredStorageFormat) -> Self {
         Self {
@@ -53,17 +64,76 @@ impl TieredStorage {
     pub fn path(&self) -> &Path {
         self.path.as_path()
     }
+
+    /// Writes the specified accounts into this TieredStorage.
+    pub fn write_accounts<
+        'a,
+        'b,
+        T: ReadableAccount + Sync,
+        U: StorableAccounts<'a, T>,
+        V: Borrow<Hash>,
+    >(
+        &self,
+        accounts: &StorableAccountsWithHashesAndWriteVersions<'a, 'b, T, U, V>,
+        skip: usize,
+    ) -> TieredStorageResult<Vec<StoredAccountInfo>> {
+        // self.format must be Some as write_accounts can only be called on a
+        // TieredStorage instance created via new_writable() where it format
+        // field is required.
+        assert!(self.format.is_some());
+        let writer = TieredStorageWriter::new(&self.path, self.format.as_ref().unwrap())?;
+        writer.write_accounts(accounts, skip)
+    }
+
+    /// Returns the size of the underlying accounts file.
+    pub fn file_size(&self) -> TieredStorageResult<u64> {
+        let file = OpenOptions::new().read(true).open(&self.path);
+
+        Ok(file
+            .and_then(|file| file.metadata())
+            .map(|metadata| metadata.len())
+            .unwrap_or(0))
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use {super::*, hot::HOT_FORMAT};
+    use {
+        super::*,
+        crate::account_storage::meta::StoredMetaWriteVersion,
+        footer::{TieredStorageFooter, TieredStorageMagicNumber},
+        hot::HOT_FORMAT,
+        solana_sdk::{account::AccountSharedData, clock::Slot, pubkey::Pubkey},
+        tempfile::NamedTempFile,
+    };
 
     #[test]
-    fn test_new_writable() {
-        let path = PathBuf::default();
-        let ts = TieredStorage::new_writable(&path, HOT_FORMAT.clone());
+    fn test_new_footer_only() {
+        let path = NamedTempFile::new().unwrap().path().to_path_buf();
 
-        assert_eq!(ts.path(), path);
+        let tiered_storage = TieredStorage::new_writable(path.clone(), HOT_FORMAT.clone());
+        assert_eq!(tiered_storage.path(), path);
+        assert_eq!(tiered_storage.file_size().unwrap(), 0);
+
+        let slot_ignored = Slot::MAX;
+        let account_refs = Vec::<(&Pubkey, &AccountSharedData)>::new();
+        let account_data = (slot_ignored, account_refs.as_slice());
+        let storable_accounts =
+            StorableAccountsWithHashesAndWriteVersions::new_with_hashes_and_write_versions(
+                &account_data,
+                Vec::<&Hash>::new(),
+                Vec::<StoredMetaWriteVersion>::new(),
+            );
+
+        let result = tiered_storage.write_accounts(&storable_accounts, 0);
+        // Expect the result to be TieredStorageError::Unsupported as the feature
+        // is not yet fully supported, but we can still check its partial results
+        // in the test.
+        assert!(result.is_err());
+        assert_eq!(
+            tiered_storage.file_size().unwrap() as usize,
+            std::mem::size_of::<TieredStorageFooter>()
+                + std::mem::size_of::<TieredStorageMagicNumber>()
+        );
     }
 }

runtime/src/tiered_storage/error.rs

@@ -1,4 +1,4 @@
-use thiserror::Error;
+use {std::path::PathBuf, thiserror::Error};
 
 #[derive(Error, Debug)]
 pub enum TieredStorageError {
@@ -7,4 +7,13 @@ pub enum TieredStorageError {
 
     #[error("MagicNumberMismatch: expected {0}, found {1}")]
     MagicNumberMismatch(u64, u64),
+
+    #[error("ReadOnlyFileUpdateError: attempted to update read-only file {0}")]
+    ReadOnlyFileUpdateError(PathBuf),
+
+    #[error("UnknownFormat: the tiered storage format is unavailable for file {0}")]
+    UnknownFormat(PathBuf),
+
+    #[error("Unsupported: the feature is not yet supported")]
+    Unsupported(),
 }

runtime/src/tiered_storage/file.rs

@@ -25,20 +25,13 @@ impl TieredStorageFile {
         )
     }
 
-    pub fn new_writable(file_path: impl AsRef<Path>) -> Self {
-        Self(
+    pub fn new_writable(file_path: impl AsRef<Path>) -> Result<Self, std::io::Error> {
+        Ok(Self(
             OpenOptions::new()
+                .create_new(true)
                 .write(true)
-                .create(true)
-                .open(&file_path)
-                .unwrap_or_else(|e| {
-                    panic!(
-                        "[TieredStorageError] Unable to create {:?} as writable: {:?}",
-                        file_path.as_ref().display(),
-                        e,
-                    );
-                }),
-        )
+                .open(file_path)?,
+        ))
     }
 
     pub fn write_type<T>(&self, value: &T) -> Result<usize, std::io::Error> {

runtime/src/tiered_storage/footer.rs

@@ -258,7 +258,7 @@ mod tests {
 
         // Persist the expected footer.
         {
-            let file = TieredStorageFile::new_writable(&path.path);
+            let file = TieredStorageFile::new_writable(&path.path).unwrap();
             expected_footer.write_footer_block(&file).unwrap();
         }
 

runtime/src/tiered_storage/index.rs

@@ -133,7 +133,7 @@ mod tests {
             .collect();
 
         {
-            let file = TieredStorageFile::new_writable(&path);
+            let file = TieredStorageFile::new_writable(&path).unwrap();
             let indexer = AccountIndexFormat::AddressAndOffset;
             indexer.write_index_block(&file, &index_entries).unwrap();
         }

runtime/src/tiered_storage/writer.rs

@@ -0,0 +1,56 @@
+//! docs/src/proposals/append-vec-storage.md
+
+use {
+    crate::{
+        account_storage::meta::{StorableAccountsWithHashesAndWriteVersions, StoredAccountInfo},
+        storable_accounts::StorableAccounts,
+        tiered_storage::{
+            error::TieredStorageError, file::TieredStorageFile, footer::TieredStorageFooter,
+            TieredStorageFormat, TieredStorageResult,
+        },
+    },
+    solana_sdk::{account::ReadableAccount, hash::Hash},
+    std::{borrow::Borrow, path::Path},
+};
+
+#[derive(Debug)]
+pub struct TieredStorageWriter<'format> {
+    storage: TieredStorageFile,
+    format: &'format TieredStorageFormat,
+}
+
+impl<'format> TieredStorageWriter<'format> {
+    pub fn new(
+        file_path: impl AsRef<Path>,
+        format: &'format TieredStorageFormat,
+    ) -> TieredStorageResult<Self> {
+        Ok(Self {
+            storage: TieredStorageFile::new_writable(file_path)?,
+            format,
+        })
+    }
+
+    pub fn write_accounts<
+        'a,
+        'b,
+        T: ReadableAccount + Sync,
+        U: StorableAccounts<'a, T>,
+        V: Borrow<Hash>,
+    >(
+        &self,
+        _accounts: &StorableAccountsWithHashesAndWriteVersions<'a, 'b, T, U, V>,
+        _skip: usize,
+    ) -> TieredStorageResult<Vec<StoredAccountInfo>> {
+        let footer = TieredStorageFooter {
+            account_meta_format: self.format.account_meta_format,
+            owners_block_format: self.format.owners_block_format,
+            account_block_format: self.format.account_block_format,
+            account_index_format: self.format.account_index_format,
+            ..TieredStorageFooter::default()
+        };
+
+        footer.write_footer_block(&self.storage)?;
+
+        Err(TieredStorageError::Unsupported())
+    }
+}