test: add rnd persist unit test

Cargo.toml

@@ -132,6 +132,8 @@ rust-s3 = "0.34.0"
 function_name = "0.3.0"
 serial_test = "2.0.0"
 light-merkle-tree-reference = "2.0.0"
+light-indexed-merkle-tree = "2.1.0"
+light-hasher = "3.1.0"
 
 [profile.dev]
 # Do not produce debug info for ~40% faster incremental compilation.

src/ingester/error.rs

@@ -14,6 +14,8 @@ pub enum IngesterError {
     EmptyBatchEvent,
     #[error("Invalid event.")]
     InvalidEvent,
+    #[error("Low element not found.")]
+    LowElementNotFound,
 }
 
 impl From<sea_orm::error::DbErr> for IngesterError {

src/ingester/parser/merkle_tree_events_parser.rs

@@ -130,7 +130,7 @@ fn parse_indexed_merkle_tree_update(
         {
             let indexed_tree_leaf_update = IndexedTreeLeafUpdate {
                 tree: tree_pubkey,
-                tree_type: tree_type.clone(),
+                tree_type,
                 hash: *hash,
                 leaf: *leaf,
                 seq,

src/ingester/parser/state_update.rs

@@ -86,14 +86,17 @@ impl From<NewAddress> for AddressQueueUpdate {
 #[derive(Default, Debug, Clone, PartialEq, Eq)]
 /// Representation of state update of the compression system that is optimal for simple persistence.
 pub struct StateUpdate {
+    // v1 and v2 tree accounts
     pub in_accounts: HashSet<Hash>,
+    // v1 and v2 tree accounts
     pub out_accounts: Vec<AccountWithContext>,
     pub account_transactions: HashSet<AccountTransaction>,
     pub transactions: HashSet<Transaction>,
     pub leaf_nullifications: HashSet<LeafNullification>,
     pub indexed_merkle_tree_updates: HashMap<(Pubkey, u64), IndexedTreeLeafUpdate>,
-
+    // v2 state and address Merkle tree updates
     pub batch_merkle_tree_events: BatchMerkleTreeEvents,
+    // v2 input accounts that are inserted into the input queue
     pub batch_nullify_context: Vec<BatchNullifyContext>,
     pub batch_new_addresses: Vec<AddressQueueUpdate>,
 }

src/ingester/persist/mod.rs

@@ -65,7 +65,9 @@ pub async fn persist_state_update(
         return Ok(());
     }
     let StateUpdate {
+        // input accounts that will be nullified
         in_accounts,
+        // Output accounts that will be created
         out_accounts,
         account_transactions,
         transactions,
@@ -433,7 +435,7 @@ async fn append_output_accounts(
             slot_created: Set(account.account.slot_created.0 as i64),
             seq: Set(account.account.seq.map(|x| x.0 as i64)),
             prev_spent: Set(None),
-            tx_hash: Default::default(), // Its sets at input queue insertion for batch updates
+            tx_hash: Set(account.context.tx_hash.as_ref().map(|x| x.to_vec())),
         });
 
         if let Some(token_data) = parse_token_data(&account.account)? {