global: snapshot

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

@@ -34,10 +34,30 @@ impl<'a> AddressLookup<'a> {
 
         match map.entry(type_index) {
             Entry::Occupied(entry) => {
+                // Address is in cache. Need to determine if it's been processed
+                // by process_received (added to a cohort) or just loaded this block.
+                //
+                // - If wrapper is New AND funded_txo_count == 0: hasn't received yet,
+                //   was just created in process_outputs this block → New
+                // - If wrapper is New AND funded_txo_count > 0: received in previous
+                //   block but still in cache (no flush) → Loaded
+                // - If wrapper is FromLoaded/FromEmpty: loaded from storage → use wrapper
                 let source = match entry.get() {
-                    WithAddressDataSource::New(_) => AddressSource::New,
+                    WithAddressDataSource::New(data) => {
+                        if data.funded_txo_count == 0 {
+                            AddressSource::New
+                        } else {
+                            AddressSource::Loaded
+                        }
+                    }
                     WithAddressDataSource::FromLoaded(..) => AddressSource::Loaded,
-                    WithAddressDataSource::FromEmpty(..) => AddressSource::FromEmpty,
+                    WithAddressDataSource::FromEmpty(_, data) => {
+                        if data.utxo_count() == 0 {
+                            AddressSource::FromEmpty
+                        } else {
+                            AddressSource::Loaded
+                        }
+                    }
                 };
                 (entry.into_mut(), source)
             }

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

@@ -62,13 +62,30 @@ pub fn process_received(
 
                 if AmountBucket::from(prev_balance) != AmountBucket::from(new_balance) {
                     // Crossing cohort boundary - subtract from old, add to new
-                    cohorts
+                    let cohort_state = cohorts
                         .amount_range
                         .get_mut(prev_balance)
                         .state
                         .as_mut()
-                        .unwrap()
-                        .subtract(addr_data);
+                        .unwrap();
+
+                    // Debug info for tracking down underflow issues
+                    if cohort_state.inner.supply.utxo_count < addr_data.utxo_count() as u64 {
+                        panic!(
+                            "process_received: cohort underflow detected!\n\
+                            output_type={:?}, type_index={:?}\n\
+                            prev_balance={}, new_balance={}, total_value={}\n\
+                            Address: {:?}",
+                            output_type,
+                            type_index,
+                            prev_balance,
+                            new_balance,
+                            total_value,
+                            addr_data
+                        );
+                    }
+
+                    cohort_state.subtract(addr_data);
                     addr_data.receive_outputs(total_value, price, output_count);
                     cohorts
                         .amount_range

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

@@ -8,7 +8,7 @@
 use brk_error::Result;
 use brk_grouper::{ByAddressType, Filtered};
 use brk_types::{CheckedSub, Dollars, Height, Sats, Timestamp, TypeIndex};
-use vecdb::VecIndex;
+use vecdb::{VecIndex, unlikely};
 
 use super::super::address::HeightToAddressTypeToVec;
 use super::super::cohorts::AddressCohorts;
@@ -63,13 +63,36 @@ pub fn process_sent(
 
                 if will_be_empty || filters_differ {
                     // Subtract from old cohort
-                    cohorts
+                    let cohort_state = cohorts
                         .amount_range
                         .get_mut(prev_balance)
                         .state
                         .as_mut()
-                        .unwrap()
-                        .subtract(addr_data);
+                        .unwrap();
+
+                    // Debug info for tracking down underflow issues
+                    if unlikely(
+                        cohort_state.inner.supply.utxo_count < addr_data.utxo_count() as u64,
+                    ) {
+                        panic!(
+                            "process_sent: cohort underflow detected!\n\
+                            Block context: prev_height={:?}, output_type={:?}, type_index={:?}\n\
+                            prev_balance={}, new_balance={}, value={}\n\
+                            will_be_empty={}, filters_differ={}\n\
+                            Address: {:?}",
+                            prev_height,
+                            output_type,
+                            type_index,
+                            prev_balance,
+                            new_balance,
+                            value,
+                            will_be_empty,
+                            filters_differ,
+                            addr_data
+                        );
+                    }
+
+                    cohort_state.subtract(addr_data);
 
                     // Update address data
                     addr_data.send(value, prev_price)?;

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

@@ -2,6 +2,7 @@ use std::path::Path;
 
 use brk_error::Result;
 use brk_types::{Dollars, Height, LoadedAddressData, Sats};
+use vecdb::unlikely;
 
 use crate::stateful::states::{RealizedState, SupplyState};
 
@@ -136,11 +137,46 @@ impl AddressCohortState {
     }
 
     pub fn subtract(&mut self, addressdata: &LoadedAddressData) {
-        self.addr_count = self.addr_count.checked_sub(1).unwrap();
+        let addr_supply: SupplyState = addressdata.into();
+        let realized_price = addressdata.realized_price();
+
+        // Check for potential underflow before it happens
+        if unlikely(self.inner.supply.utxo_count < addr_supply.utxo_count) {
+            panic!(
+                "AddressCohortState::subtract underflow!\n\
+                Cohort state: addr_count={}, supply={}\n\
+                Address being subtracted: {}\n\
+                Address supply: {}\n\
+                Realized price: {}\n\
+                This means the address is not properly tracked in this cohort.",
+                self.addr_count, self.inner.supply, addressdata, addr_supply, realized_price
+            );
+        }
+        if unlikely(self.inner.supply.value < addr_supply.value) {
+            panic!(
+                "AddressCohortState::subtract value underflow!\n\
+                Cohort state: addr_count={}, supply={}\n\
+                Address being subtracted: {}\n\
+                Address supply: {}\n\
+                Realized price: {}\n\
+                This means the address is not properly tracked in this cohort.",
+                self.addr_count, self.inner.supply, addressdata, addr_supply, realized_price
+            );
+        }
+
+        self.addr_count = self.addr_count.checked_sub(1).unwrap_or_else(|| {
+            panic!(
+                "AddressCohortState::subtract addr_count underflow! addr_count=0\n\
+                Address being subtracted: {}\n\
+                Realized price: {}",
+                addressdata, realized_price
+            )
+        });
+
         self.inner.decrement_(
-            &addressdata.into(),
+            &addr_supply,
             addressdata.realized_cap,
-            addressdata.realized_price(),
+            realized_price,
         );
     }
 

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

@@ -9,7 +9,7 @@ use brk_types::{Dollars, Height, Sats};
 use derive_deref::{Deref, DerefMut};
 use pco::standalone::{simple_decompress, simpler_compress};
 use serde::{Deserialize, Serialize};
-use vecdb::Bytes;
+use vecdb::{Bytes, unlikely};
 
 use crate::{grouped::PERCENTILES_LEN, utils::OptionExt};
 
@@ -87,6 +87,25 @@ impl PriceToAmount {
 
     pub fn decrement(&mut self, price: Dollars, supply_state: &SupplyState) {
         if let Some(amount) = self.state.um().get_mut(&price) {
+            if unlikely(*amount < supply_state.value) {
+                let amount = *amount;
+                panic!(
+                    "PriceToAmount::decrement underflow!\n\
+                    Path: {:?}\n\
+                    Price: {}\n\
+                    Bucket amount: {}\n\
+                    Trying to decrement by: {}\n\
+                    Supply state: utxo_count={}, value={}\n\
+                    All buckets: {:?}",
+                    self.pathbuf,
+                    price,
+                    amount,
+                    supply_state.value,
+                    supply_state.utxo_count,
+                    supply_state.value,
+                    self.state.u().iter().collect::<Vec<_>>()
+                );
+            }
             *amount -= supply_state.value;
             if *amount == Sats::ZERO {
                 self.state.um().remove(&price);
@@ -95,8 +114,18 @@ impl PriceToAmount {
                 buckets.decrement(price, supply_state.value);
             }
         } else {
-            dbg!(price, &self.pathbuf);
-            unreachable!();
+            panic!(
+                "PriceToAmount::decrement price not found!\n\
+                Path: {:?}\n\
+                Price: {}\n\
+                Supply state: utxo_count={}, value={}\n\
+                All buckets: {:?}",
+                self.pathbuf,
+                price,
+                supply_state.utxo_count,
+                supply_state.value,
+                self.state.u().iter().collect::<Vec<_>>()
+            );
         }
     }
 

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

@@ -39,8 +39,22 @@ impl AddAssign<&SupplyState> for SupplyState {
 
 impl SubAssign<&SupplyState> for SupplyState {
     fn sub_assign(&mut self, rhs: &Self) {
-        self.utxo_count = self.utxo_count.checked_sub(rhs.utxo_count).unwrap();
-        self.value = self.value.checked_sub(rhs.value).unwrap();
+        self.utxo_count = self.utxo_count.checked_sub(rhs.utxo_count).unwrap_or_else(|| {
+            panic!(
+                "SupplyState underflow: cohort utxo_count {} < address utxo_count {}. \
+                This indicates a desync between cohort state and address data. \
+                Try deleting the compute cache and restarting fresh.",
+                self.utxo_count, rhs.utxo_count
+            )
+        });
+        self.value = self.value.checked_sub(rhs.value).unwrap_or_else(|| {
+            panic!(
+                "SupplyState underflow: cohort value {} < address value {}. \
+                This indicates a desync between cohort state and address data. \
+                Try deleting the compute cache and restarting fresh.",
+                self.value, rhs.value
+            )
+        });
     }
 }