global: snapshot

crates/brk_computer/src/distribution/address/address_count.rs

@@ -8,7 +8,7 @@ use vecdb::{
     AnyStoredVec, AnyVec, Database, EagerVec, Exit, GenericStoredVec, PcoVec, TypedVecIterator,
 };
 
-use crate::{ComputeIndexes, indexes, internal::ComputedBlockLast};
+use crate::{ComputeIndexes, indexes, internal::ComputedFromHeightLast};
 
 /// Address count per address type (runtime state).
 #[derive(Debug, Default, Deref, DerefMut)]
@@ -78,11 +78,11 @@ impl From<(&AddressTypeToAddrCountVecs, Height)> for AddressTypeToAddressCount {
 
 /// Address count per address type, with height + derived indexes.
 #[derive(Clone, Deref, DerefMut, Traversable)]
-pub struct AddressTypeToAddrCountVecs(ByAddressType<ComputedBlockLast<StoredU64>>);
+pub struct AddressTypeToAddrCountVecs(ByAddressType<ComputedFromHeightLast<StoredU64>>);
 
-impl From<ByAddressType<ComputedBlockLast<StoredU64>>> for AddressTypeToAddrCountVecs {
+impl From<ByAddressType<ComputedFromHeightLast<StoredU64>>> for AddressTypeToAddrCountVecs {
     #[inline]
-    fn from(value: ByAddressType<ComputedBlockLast<StoredU64>>) -> Self {
+    fn from(value: ByAddressType<ComputedFromHeightLast<StoredU64>>) -> Self {
         Self(value)
     }
 }
@@ -95,8 +95,8 @@ impl AddressTypeToAddrCountVecs {
         indexes: &indexes::Vecs,
     ) -> Result<Self> {
         Ok(Self::from(
-            ByAddressType::<ComputedBlockLast<StoredU64>>::new_with_name(|type_name| {
-                ComputedBlockLast::forced_import(
+            ByAddressType::<ComputedFromHeightLast<StoredU64>>::new_with_name(|type_name| {
+                ComputedFromHeightLast::forced_import(
                     db,
                     &format!("{type_name}_{name}"),
                     version,
@@ -224,7 +224,7 @@ impl AddressTypeToAddrCountVecs {
 
 #[derive(Clone, Traversable)]
 pub struct AddrCountVecs {
-    pub all: ComputedBlockLast<StoredU64>,
+    pub all: ComputedFromHeightLast<StoredU64>,
     #[traversable(flatten)]
     pub by_addresstype: AddressTypeToAddrCountVecs,
 }
@@ -237,7 +237,7 @@ impl AddrCountVecs {
         indexes: &indexes::Vecs,
     ) -> Result<Self> {
         Ok(Self {
-            all: ComputedBlockLast::forced_import(db, name, version, indexes)?,
+            all: ComputedFromHeightLast::forced_import(db, name, version, indexes)?,
             by_addresstype: AddressTypeToAddrCountVecs::forced_import(db, name, version, indexes)?,
         })
     }

crates/brk_computer/src/distribution/address/indexes/any.rs

@@ -1,3 +1,6 @@
+use std::thread;
+
+use brk_cohort::ByAddressType;
 use brk_error::{Error, Result};
 use brk_traversable::Traversable;
 use brk_types::{
@@ -6,10 +9,14 @@ use brk_types::{
     TypeIndex, Version,
 };
 use rayon::prelude::*;
+use rustc_hash::FxHashMap;
 use vecdb::{
-    AnyStoredVec, BytesVec, Database, GenericStoredVec, ImportOptions, ImportableVec, Reader, Stamp,
+    AnyStoredVec, AnyVec, BytesVec, Database, GenericStoredVec, ImportOptions, ImportableVec,
+    Reader, Stamp,
 };
 
+use super::super::AddressTypeToTypeIndexMap;
+
 const SAVED_STAMPED_CHANGES: u16 = 10;
 
 /// Macro to define AnyAddressIndexesVecs and its methods.
@@ -75,6 +82,31 @@ macro_rules! define_any_address_indexes_vecs {
                 Ok(())
             }
 
+            /// Get length for a given address type.
+            pub fn len_of(&self, address_type: OutputType) -> usize {
+                match address_type {
+                    $(OutputType::$variant => self.$field.len(),)*
+                    _ => unreachable!("Invalid address type: {:?}", address_type),
+                }
+            }
+
+            /// Update existing entry (must be within bounds).
+            pub fn update(&mut self, address_type: OutputType, typeindex: TypeIndex, index: AnyAddressIndex) -> Result<()> {
+                match address_type {
+                    $(OutputType::$variant => self.$field.update(typeindex.into(), index)?,)*
+                    _ => unreachable!("Invalid address type: {:?}", address_type),
+                }
+                Ok(())
+            }
+
+            /// Push new entry (must be at exactly len position).
+            pub fn push(&mut self, address_type: OutputType, index: AnyAddressIndex) {
+                match address_type {
+                    $(OutputType::$variant => self.$field.push(index),)*
+                    _ => unreachable!("Invalid address type: {:?}", address_type),
+                }
+            }
+
             /// Write all address types with stamp.
             pub fn write(&mut self, stamp: Stamp, with_changes: bool) -> Result<()> {
                 $(self.$field.stamped_write_maybe_with_changes(stamp, with_changes)?;)*
@@ -100,3 +132,102 @@ define_any_address_indexes_vecs!(
     (p2wpkh, P2WPKH, P2WPKHAddressIndex),
     (p2wsh, P2WSH, P2WSHAddressIndex),
 );
+
+impl AnyAddressIndexesVecs {
+    /// Process index updates in parallel by address type.
+    /// Accepts two maps (e.g. from empty and loaded processing) and merges per-thread.
+    /// Updates existing entries and pushes new ones (sorted).
+    /// Returns (update_count, push_count).
+    pub fn par_batch_update(
+        &mut self,
+        updates1: AddressTypeToTypeIndexMap<AnyAddressIndex>,
+        updates2: AddressTypeToTypeIndexMap<AnyAddressIndex>,
+    ) -> Result<(usize, usize)> {
+        let ByAddressType {
+            p2a: u1_p2a,
+            p2pk33: u1_p2pk33,
+            p2pk65: u1_p2pk65,
+            p2pkh: u1_p2pkh,
+            p2sh: u1_p2sh,
+            p2tr: u1_p2tr,
+            p2wpkh: u1_p2wpkh,
+            p2wsh: u1_p2wsh,
+        } = updates1.into_inner();
+
+        let ByAddressType {
+            p2a: u2_p2a,
+            p2pk33: u2_p2pk33,
+            p2pk65: u2_p2pk65,
+            p2pkh: u2_p2pkh,
+            p2sh: u2_p2sh,
+            p2tr: u2_p2tr,
+            p2wpkh: u2_p2wpkh,
+            p2wsh: u2_p2wsh,
+        } = updates2.into_inner();
+
+        let Self {
+            p2a,
+            p2pk33,
+            p2pk65,
+            p2pkh,
+            p2sh,
+            p2tr,
+            p2wpkh,
+            p2wsh,
+        } = self;
+
+        thread::scope(|s| {
+            let h_p2a = s.spawn(|| process_single_type_merged(p2a, u1_p2a, u2_p2a));
+            let h_p2pk33 = s.spawn(|| process_single_type_merged(p2pk33, u1_p2pk33, u2_p2pk33));
+            let h_p2pk65 = s.spawn(|| process_single_type_merged(p2pk65, u1_p2pk65, u2_p2pk65));
+            let h_p2pkh = s.spawn(|| process_single_type_merged(p2pkh, u1_p2pkh, u2_p2pkh));
+            let h_p2sh = s.spawn(|| process_single_type_merged(p2sh, u1_p2sh, u2_p2sh));
+            let h_p2tr = s.spawn(|| process_single_type_merged(p2tr, u1_p2tr, u2_p2tr));
+            let h_p2wpkh = s.spawn(|| process_single_type_merged(p2wpkh, u1_p2wpkh, u2_p2wpkh));
+            let h_p2wsh = s.spawn(|| process_single_type_merged(p2wsh, u1_p2wsh, u2_p2wsh));
+
+            let mut total_updates = 0usize;
+            let mut total_pushes = 0usize;
+
+            for h in [
+                h_p2a, h_p2pk33, h_p2pk65, h_p2pkh, h_p2sh, h_p2tr, h_p2wpkh, h_p2wsh,
+            ] {
+                let (updates, pushes) = h.join().unwrap()?;
+                total_updates += updates;
+                total_pushes += pushes;
+            }
+
+            Ok((total_updates, total_pushes))
+        })
+    }
+}
+
+/// Process updates for a single address type's BytesVec, merging two maps.
+fn process_single_type_merged<I: vecdb::VecIndex>(
+    vec: &mut BytesVec<I, AnyAddressIndex>,
+    map1: FxHashMap<TypeIndex, AnyAddressIndex>,
+    map2: FxHashMap<TypeIndex, AnyAddressIndex>,
+) -> Result<(usize, usize)> {
+    let current_len = vec.len();
+    let mut pushes = Vec::with_capacity(map1.len() + map2.len());
+    let mut update_count = 0usize;
+
+    for (typeindex, any_index) in map1.into_iter().chain(map2) {
+        if usize::from(typeindex) < current_len {
+            vec.update(I::from(usize::from(typeindex)), any_index)?;
+            update_count += 1;
+        } else {
+            pushes.push((typeindex, any_index));
+        }
+    }
+
+    let push_count = pushes.len();
+    if !pushes.is_empty() {
+        pushes.sort_unstable_by_key(|(typeindex, _)| *typeindex);
+        for (_, any_index) in pushes {
+            vec.push(any_index);
+        }
+    }
+
+    Ok((update_count, push_count))
+}

crates/brk_computer/src/distribution/address/type_map/index_map.rs

@@ -92,6 +92,11 @@ impl<T> AddressTypeToTypeIndexMap<T> {
         self.0.into_iter()
     }
 
+    /// Consume and return the inner ByAddressType.
+    pub fn into_inner(self) -> ByAddressType<FxHashMap<TypeIndex, T>> {
+        self.0
+    }
+
     /// Iterate mutably over entries by address type.
     pub fn iter_mut(&mut self) -> impl Iterator<Item = (OutputType, &mut FxHashMap<TypeIndex, T>)> {
         self.0.iter_mut()