fix: cleanup

Signed-off-by: William Hankins <[email protected]>

Author: whankinsiv

common/src/types.rs

@@ -186,15 +186,31 @@ impl PartialOrd for BlockInfo {
     }
 }
 
+// Individual transaction UTxO deltas
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct TxUTxODeltas {
+    // Transaction in which delta occured
+    pub tx_identifier: TxIdentifier,
+
+    // Created and spent UTxOs
+    pub inputs: Vec<UTxOIdentifier>,
+    pub outputs: Vec<TxOutput>,
+}
+
 /// Individual address balance change
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct AddressDelta {
+    // Address involved in delta
     pub address: Address,
+
+    // Transaction in which delta occured
     pub tx_identifier: TxIdentifier,
 
+    // Address impacted UTxOs
     pub spent_utxos: Vec<UTxOIdentifier>,
     pub created_utxos: Vec<UTxOIdentifier>,
 
+    // Sums of spent and created UTxOs
     pub sent: ValueDelta,
     pub received: Value,
 }
@@ -405,63 +421,6 @@ impl AddAssign for ValueMap {
     }
 }
 
-/// Hashmap representation of ValueDelta (lovelace + multiasset)
-pub struct ValueDeltaMap {
-    pub lovelace: i64,
-    pub assets: NativeAssetsDeltaMap,
-}
-
-impl From<ValueDelta> for ValueDeltaMap {
-    fn from(value: ValueDelta) -> Self {
-        let mut assets = HashMap::new();
-
-        for (policy, asset_list) in value.assets {
-            let policy_entry = assets.entry(policy).or_insert_with(HashMap::new);
-            for asset in asset_list {
-                *policy_entry.entry(asset.name).or_insert(0) += asset.amount;
-            }
-        }
-
-        ValueDeltaMap {
-            lovelace: value.lovelace,
-            assets,
-        }
-    }
-}
-
-impl AddAssign<ValueDelta> for ValueDeltaMap {
-    fn add_assign(&mut self, delta: ValueDelta) {
-        self.lovelace += delta.lovelace;
-
-        for (policy, assets) in delta.assets {
-            let policy_entry = self.assets.entry(policy).or_default();
-            for asset in assets {
-                *policy_entry.entry(asset.name).or_insert(0) += asset.amount;
-            }
-        }
-    }
-}
-
-impl From<ValueDeltaMap> for ValueDelta {
-    fn from(map: ValueDeltaMap) -> Self {
-        let mut assets_vec = Vec::with_capacity(map.assets.len());
-
-        for (policy, asset_map) in map.assets {
-            let inner_assets = asset_map
-                .into_iter()
-                .map(|(name, amount)| NativeAssetDelta { name, amount })
-                .collect();
-
-            assets_vec.push((policy, inner_assets));
-        }
-
-        ValueDelta {
-            lovelace: map.lovelace,
-            assets: assets_vec,
-        }
-    }
-}
-
 impl ValueMap {
     pub fn add_value(&mut self, other: &Value) {
         // Handle lovelace
@@ -619,23 +578,6 @@ pub struct TxOutput {
     pub reference_script: Option<ReferenceScript>,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct TxUTxODeltas {
-    pub tx_identifier: TxIdentifier,
-    pub inputs: Vec<UTxOIdentifier>,
-    pub outputs: Vec<TxOutput>,
-}
-
-impl TxUTxODeltas {
-    pub fn new(tx_identifier: TxIdentifier) -> Self {
-        Self {
-            tx_identifier,
-            inputs: Vec::new(),
-            outputs: Vec::new(),
-        }
-    }
-}
-
 /// Key hash
 pub type KeyHash = Hash<28>;
 
@@ -2185,6 +2127,12 @@ pub struct AssetAddressEntry {
     pub quantity: u64,
 }
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct TxTotals {
+    pub sent: ValueDelta,
+    pub received: Value,
+}
+
 #[derive(
     Debug, Default, Clone, serde::Serialize, serde::Deserialize, minicbor::Encode, minicbor::Decode,
 )]
@@ -2197,12 +2145,6 @@ pub struct AddressTotals {
     pub tx_count: u64,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct TxTotals {
-    pub sent: ValueDelta,
-    pub received: Value,
-}
-
 impl AddAssign for AddressTotals {
     fn add_assign(&mut self, other: Self) {
         self.sent += other.sent;

modules/assets_state/src/state.rs

@@ -1,5 +1,7 @@
 //! Acropolis AssetsState: State storage
 
+use std::collections::HashSet;
+
 use crate::asset_registry::{AssetId, AssetRegistry};
 use acropolis_common::{
     math::update_value_with_delta,
@@ -370,24 +372,27 @@ impl State {
         let store_cfg = self.config.store_transactions;
 
         for tx in deltas {
+            let mut tx_asset_ids = HashSet::new();
             for output in &tx.outputs {
-                let tx_identifier = TxIdentifier::from(output.utxo_identifier);
                 for (policy_id, assets) in &output.value.assets {
                     for asset in assets {
                         if let Some(asset_id) = registry.lookup_id(policy_id, &asset.name) {
-                            let entry = txs_map.entry(asset_id).or_default();
+                            tx_asset_ids.insert(asset_id);
+                        }
+                    }
+                }
+            }
 
-                            let should_push = entry.back() != Some(&tx_identifier);
+            for asset_id in &tx_asset_ids {
+                let entry = txs_map.entry(*asset_id).or_default();
 
-                            if should_push {
-                                entry.push_back(tx_identifier);
+                let last = entry.back().copied();
+                if last != Some(tx.tx_identifier) {
+                    entry.push_back(tx.tx_identifier);
 
-                                if let StoreTransactions::Last(max) = store_cfg {
-                                    if entry.len() as u64 > max {
-                                        entry.pop_front();
-                                    }
-                                }
-                            }
+                    if let StoreTransactions::Last(max) = store_cfg {
+                        if entry.len() as u64 > max {
+                            entry.pop_front();
                         }
                     }
                 }

modules/stake_delta_filter/src/utils.rs

@@ -432,20 +432,20 @@ pub fn process_message(
 mod test {
     use crate::*;
     use acropolis_common::hash::Hash;
-    use acropolis_common::Value;
     use acropolis_common::{
         messages::AddressDeltasMessage, Address, AddressDelta, BlockHash, BlockInfo, BlockStatus,
         ByronAddress, Era, ShelleyAddress, ShelleyAddressDelegationPart, ShelleyAddressPaymentPart,
         ShelleyAddressPointer, StakeAddress, StakeCredential, ValueDelta,
     };
+    use acropolis_common::{TxIdentifier, Value};
     use bech32::{Bech32, Hrp};
     use pallas::ledger::addresses::{PaymentKeyHash, ScriptHash, StakeKeyHash};
 
     fn parse_addr(s: &str) -> Result<AddressDelta> {
         let a = pallas::ledger::addresses::Address::from_bech32(s)?;
         Ok(AddressDelta {
             address: map_address(&a)?,
-            tx_identifier: Default::default(),
+            tx_identifier: TxIdentifier::default(),
             spent_utxos: Vec::new(),
             created_utxos: Vec::new(),
             sent: ValueDelta::default(),

modules/tx_unpacker/src/tx_unpacker.rs

@@ -172,7 +172,6 @@ impl TxUnpacker {
                                     Ok(tx) => {
                                         let tx_hash: TxHash = tx.hash().to_vec().try_into().expect("invalid tx hash length");
                                         let tx_identifier = TxIdentifier::new(block_number, tx_index);
-                                        let mut tx_utxo_deltas  = TxUTxODeltas::new(tx_identifier);
 
                                         let inputs = tx.consumes();
                                         let outputs = tx.produces();
@@ -187,15 +186,16 @@ impl TxUnpacker {
                                         }
 
                                         if publish_utxo_deltas_topic.is_some() {
-                                            // Add all the inputs
-                                            for input in inputs {  // MultiEraInput
-                                                // Lookup and remove UTxOIdentifier from registry 
+                                            // Group deltas by tx
+                                            let mut tx_utxo_deltas  = TxUTxODeltas {tx_identifier, inputs: Vec::new(), outputs: Vec::new()};
+
+                                            // Remove inputs from UTxORegistry and push to UTxOIdentifiers to delta
+                                            for input in inputs {
                                                 let oref = input.output_ref();
                                                 let tx_ref = TxOutRef::new(TxHash::from(**oref.hash()), oref.index() as u16);
 
                                                 match utxo_registry.consume(&tx_ref) {
                                                     Ok(tx_identifier) => {
-                                                        // Add TxInput to utxo_deltas
                                                         tx_utxo_deltas.inputs.push(
                                                             UTxOIdentifier::new(
                                                                 tx_identifier.block_number(),
@@ -210,9 +210,8 @@ impl TxUnpacker {
                                                 }
                                             }
 
-                                            // Add all the outputs
+                                            // Add outputs to UTxORegistry and push TxOutputs to delta
                                             for (index, output) in outputs {
-                                                // Add TxOutRef to registry
                                                 match utxo_registry.add(
                                                     block_number,
                                                     tx_index,
@@ -225,7 +224,6 @@ impl TxUnpacker {
                                                         match output.address() {
                                                             Ok(pallas_address) => match map_parameters::map_address(&pallas_address) {
                                                                 Ok(address) => {
-                                                                    // Add TxOutput to utxo_deltas
                                                                     tx_utxo_deltas.outputs.push(TxOutput {
                                                                         utxo_identifier: utxo_id,
                                                                         address,

modules/utxo_state/src/state.rs

@@ -146,8 +146,7 @@ impl State {
             );
 
             // Delete all UTXOs created in or after this block
-            let created_after = self.volatile_created.prune_on_or_after(block.number);
-            for key in created_after {
+            for key in self.volatile_created.prune_on_or_after(block.number) {
                 self.volatile_utxos.remove(&key);
             }
 
@@ -334,43 +333,40 @@ impl State {
 
         // Process the deltas
         for tx in &deltas.deltas {
-            // Temporary map to sum UTxO deltas efficently
-            let mut address_map: HashMap<
-                Address,
-                (ValueMap, ValueMap, Vec<UTxOIdentifier>, Vec<UTxOIdentifier>),
-            > = HashMap::new();
+            // Temporary map to sum UTxO deltas efficiently
+            let mut address_map: HashMap<Address, AddressTxMap> = HashMap::new();
 
             for input in &tx.inputs {
                 if let Some(utxo) = self.lookup_utxo(input).await? {
                     // Remove or mark spent
                     self.observe_input(input, block).await?;
 
                     let addr = utxo.address.clone();
-                    let (sent, _, spent_utxos, _) = address_map.entry(addr.clone()).or_default();
+                    let entry = address_map.entry(addr.clone()).or_default();
 
-                    spent_utxos.push(*input);
-                    sent.add_value(&utxo.value);
+                    entry.spent_utxos.push(*input);
+                    entry.sent.add_value(&utxo.value);
                 }
             }
 
             for output in &tx.outputs {
                 self.observe_output(output, block).await?;
 
                 let addr = output.address.clone();
-                let (_, received, _, created_utxos) = address_map.entry(addr.clone()).or_default();
+                let entry = address_map.entry(addr.clone()).or_default();
 
-                created_utxos.push(output.utxo_identifier);
-                received.add_value(&output.value);
+                entry.created_utxos.push(output.utxo_identifier);
+                entry.received.add_value(&output.value);
             }
 
-            for (addr, (sent, received, spent_utxos, created_utxos)) in address_map {
+            for (addr, entry) in address_map {
                 let delta = AddressDelta {
                     address: addr,
                     tx_identifier: tx.tx_identifier,
-                    spent_utxos,
-                    created_utxos,
-                    sent: ValueDelta::from(sent),
-                    received: Value::from(received),
+                    spent_utxos: entry.spent_utxos,
+                    created_utxos: entry.created_utxos,
+                    sent: ValueDelta::from(entry.sent),
+                    received: Value::from(entry.received),
                 };
                 if let Some(observer) = self.address_delta_observer.as_ref() {
                     observer.observe_delta(&delta).await;
@@ -387,6 +383,15 @@ impl State {
     }
 }
 
+/// Internal helper used during `handle` aggregation for summing UTxO deltas.
+#[derive(Default)]
+struct AddressTxMap {
+    sent: ValueMap,
+    received: ValueMap,
+    spent_utxos: Vec<UTxOIdentifier>,
+    created_utxos: Vec<UTxOIdentifier>,
+}
+
 // -- Tests --
 #[cfg(test)]
 mod tests {