global: snapshot

crates/brk_computer/src/grouped/builder_eager.rs

@@ -1,6 +1,7 @@
 use brk_error::{Error, Result};
 use brk_traversable::Traversable;
 use brk_types::{CheckedSub, StoredU64, Version};
+use schemars::JsonSchema;
 use vecdb::{
     AnyStoredVec, Database, EagerVec, Exit, GenericStoredVec, ImportableVec, IterableVec, PcoVec,
     VecIndex, VecValue,
@@ -16,7 +17,7 @@ const VERSION: Version = Version::ZERO;
 pub struct EagerVecsBuilder<I, T>
 where
     I: VecIndex,
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     pub first: Option<Box<EagerVec<PcoVec<I, T>>>>,
     pub average: Option<Box<EagerVec<PcoVec<I, T>>>>,
@@ -35,7 +36,7 @@ where
 impl<I, T> EagerVecsBuilder<I, T>
 where
     I: VecIndex,
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     pub fn forced_import(
         db: &Database,
@@ -159,7 +160,12 @@ where
     /// Compute percentiles from sorted values (assumes values is already sorted)
     fn compute_percentiles_from_sorted(&mut self, index: usize, values: &[T]) -> Result<()> {
         if let Some(max) = self.max.as_mut() {
-            max.truncate_push_at(index, *values.last().ok_or(Error::Internal("Empty values for percentiles"))?)?;
+            max.truncate_push_at(
+                index,
+                *values
+                    .last()
+                    .ok_or(Error::Internal("Empty values for percentiles"))?,
+            )?;
         }
         if let Some(pct90) = self.pct90.as_mut() {
             pct90.truncate_push_at(index, get_percentile(values, 0.90))?;

crates/brk_computer/src/grouped/builder_lazy.rs

@@ -1,5 +1,6 @@
 use brk_traversable::Traversable;
 use brk_types::Version;
+use schemars::JsonSchema;
 use vecdb::{FromCoarserIndex, IterableBoxedVec, IterableCloneableVec, LazyVecFrom2, VecIndex};
 
 use crate::grouped::{EagerVecsBuilder, VecBuilderOptions};
@@ -12,7 +13,7 @@ use super::ComputedVecValue;
 pub struct LazyVecsBuilder<I, T, S1I, S2T>
 where
     I: VecIndex,
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
     S1I: VecIndex,
     S2T: ComputedVecValue,
 {
@@ -30,7 +31,7 @@ const VERSION: Version = Version::ZERO;
 impl<I, T, S1I, S2T> LazyVecsBuilder<I, T, S1I, S2T>
 where
     I: VecIndex,
-    T: ComputedVecValue + 'static,
+    T: ComputedVecValue + JsonSchema + 'static,
     S1I: VecIndex + 'static + FromCoarserIndex<I>,
     S2T: ComputedVecValue,
 {

crates/brk_computer/src/grouped/from_dateindex.rs

@@ -4,6 +4,7 @@ use brk_traversable::Traversable;
 use brk_types::{
     DateIndex, DecadeIndex, MonthIndex, QuarterIndex, SemesterIndex, Version, WeekIndex, YearIndex,
 };
+use schemars::JsonSchema;
 use vecdb::{
     AnyExportableVec, Database, EagerVec, Exit, ImportableVec, IterableCloneableVec, IterableVec,
     PcoVec,
@@ -16,7 +17,7 @@ use super::{ComputedVecValue, EagerVecsBuilder, Source, VecBuilderOptions};
 #[derive(Clone)]
 pub struct ComputedVecsFromDateIndex<T>
 where
-    T: ComputedVecValue + PartialOrd,
+    T: ComputedVecValue + PartialOrd + JsonSchema,
 {
     pub dateindex: Option<EagerVec<PcoVec<DateIndex, T>>>,
     pub dateindex_extra: EagerVecsBuilder<DateIndex, T>,
@@ -32,7 +33,7 @@ const VERSION: Version = Version::ZERO;
 
 impl<T> ComputedVecsFromDateIndex<T>
 where
-    T: ComputedVecValue + 'static,
+    T: ComputedVecValue + JsonSchema + 'static,
 {
     #[allow(clippy::too_many_arguments)]
     pub fn forced_import(
@@ -149,7 +150,7 @@ where
 
 impl<T> Traversable for ComputedVecsFromDateIndex<T>
 where
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     fn to_tree_node(&self) -> brk_traversable::TreeNode {
         let dateindex_extra_node = self.dateindex_extra.to_tree_node();

crates/brk_computer/src/grouped/from_height.rs

@@ -5,6 +5,7 @@ use brk_types::{
     DateIndex, DecadeIndex, DifficultyEpoch, Height, MonthIndex, QuarterIndex, SemesterIndex,
     Version, WeekIndex, YearIndex,
 };
+use schemars::JsonSchema;
 use vecdb::{
     AnyExportableVec, Database, EagerVec, Exit, ImportableVec, IterableCloneableVec, IterableVec,
     PcoVec,
@@ -22,7 +23,7 @@ use super::{ComputedVecValue, EagerVecsBuilder, VecBuilderOptions};
 #[derive(Clone)]
 pub struct ComputedVecsFromHeight<T>
 where
-    T: ComputedVecValue + PartialOrd,
+    T: ComputedVecValue + PartialOrd + JsonSchema,
 {
     pub height: Option<EagerVec<PcoVec<Height, T>>>,
     pub height_extra: EagerVecsBuilder<Height, T>,
@@ -41,7 +42,7 @@ const VERSION: Version = Version::ZERO;
 
 impl<T> ComputedVecsFromHeight<T>
 where
-    T: ComputedVecValue + Ord + From<f64> + 'static,
+    T: ComputedVecValue + Ord + From<f64> + JsonSchema + 'static,
     f64: From<T>,
 {
     #[allow(clippy::too_many_arguments)]
@@ -202,7 +203,7 @@ where
 
 impl<T> Traversable for ComputedVecsFromHeight<T>
 where
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     fn to_tree_node(&self) -> brk_traversable::TreeNode {
         let height_extra_node = self.height_extra.to_tree_node();

crates/brk_computer/src/grouped/from_height_strict.rs

@@ -2,6 +2,7 @@ use brk_error::Result;
 
 use brk_traversable::Traversable;
 use brk_types::{DifficultyEpoch, Height, Version};
+use schemars::JsonSchema;
 use vecdb::{AnyExportableVec, Database, EagerVec, Exit, ImportableVec, PcoVec};
 
 use crate::{Indexes, indexes};
@@ -11,7 +12,7 @@ use super::{ComputedVecValue, EagerVecsBuilder, VecBuilderOptions};
 #[derive(Clone)]
 pub struct ComputedVecsFromHeightStrict<T>
 where
-    T: ComputedVecValue + PartialOrd,
+    T: ComputedVecValue + PartialOrd + JsonSchema,
 {
     pub height: EagerVec<PcoVec<Height, T>>,
     pub height_extra: EagerVecsBuilder<Height, T>,
@@ -23,7 +24,7 @@ const VERSION: Version = Version::ZERO;
 
 impl<T> ComputedVecsFromHeightStrict<T>
 where
-    T: ComputedVecValue + Ord + From<f64>,
+    T: ComputedVecValue + Ord + From<f64> + JsonSchema,
     f64: From<T>,
 {
     pub fn forced_import(
@@ -85,7 +86,7 @@ where
 
 impl<T> Traversable for ComputedVecsFromHeightStrict<T>
 where
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     fn to_tree_node(&self) -> brk_traversable::TreeNode {
         let height_extra_node = self.height_extra.to_tree_node();

crates/brk_computer/src/grouped/from_txindex.rs

@@ -5,6 +5,7 @@ use brk_types::{
     Bitcoin, DateIndex, DecadeIndex, DifficultyEpoch, Dollars, Height, MonthIndex, QuarterIndex,
     Sats, SemesterIndex, TxIndex, Version, WeekIndex, YearIndex,
 };
+use schemars::JsonSchema;
 use vecdb::{
     AnyExportableVec, AnyVec, CollectableVec, Database, EagerVec, Exit, GenericStoredVec,
     ImportableVec, IterableCloneableVec, PcoVec, TypedVecIterator, VecIndex,
@@ -22,7 +23,7 @@ use super::{ComputedVecValue, EagerVecsBuilder, VecBuilderOptions};
 #[derive(Clone)]
 pub struct ComputedVecsFromTxindex<T>
 where
-    T: ComputedVecValue + PartialOrd,
+    T: ComputedVecValue + PartialOrd + JsonSchema,
 {
     pub txindex: Option<Box<EagerVec<PcoVec<TxIndex, T>>>>,
     pub height: EagerVecsBuilder<Height, T>,
@@ -41,7 +42,7 @@ const VERSION: Version = Version::ZERO;
 
 impl<T> ComputedVecsFromTxindex<T>
 where
-    T: ComputedVecValue + Ord + From<f64> + 'static,
+    T: ComputedVecValue + Ord + From<f64> + JsonSchema + 'static,
     f64: From<T>,
 {
     #[allow(clippy::too_many_arguments)]
@@ -256,10 +257,8 @@ impl ComputedVecsFromTxindex<Bitcoin> {
             .map(Height::from)
             .try_for_each(|height| -> Result<()> {
                 if let Some(first) = self.height.first.as_mut() {
-                    first.truncate_push(
-                        height,
-                        Bitcoin::from(first_iter.um().get_unwrap(height)),
-                    )?;
+                    first
+                        .truncate_push(height, Bitcoin::from(first_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(average) = self.height.average.as_mut() {
                     average.truncate_push(
@@ -268,28 +267,18 @@ impl ComputedVecsFromTxindex<Bitcoin> {
                     )?;
                 }
                 if let Some(sum) = self.height.sum.as_mut() {
-                    sum.truncate_push(
-                        height,
-                        Bitcoin::from(sum_iter.um().get_unwrap(height)),
-                    )?;
+                    sum.truncate_push(height, Bitcoin::from(sum_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(max) = self.height.max.as_mut() {
-                    max.truncate_push(
-                        height,
-                        Bitcoin::from(max_iter.um().get_unwrap(height)),
-                    )?;
+                    max.truncate_push(height, Bitcoin::from(max_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(pct90) = self.height.pct90.as_mut() {
-                    pct90.truncate_push(
-                        height,
-                        Bitcoin::from(pct90_iter.um().get_unwrap(height)),
-                    )?;
+                    pct90
+                        .truncate_push(height, Bitcoin::from(pct90_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(pct75) = self.height.pct75.as_mut() {
-                    pct75.truncate_push(
-                        height,
-                        Bitcoin::from(pct75_iter.um().get_unwrap(height)),
-                    )?;
+                    pct75
+                        .truncate_push(height, Bitcoin::from(pct75_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(median) = self.height.median.as_mut() {
                     median.truncate_push(
@@ -298,28 +287,18 @@ impl ComputedVecsFromTxindex<Bitcoin> {
                     )?;
                 }
                 if let Some(pct25) = self.height.pct25.as_mut() {
-                    pct25.truncate_push(
-                        height,
-                        Bitcoin::from(pct25_iter.um().get_unwrap(height)),
-                    )?;
+                    pct25
+                        .truncate_push(height, Bitcoin::from(pct25_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(pct10) = self.height.pct10.as_mut() {
-                    pct10.truncate_push(
-                        height,
-                        Bitcoin::from(pct10_iter.um().get_unwrap(height)),
-                    )?;
+                    pct10
+                        .truncate_push(height, Bitcoin::from(pct10_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(min) = self.height.min.as_mut() {
-                    min.truncate_push(
-                        height,
-                        Bitcoin::from(min_iter.um().get_unwrap(height)),
-                    )?;
+                    min.truncate_push(height, Bitcoin::from(min_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(last) = self.height.last.as_mut() {
-                    last.truncate_push(
-                        height,
-                        Bitcoin::from(last_iter.um().get_unwrap(height)),
-                    )?;
+                    last.truncate_push(height, Bitcoin::from(last_iter.um().get_unwrap(height)))?;
                 }
                 if let Some(cumulative) = self.height.cumulative.as_mut() {
                     cumulative.truncate_push(
@@ -381,76 +360,41 @@ impl ComputedVecsFromTxindex<Dollars> {
                 let price = *close_iter.get_unwrap(height);
 
                 if let Some(first) = self.height.first.as_mut() {
-                    first.truncate_push(
-                        height,
-                        price * first_iter.um().get_unwrap(height),
-                    )?;
+                    first.truncate_push(height, price * first_iter.um().get_unwrap(height))?;
                 }
                 if let Some(average) = self.height.average.as_mut() {
-                    average.truncate_push(
-                        height,
-                        price * average_iter.um().get_unwrap(height),
-                    )?;
+                    average.truncate_push(height, price * average_iter.um().get_unwrap(height))?;
                 }
                 if let Some(sum) = self.height.sum.as_mut() {
-                    sum.truncate_push(
-                        height,
-                        price * sum_iter.um().get_unwrap(height),
-                    )?;
+                    sum.truncate_push(height, price * sum_iter.um().get_unwrap(height))?;
                 }
                 if let Some(max) = self.height.max.as_mut() {
-                    max.truncate_push(
-                        height,
-                        price * max_iter.um().get_unwrap(height),
-                    )?;
+                    max.truncate_push(height, price * max_iter.um().get_unwrap(height))?;
                 }
                 if let Some(pct90) = self.height.pct90.as_mut() {
-                    pct90.truncate_push(
-                        height,
-                        price * pct90_iter.um().get_unwrap(height),
-                    )?;
+                    pct90.truncate_push(height, price * pct90_iter.um().get_unwrap(height))?;
                 }
                 if let Some(pct75) = self.height.pct75.as_mut() {
-                    pct75.truncate_push(
-                        height,
-                        price * pct75_iter.um().get_unwrap(height),
-                    )?;
+                    pct75.truncate_push(height, price * pct75_iter.um().get_unwrap(height))?;
                 }
                 if let Some(median) = self.height.median.as_mut() {
-                    median.truncate_push(
-                        height,
-                        price * median_iter.um().get_unwrap(height),
-                    )?;
+                    median.truncate_push(height, price * median_iter.um().get_unwrap(height))?;
                 }
                 if let Some(pct25) = self.height.pct25.as_mut() {
-                    pct25.truncate_push(
-                        height,
-                        price * pct25_iter.um().get_unwrap(height),
-                    )?;
+                    pct25.truncate_push(height, price * pct25_iter.um().get_unwrap(height))?;
                 }
                 if let Some(pct10) = self.height.pct10.as_mut() {
-                    pct10.truncate_push(
-                        height,
-                        price * pct10_iter.um().get_unwrap(height),
-                    )?;
+                    pct10.truncate_push(height, price * pct10_iter.um().get_unwrap(height))?;
                 }
                 if let Some(min) = self.height.min.as_mut() {
-                    min.truncate_push(
-                        height,
-                        price * min_iter.um().get_unwrap(height),
-                    )?;
+                    min.truncate_push(height, price * min_iter.um().get_unwrap(height))?;
                 }
                 if let Some(last) = self.height.last.as_mut() {
-                    last.truncate_push(
-                        height,
-                        price * last_iter.um().get_unwrap(height),
-                    )?;
+                    last.truncate_push(height, price * last_iter.um().get_unwrap(height))?;
                 }
                 if let Some(cumulative) = self.height.cumulative.as_mut() {
-                    cumulative.truncate_push(
-                        height,
-                        price * cumulative_iter.um().get_unwrap(height),
-                    )?;
+                    cumulative
+                        .truncate_push(height, price * cumulative_iter.um().get_unwrap(height))?;
                 }
                 Ok(())
             })?;
@@ -463,7 +407,7 @@ impl ComputedVecsFromTxindex<Dollars> {
 
 impl<T> Traversable for ComputedVecsFromTxindex<T>
 where
-    T: ComputedVecValue,
+    T: ComputedVecValue + JsonSchema,
 {
     fn to_tree_node(&self) -> brk_traversable::TreeNode {
         brk_traversable::TreeNode::Branch(

crates/brk_computer/src/grouped/sd_from_dateindex.rs

@@ -1,9 +1,11 @@
+use std::mem;
+
 use brk_error::Result;
 use brk_traversable::Traversable;
 use brk_types::{Date, DateIndex, Dollars, StoredF32, Version};
-use vecdb::{PcoVec, 
+use vecdb::{
     AnyStoredVec, AnyVec, BoxedVecIterator, CollectableVec, Database, EagerVec, Exit,
-    GenericStoredVec, IterableVec, VecIndex,
+    GenericStoredVec, IterableVec, PcoVec, VecIndex,
 };
 
 use crate::{Indexes, grouped::source::Source, indexes, utils::OptionExt};
@@ -109,8 +111,14 @@ impl ComputedStandardDeviationVecsFromDateIndex {
         macro_rules! import {
             ($suffix:expr) => {
                 ComputedVecsFromDateIndex::forced_import(
-                    db, &format!("{name}_{}", $suffix), Source::Compute, version, indexes, opts,
-                ).unwrap()
+                    db,
+                    &format!("{name}_{}", $suffix),
+                    Source::Compute,
+                    version,
+                    indexes,
+                    opts,
+                )
+                .unwrap()
             };
         }
 
@@ -183,9 +191,7 @@ impl ComputedStandardDeviationVecsFromDateIndex {
         source: &impl CollectableVec<DateIndex, StoredF32>,
         price_opt: Option<&impl IterableVec<DateIndex, Dollars>>,
     ) -> Result<()> {
-        let sma = sma_opt.unwrap_or_else(|| unsafe {
-            std::mem::transmute(&self.sma.u().dateindex)
-        });
+        let sma = sma_opt.unwrap_or_else(|| unsafe { mem::transmute(&self.sma.u().dateindex) });
 
         let min_date = DateIndex::try_from(Date::MIN_RATIO).unwrap();
 
@@ -345,7 +351,11 @@ impl ComputedStandardDeviationVecsFromDateIndex {
             .try_for_each(|v| v.safe_flush(exit))?;
 
         self.mut_stateful_computed().try_for_each(|v| {
-            v.compute_rest(starting_indexes, exit, None as Option<&EagerVec<PcoVec<_, _>>>)
+            v.compute_rest(
+                starting_indexes,
+                exit,
+                None as Option<&EagerVec<PcoVec<_, _>>>,
+            )
         })?;
 
         if let Some(zscore) = self.zscore.as_mut() {
@@ -536,7 +546,6 @@ impl ComputedStandardDeviationVecsFromDateIndex {
     fn mut_stateful_date_vecs(
         &mut self,
     ) -> impl Iterator<Item = &mut EagerVec<PcoVec<DateIndex, StoredF32>>> {
-        self.mut_stateful_computed()
-            .map(|c| c.dateindex.um())
+        self.mut_stateful_computed().map(|c| c.dateindex.um())
     }
 }

crates/brk_computer/src/pools/mod.rs

@@ -217,7 +217,7 @@ impl Vecs {
                     .or_else(|| self.pools.find_from_coinbase_tag(&coinbase_tag))
                     .unwrap_or(unknown);
 
-                self.height_to_pool.push_if_needed(height, pool.slug)?;
+                self.height_to_pool.truncate_push(height, pool.slug)?;
                 Ok(())
             })?;
 

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

@@ -6,7 +6,7 @@ use brk_traversable::Traversable;
 use brk_types::{Height, StoredU64, Version};
 use derive_deref::{Deref, DerefMut};
 use vecdb::{
-    AnyStoredVec, Database, EagerVec, Exit, GenericStoredVec, ImportableVec, PcoVec,
+    AnyStoredVec, AnyVec, Database, EagerVec, Exit, GenericStoredVec, ImportableVec, PcoVec,
     TypedVecIterator,
 };
 
@@ -54,6 +54,18 @@ impl From<ByAddressType<EagerVec<PcoVec<Height, StoredU64>>>>
 }
 
 impl AddressTypeToHeightToAddressCount {
+    pub fn min_len(&self) -> usize {
+        self.p2pk65
+            .len()
+            .min(self.p2pk33.len())
+            .min(self.p2pkh.len())
+            .min(self.p2sh.len())
+            .min(self.p2wpkh.len())
+            .min(self.p2wsh.len())
+            .min(self.p2tr.len())
+            .min(self.p2a.len())
+    }
+
     pub fn forced_import(db: &Database, name: &str, version: Version) -> Result<Self> {
         Ok(Self::from(ByAddressType::new_with_name(|type_name| {
             Ok(EagerVec::forced_import(

crates/brk_computer/src/stateful/compute/write.rs

@@ -36,6 +36,8 @@ pub fn process_address_updates(
     empty_updates: AddressTypeToTypeIndexMap<EmptyAddressDataWithSource>,
     loaded_updates: AddressTypeToTypeIndexMap<LoadedAddressDataWithSource>,
 ) -> Result<()> {
+    info!("Processing address updates...");
+
     let empty_result = process_empty_addresses(addresses_data, empty_updates)?;
     let loaded_result = process_loaded_addresses(addresses_data, loaded_updates)?;
     let all_updates = empty_result.merge(loaded_result);

crates/brk_computer/src/stateful/process/lookup.rs

@@ -5,52 +5,52 @@ use brk_types::{LoadedAddressData, OutputType, TypeIndex};
 use super::super::address::AddressTypeToTypeIndexMap;
 use super::{EmptyAddressDataWithSource, LoadedAddressDataWithSource, WithAddressDataSource};
 
+/// Source of an address in lookup - reports where the data came from.
+#[derive(Clone, Copy)]
+pub enum AddressSource {
+    /// Brand new address (never seen before)
+    New,
+    /// Loaded from disk (has existing balance)
+    Loaded,
+    /// Was empty (zero balance), now receiving
+    FromEmpty,
+}
+
 /// Context for looking up and storing address data during block processing.
-///
-/// All addresses should be pre-fetched into the cache before using this.
-/// - `loaded`: addresses with non-zero balance (wrapped with source info)
-/// - `empty`: addresses that became empty this block (wrapped with source info)
 pub struct AddressLookup<'a> {
-    /// Loaded addresses touched in current block
     pub loaded: &'a mut AddressTypeToTypeIndexMap<LoadedAddressDataWithSource>,
-    /// Empty addresses touched in current block
     pub empty: &'a mut AddressTypeToTypeIndexMap<EmptyAddressDataWithSource>,
 }
 
 impl<'a> AddressLookup<'a> {
-    /// Get or create address data for a receive operation.
-    ///
-    /// Returns (address_data, is_new, from_empty)
     pub fn get_or_create_for_receive(
         &mut self,
         output_type: OutputType,
         type_index: TypeIndex,
-    ) -> (&mut LoadedAddressDataWithSource, bool, bool) {
+    ) -> (&mut LoadedAddressDataWithSource, AddressSource) {
         use std::collections::hash_map::Entry;
 
         let map = self.loaded.get_mut(output_type).unwrap();
 
         match map.entry(type_index) {
             Entry::Occupied(entry) => {
-                // Entry already exists - check its source
-                let data = entry.into_mut();
-                let is_new = data.is_new();
-                let from_empty = data.is_from_emptyaddressdata();
-                (data, is_new, from_empty)
+                let source = match entry.get() {
+                    WithAddressDataSource::New(_) => AddressSource::New,
+                    WithAddressDataSource::FromLoaded(..) => AddressSource::Loaded,
+                    WithAddressDataSource::FromEmpty(..) => AddressSource::FromEmpty,
+                };
+                (entry.into_mut(), source)
             }
             Entry::Vacant(entry) => {
-                // Check if it was in empty set
                 if let Some(empty_data) =
                     self.empty.get_mut(output_type).unwrap().remove(&type_index)
                 {
-                    let data = entry.insert(empty_data.into());
-                    return (data, false, true);
+                    return (entry.insert(empty_data.into()), AddressSource::FromEmpty);
                 }
-
-                // Not found - create new address
-                let data =
-                    entry.insert(WithAddressDataSource::New(LoadedAddressData::default()));
-                (data, true, false)
+                (
+                    entry.insert(WithAddressDataSource::New(LoadedAddressData::default())),
+                    AddressSource::New,
+                )
             }
         }
     }

crates/brk_computer/src/stateful/process/received.rs

@@ -1,23 +1,13 @@
 //! Process received outputs for address cohorts.
-//!
-//! Updates address cohort states when addresses receive funds:
-//! - New addresses enter a cohort
-//! - Existing addresses may cross cohort boundaries
-//! - Empty addresses become non-empty again
 
-use brk_grouper::{ByAddressType, Filtered};
+use brk_grouper::{AmountBucket, ByAddressType};
 use brk_types::{Dollars, Sats, TypeIndex};
+use rustc_hash::FxHashMap;
 
 use super::super::address::AddressTypeToVec;
 use super::super::cohorts::AddressCohorts;
-use super::lookup::AddressLookup;
+use super::lookup::{AddressLookup, AddressSource};
 
-/// Process received outputs for address cohorts.
-///
-/// For each received output:
-/// 1. Look up or create address data
-/// 2. Update address balance and cohort membership
-/// 3. Update cohort states (add/subtract for boundary crossings, receive otherwise)
 pub fn process_received(
     received_data: AddressTypeToVec<(TypeIndex, Sats)>,
     cohorts: &mut AddressCohorts,
@@ -31,30 +21,47 @@ pub fn process_received(
             continue;
         }
 
+        // Aggregate receives by address - each address processed exactly once
+        // Track (total_value, output_count) for correct UTXO counting
+        let mut aggregated: FxHashMap<TypeIndex, (Sats, u32)> = FxHashMap::default();
         for (type_index, value) in vec {
-            let (addr_data, is_new, from_empty) =
-                lookup.get_or_create_for_receive(output_type, type_index);
+            let entry = aggregated.entry(type_index).or_default();
+            entry.0 += value;
+            entry.1 += 1;
+        }
 
-            // Update address counts
-            if is_new || from_empty {
-                *addr_count.get_mut(output_type).unwrap() += 1;
-                if from_empty {
+        for (type_index, (total_value, output_count)) in aggregated {
+            let (addr_data, source) = lookup.get_or_create_for_receive(output_type, type_index);
+
+            match source {
+                AddressSource::New => {
+                    *addr_count.get_mut(output_type).unwrap() += 1;
+                }
+                AddressSource::FromEmpty => {
+                    *addr_count.get_mut(output_type).unwrap() += 1;
                     *empty_addr_count.get_mut(output_type).unwrap() -= 1;
                 }
+                AddressSource::Loaded => {}
             }
 
-            let prev_balance = addr_data.balance();
-            let new_balance = prev_balance + value;
+            let is_new_entry = matches!(source, AddressSource::New | AddressSource::FromEmpty);
 
-            // Check if crossing cohort boundary
-            let prev_cohort = cohorts.amount_range.get(prev_balance);
-            let new_cohort = cohorts.amount_range.get(new_balance);
-            let filters_differ = prev_cohort.filter() != new_cohort.filter();
+            if is_new_entry {
+                // New/from-empty address - just add to cohort
+                addr_data.receive_outputs(total_value, price, output_count);
+                cohorts
+                    .amount_range
+                    .get_mut(total_value) // new_balance = 0 + total_value
+                    .state
+                    .as_mut()
+                    .unwrap()
+                    .add(addr_data);
+            } else {
+                let prev_balance = addr_data.balance();
+                let new_balance = prev_balance + total_value;
 
-            if is_new || from_empty || filters_differ {
-                // Address entering or changing cohorts
-                if !is_new && !from_empty {
-                    // Subtract from old cohort
+                if AmountBucket::from(prev_balance) != AmountBucket::from(new_balance) {
+                    // Crossing cohort boundary - subtract from old, add to new
                     cohorts
                         .amount_range
                         .get_mut(prev_balance)
@@ -62,28 +69,24 @@ pub fn process_received(
                         .as_mut()
                         .unwrap()
                         .subtract(addr_data);
+                    addr_data.receive_outputs(total_value, price, output_count);
+                    cohorts
+                        .amount_range
+                        .get_mut(new_balance)
+                        .state
+                        .as_mut()
+                        .unwrap()
+                        .add(addr_data);
+                } else {
+                    // Staying in same cohort - just receive
+                    cohorts
+                        .amount_range
+                        .get_mut(new_balance)
+                        .state
+                        .as_mut()
+                        .unwrap()
+                        .receive_outputs(addr_data, total_value, price, output_count);
                 }
-
-                // Update address data
-                addr_data.receive(value, price);
-
-                // Add to new cohort
-                cohorts
-                    .amount_range
-                    .get_mut(new_balance)
-                    .state
-                    .as_mut()
-                    .unwrap()
-                    .add(addr_data);
-            } else {
-                // Address staying in same cohort - update in place
-                cohorts
-                    .amount_range
-                    .get_mut(new_balance)
-                    .state
-                    .as_mut()
-                    .unwrap()
-                    .receive(addr_data, value, price);
             }
         }
     }

crates/brk_computer/src/stateful/process/with_source.rs

@@ -26,16 +26,6 @@ pub enum WithAddressDataSource<T> {
     FromEmpty(EmptyAddressIndex, T),
 }
 
-impl<T> WithAddressDataSource<T> {
-    pub fn is_new(&self) -> bool {
-        matches!(self, Self::New(_))
-    }
-
-    pub fn is_from_emptyaddressdata(&self) -> bool {
-        matches!(self, Self::FromEmpty(..))
-    }
-}
-
 impl<T> std::ops::Deref for WithAddressDataSource<T> {
     type Target = T;
 

crates/brk_computer/src/stateful/states/address_cohort.rs

@@ -89,6 +89,16 @@ impl AddressCohortState {
         address_data: &mut LoadedAddressData,
         value: Sats,
         price: Option<Dollars>,
+    ) {
+        self.receive_outputs(address_data, value, price, 1);
+    }
+
+    pub fn receive_outputs(
+        &mut self,
+        address_data: &mut LoadedAddressData,
+        value: Sats,
+        price: Option<Dollars>,
+        output_count: u32,
     ) {
         let compute_price = price.is_some();
 
@@ -98,7 +108,7 @@ impl AddressCohortState {
             value: address_data.balance(),
         };
 
-        address_data.receive(value, price);
+        address_data.receive_outputs(value, price, output_count);
 
         let supply_state = SupplyState {
             utxo_count: address_data.utxo_count() as u64,
@@ -107,7 +117,7 @@ impl AddressCohortState {
 
         self.inner.receive_(
             &SupplyState {
-                utxo_count: 1,
+                utxo_count: output_count as u64,
                 value,
             },
             price,

crates/brk_computer/src/stateful/states/supply.rs

@@ -1,10 +1,11 @@
 use std::ops::{Add, AddAssign, SubAssign};
 
 use brk_types::{CheckedSub, LoadedAddressData, Sats};
+use schemars::JsonSchema;
 use serde::Serialize;
 use vecdb::{Bytes, Formattable};
 
-#[derive(Debug, Default, Clone, Serialize)]
+#[derive(Debug, Default, Clone, Serialize, JsonSchema)]
 pub struct SupplyState {
     pub utxo_count: u64,
     pub value: Sats,

crates/brk_computer/src/stateful/vecs.rs

@@ -269,7 +269,11 @@ impl Vecs {
             .min(self.any_address_indexes.min_stamped_height())
             .min(self.addresses_data.min_stamped_height())
             .min(Height::from(self.height_to_unspendable_supply.len()))
-            .min(Height::from(self.height_to_opreturn_supply.len()));
+            .min(Height::from(self.height_to_opreturn_supply.len()))
+            .min(Height::from(self.addresstype_to_height_to_addr_count.min_len()))
+            .min(Height::from(
+                self.addresstype_to_height_to_empty_addr_count.min_len(),
+            ));
 
         // 2. Determine start mode and recover/reset state
         let start_mode = determine_start_mode(stateful_min, chain_state_height);