feat(etl): direct snapshot into the account backfill stream

README.md

@@ -30,6 +30,10 @@ If you are building the Metaplex RPC API infrastructure please follow the instru
 
 If you are using this plugin for your bespoke use case then the build steps are below.
 
+#### A note on formatting
+
+Since `rustfmt.toml` uses unstable configuration options, it is required to run formatting with the nightly toolchain: `cargo +nightly fmt`.
+
 ### Building Locally
 
 #### Linux
@@ -80,7 +84,7 @@ The process running the validator must have access to environment variables. Tho
 
 ```bash
 RUST_LOG=warn
-PLUGIN_MESSENGER_CONFIG='{ messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }'
+PLUGIN_MESSENGER_CONFIG={ messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }
 ```
 
 The PLUGIN_MESSENGER_CONFIG determines which compiled messenger to select and a specific configuration for the messenger.
@@ -101,11 +105,11 @@ The PLUGIN_MESSENGER_CONFIG determines which compiled messenger to select and a
 ```
 Lower Scale Low network latency
 
-PLUGIN_MESSENGER_CONFIG='{pipeline_size_bytes=1000000,local_buffer_max_window=10, messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }'
+PLUGIN_MESSENGER_CONFIG={pipeline_size_bytes=1000000,local_buffer_max_window=10, messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }
 
 High Scale Higher latency
 
-PLUGIN_MESSENGER_CONFIG='{pipeline_size_bytes=50000000,local_buffer_max_window=500, messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }'
+PLUGIN_MESSENGER_CONFIG={pipeline_size_bytes=50000000,local_buffer_max_window=500, messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }
 
 
 ```
@@ -120,7 +124,7 @@ PLUGIN_MESSENGER_CONFIG='{pipeline_size_bytes=50000000,local_buffer_max_window=5
 
 ```
 
-PLUGIN_MESSENGER_CONFIG='{batch_size=1000, message_wait_timeout=5, retries=5, consumer_id="random_string",messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }'
+PLUGIN_MESSENGER_CONFIG={batch_size=1000, message_wait_timeout=5, retries=5, consumer_id="random_string",messenger_type="Redis", connection_config={ redis_connection_str="redis://redis" } }
 PLUGIN_ACCOUNT_STREAM_SIZE=250000000
 PLUGIN_SLOT_STREAM_SIZE=250000
 PLUGIN_TRANSACTION_STREAM_SIZE=25000000

plerkle/src/lib.rs

@@ -1,6 +1,6 @@
 pub mod accounts_selector;
+pub mod config;
 pub mod error;
 pub mod geyser_plugin_nft;
-pub mod config;
 pub mod metrics;
 pub mod transaction_selector;

plerkle_messenger/src/lib.rs

@@ -5,4 +5,5 @@ mod plerkle_messenger;
 pub mod redis;
 pub use redis::*;
 
-pub use {crate::error::*, plerkle_messenger::*};
+pub use crate::error::*;
+pub use plerkle_messenger::*;

plerkle_messenger/src/plerkle_messenger.rs

@@ -88,12 +88,12 @@ pub async fn select_messenger_read(
     config: MessengerConfig,
 ) -> Result<Box<dyn Messenger>, MessengerError> {
     match config.messenger_type {
-        MessengerType::Redis => {
-            RedisMessenger::new(config).await.map(|a| Box::new(a) as Box<dyn Messenger>)
-        }
+        MessengerType::Redis => RedisMessenger::new(config)
+            .await
+            .map(|a| Box::new(a) as Box<dyn Messenger>),
         _ => Err(MessengerError::ConfigurationError {
-            msg: "This Messenger type is not valid or not unimplemented.".to_string()
-        })
+            msg: "This Messenger type is not valid or not unimplemented.".to_string(),
+        }),
     }
 }
 

plerkle_messenger/src/redis/redis_messenger.rs

@@ -368,11 +368,10 @@ impl MessageStreamer for RedisMessenger {
         );
 
         // Add stream to Redis.
-        let result: RedisResult<()> = self.connection.xgroup_create_mkstream(
-            stream_key,
-            self.consumer_group_name.as_str(),
-            "$",
-        ).await;
+        let result: RedisResult<()> = self
+            .connection
+            .xgroup_create_mkstream(stream_key, self.consumer_group_name.as_str(), "$")
+            .await;
 
         if let Err(e) = result {
             info!("Group already exists: {:?}", e)
@@ -424,7 +423,8 @@ impl MessageStreamer for RedisMessenger {
             for bytes in stream.local_buffer.iter() {
                 pipe.xadd_maxlen(stream_key, maxlen, "*", &[(DATA_KEY, &bytes)]);
             }
-            let result: Result<Vec<String>, redis::RedisError> = pipe.query_async(&mut self.connection).await;
+            let result: Result<Vec<String>, redis::RedisError> =
+                pipe.query_async(&mut self.connection).await;
             if let Err(e) = result {
                 error!("Redis send error: {e}");
                 return Err(MessengerError::SendError { msg: e.to_string() });

plerkle_serialization/src/error.rs

@@ -1,6 +1,6 @@
 use thiserror::Error;
-#[derive(Debug, Clone, PartialEq, Eq,Error)]
+#[derive(Debug, Clone, PartialEq, Eq, Error)]
 pub enum PlerkleSerializationError {
     #[error("Serialization error: {0}")]
-    SerializationError(String)
-}
+    SerializationError(String),
+}

plerkle_serialization/src/lib.rs

@@ -1,14 +1,20 @@
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod account_info_generated;
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod block_info_generated;
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod common_generated;
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod compiled_instruction_generated;
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod slot_status_info_generated;
 #[allow(unused_imports)]
+#[rustfmt::skip]
 mod transaction_info_generated;
 
 pub mod deserializer;

plerkle_serialization/src/serializer/mod.rs

@@ -1,3 +1,3 @@
 mod serializer_common;
 mod serializer_stable;
-pub use serializer_stable::*;
+pub use serializer_stable::*;

plerkle_serialization/src/serializer/serializer_stable.rs

@@ -1,21 +1,23 @@
-use crate::error::PlerkleSerializationError;
-use crate::solana_geyser_plugin_interface_shims::{
-    ReplicaAccountInfoV2, ReplicaBlockInfoV2, ReplicaTransactionInfoV2, SlotStatus,
-};
 use crate::{
+    error::PlerkleSerializationError,
+    solana_geyser_plugin_interface_shims::{
+        ReplicaAccountInfoV2, ReplicaBlockInfoV2, ReplicaTransactionInfoV2, SlotStatus,
+    },
     AccountInfo, AccountInfoArgs, BlockInfo, BlockInfoArgs, CompiledInnerInstruction,
     CompiledInnerInstructionArgs, CompiledInnerInstructions, CompiledInnerInstructionsArgs,
-    CompiledInstruction, CompiledInstructionArgs,
-    Pubkey as FBPubkey, Pubkey, SlotStatusInfo, SlotStatusInfoArgs, Status as FBSlotStatus,
-    TransactionInfo, TransactionInfoArgs, TransactionVersion,
+    CompiledInstruction, CompiledInstructionArgs, Pubkey as FBPubkey, Pubkey, SlotStatusInfo,
+    SlotStatusInfoArgs, Status as FBSlotStatus, TransactionInfo, TransactionInfoArgs,
+    TransactionVersion,
 };
 use chrono::Utc;
 use flatbuffers::{FlatBufferBuilder, WIPOffset};
-use solana_sdk::message::{SanitizedMessage, VersionedMessage};
-use solana_sdk::transaction::VersionedTransaction;
-use solana_transaction_status::option_serializer::OptionSerializer;
+use solana_sdk::{
+    message::{SanitizedMessage, VersionedMessage},
+    transaction::VersionedTransaction,
+};
 use solana_transaction_status::{
-    EncodedConfirmedTransactionWithStatusMeta, UiInstruction, UiTransactionStatusMeta,
+    option_serializer::OptionSerializer, EncodedConfirmedTransactionWithStatusMeta, UiInstruction,
+    UiTransactionStatusMeta,
 };
 
 pub fn serialize_account<'a>(

plerkle_snapshot/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 # Renamed from original "solana-snapshot-etl"
 name = "plerkle_snapshot"
-version = "0.7.0"
+version = "0.8.0"
 edition = "2021"
 license = "Apache-2.0"
 documentation = "https://docs.rs/solana-snapshot-etl"

plerkle_snapshot/src/append_vec.rs

@@ -15,23 +15,22 @@
 // This file contains code vendored from https://github.com/solana-labs/solana
 // Source: solana/runtime/src/append_vec.rs
 
-use {
-    log::*,
-    memmap2::{Mmap, MmapMut},
-    serde::{Deserialize, Serialize},
-    solana_sdk::{
-        account::{Account, AccountSharedData, ReadableAccount},
-        clock::Epoch,
-        hash::Hash,
-        pubkey::Pubkey,
-    },
-    std::{
-        convert::TryFrom,
-        fs::OpenOptions,
-        io::{self, Read},
-        mem,
-        path::Path,
-    },
+use std::{
+    convert::TryFrom,
+    fs::OpenOptions,
+    io::{self, Read},
+    mem,
+    path::Path,
+};
+
+use log::*;
+use memmap2::{Mmap, MmapMut};
+use serde::{Deserialize, Serialize};
+use solana_sdk::{
+    account::{Account, AccountSharedData, ReadableAccount},
+    clock::Epoch,
+    hash::Hash,
+    pubkey::Pubkey,
 };
 
 // Data placement should be aligned at the next boundary. Without alignment accessing the memory may
@@ -186,11 +185,7 @@ impl AppendVec {
         current_len: usize,
         slot: u64,
     ) -> io::Result<Self> {
-        let data = OpenOptions::new()
-            .read(true)
-            .write(false)
-            .create(false)
-            .open(&path)?;
+        let data = OpenOptions::new().read(true).write(false).create(false).open(&path)?;
 
         let file_size = std::fs::metadata(&path)?.len();
         AppendVec::sanitize_len_and_size(current_len, file_size as usize)?;
@@ -204,12 +199,7 @@ impl AppendVec {
             result?
         };
 
-        let new = AppendVec {
-            map,
-            current_len,
-            file_size,
-            slot,
-        };
+        let new = AppendVec { map, current_len, file_size, slot };
 
         Ok(new)
     }
@@ -269,17 +259,7 @@ impl AppendVec {
         let (hash, next): (&'a Hash, _) = self.get_type(next)?;
         let (data, next) = self.get_slice(next, meta.data_len as usize)?;
         let stored_size = next - offset;
-        Some((
-            StoredAccountMeta {
-                meta,
-                account_meta,
-                data,
-                offset,
-                stored_size,
-                hash,
-            },
-            next,
-        ))
+        Some((StoredAccountMeta { meta, account_meta, data, offset, stored_size, hash }, next))
     }
 
     pub fn get_slot(&self) -> u64 {

plerkle_snapshot/src/archived.rs

@@ -1,15 +1,19 @@
+use std::{
+    fs::File,
+    io::{BufReader, Read},
+    path::{Component, Path},
+    pin::Pin,
+    time::Instant,
+};
+
+use log::info;
+use tar::{Archive, Entries, Entry};
+
 use crate::{
     deserialize_from, parse_append_vec_name, AccountsDbFields, AppendVec, AppendVecIterator,
     DeserializableVersionedBank, Result, SerializableAccountStorageEntry, SnapshotError,
     SnapshotExtractor,
 };
-use log::info;
-use std::fs::File;
-use std::io::{BufReader, Read};
-use std::path::{Component, Path};
-use std::pin::Pin;
-use std::time::Instant;
-use tar::{Archive, Entries, Entry};
 
 /// Extracts account data from a .tar.zst stream.
 pub struct ArchiveSnapshotExtractor<Source>
@@ -72,10 +76,7 @@ where
         let accounts_db_fields_post_time = Instant::now();
         drop(snapshot_file);
 
-        info!(
-            "Read bank fields in {:?}",
-            versioned_bank_post_time - pre_unpack
-        );
+        info!("Read bank fields in {:?}", versioned_bank_post_time - pre_unpack);
         info!(
             "Read accounts DB fields in {:?}",
             accounts_db_fields_post_time - versioned_bank_post_time
@@ -89,22 +90,18 @@ where
     }
 
     fn unboxed_iter(&mut self) -> impl Iterator<Item = Result<AppendVec>> + '_ {
-        self.entries
-            .take()
-            .into_iter()
-            .flatten()
-            .filter_map(|entry| {
-                let mut entry = match entry {
-                    Ok(x) => x,
-                    Err(e) => return Some(Err(e.into())),
-                };
-                let path = match entry.path() {
-                    Ok(x) => x,
-                    Err(e) => return Some(Err(e.into())),
-                };
-                let (slot, id) = path.file_name().and_then(parse_append_vec_name)?;
-                Some(self.process_entry(&mut entry, slot, id))
-            })
+        self.entries.take().into_iter().flatten().filter_map(|entry| {
+            let mut entry = match entry {
+                Ok(x) => x,
+                Err(e) => return Some(Err(e.into())),
+            };
+            let path = match entry.path() {
+                Ok(x) => x,
+                Err(e) => return Some(Err(e.into())),
+            };
+            let (slot, id) = path.file_name().and_then(parse_append_vec_name)?;
+            Some(self.process_entry(&mut entry, slot, id))
+        })
     }
 
     fn process_entry(
@@ -113,22 +110,13 @@ where
         slot: u64,
         id: u64,
     ) -> Result<AppendVec> {
-        let known_vecs = self
-            .accounts_db_fields
-            .0
-            .get(&slot)
-            .map(|v| &v[..])
-            .unwrap_or(&[]);
+        let known_vecs = self.accounts_db_fields.0.get(&slot).map(|v| &v[..]).unwrap_or(&[]);
         let known_vec = known_vecs.iter().find(|entry| entry.id == (id as usize));
         let known_vec = match known_vec {
             None => return Err(SnapshotError::UnexpectedAppendVec),
             Some(v) => v,
         };
-        Ok(AppendVec::new_from_reader(
-            entry,
-            known_vec.accounts_current_len,
-            slot,
-        )?)
+        Ok(AppendVec::new_from_reader(entry, known_vec.accounts_current_len, slot)?)
     }
 
     fn is_snapshot_manifest_file(path: &Path) -> bool {
@@ -141,11 +129,7 @@ where
             _ => return false,
         };
         // Check if slot number file is valid u64.
-        if slot_number_str_1
-            .to_str()
-            .and_then(|s| s.parse::<u64>().ok())
-            .is_none()
-        {
+        if slot_number_str_1.to_str().and_then(|s| s.parse::<u64>().ok()).is_none() {
             return false;
         }
         let slot_number_str_2 = match components.next() {