global: big snapshot

crates/brk_computer/src/internal/by_type_counts.rs

@@ -0,0 +1,125 @@
+//! Shared per-block per-address-type counters.
+//!
+//! Used by `outputs/by_type/` (counts outputs per type) and `inputs/by_type/`
+//! (counts inputs per type). Walks each block's tx range, calls a scanner
+//! callback that fills a `[u32; 12]` per-tx counter, and produces two
+//! per-block aggregates in a single pass:
+//!
+//! - `entry_count` — total number of items (outputs / inputs) per type
+//! - `tx_count`    — number of txs that contain at least one item of each type
+
+use brk_cohort::ByAddrType;
+use brk_error::Result;
+use brk_types::{BasisPoints16, Height, Indexes, OutputType, StoredU64, TxIndex};
+use vecdb::{AnyStoredVec, Exit, VecIndex, WritableVec};
+
+use crate::internal::{
+    PerBlockCumulativeRolling, PerBlockFull, PercentCumulativeRolling, RatioU64Bp16,
+};
+
+/// Per-block scan that simultaneously computes:
+/// - `entry_count[type] += per_tx[type]` (sum of items)
+/// - `tx_count[type] += 1 if per_tx[type] > 0` (presence flag)
+///
+/// `scan_tx` is called once per tx with a zeroed `[u32; 12]` buffer that
+/// it must fill with the per-type item count for that tx.
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn compute_by_addr_type_block_counts(
+    entry_count: &mut ByAddrType<PerBlockCumulativeRolling<StoredU64, StoredU64>>,
+    tx_count: &mut ByAddrType<PerBlockCumulativeRolling<StoredU64, StoredU64>>,
+    fi_batch: &[TxIndex],
+    txid_len: usize,
+    skip_first_tx: bool,
+    starting_height: Height,
+    exit: &Exit,
+    mut scan_tx: impl FnMut(usize, &mut [u32; 12]) -> Result<()>,
+) -> Result<()> {
+    for (j, first_tx) in fi_batch.iter().enumerate() {
+        let fi = first_tx.to_usize();
+        let next_fi = fi_batch
+            .get(j + 1)
+            .map(|v| v.to_usize())
+            .unwrap_or(txid_len);
+
+        let start_tx = if skip_first_tx { fi + 1 } else { fi };
+
+        let mut entries_per_block = [0u64; 12];
+        let mut txs_per_block = [0u64; 12];
+
+        for tx_pos in start_tx..next_fi {
+            let mut per_tx = [0u32; 12];
+            scan_tx(tx_pos, &mut per_tx)?;
+            for (i, &n) in per_tx.iter().enumerate() {
+                if n > 0 {
+                    entries_per_block[i] += u64::from(n);
+                    txs_per_block[i] += 1;
+                }
+            }
+        }
+
+        for otype in OutputType::ADDR_TYPES {
+            let idx = otype as usize;
+            entry_count
+                .get_mut_unwrap(otype)
+                .block
+                .push(StoredU64::from(entries_per_block[idx]));
+            tx_count
+                .get_mut_unwrap(otype)
+                .block
+                .push(StoredU64::from(txs_per_block[idx]));
+        }
+
+        if entry_count.p2pkh.block.batch_limit_reached() {
+            let _lock = exit.lock();
+            for (_, v) in entry_count.iter_mut() {
+                v.block.write()?;
+            }
+            for (_, v) in tx_count.iter_mut() {
+                v.block.write()?;
+            }
+        }
+    }
+
+    {
+        let _lock = exit.lock();
+        for (_, v) in entry_count.iter_mut() {
+            v.block.write()?;
+        }
+        for (_, v) in tx_count.iter_mut() {
+            v.block.write()?;
+        }
+    }
+
+    for (_, v) in entry_count.iter_mut() {
+        v.compute_rest(starting_height, exit)?;
+    }
+    for (_, v) in tx_count.iter_mut() {
+        v.compute_rest(starting_height, exit)?;
+    }
+
+    Ok(())
+}
+
+/// Compute per-type tx-count percent over total tx count, for all 8 address types.
+pub(crate) fn compute_by_addr_type_tx_percents(
+    tx_count: &ByAddrType<PerBlockCumulativeRolling<StoredU64, StoredU64>>,
+    tx_percent: &mut ByAddrType<PercentCumulativeRolling<BasisPoints16>>,
+    count_total: &PerBlockFull<StoredU64>,
+    starting_indexes: &Indexes,
+    exit: &Exit,
+) -> Result<()> {
+    for otype in OutputType::ADDR_TYPES {
+        let source = tx_count.get_unwrap(otype);
+        tx_percent
+            .get_mut_unwrap(otype)
+            .compute_binary::<StoredU64, StoredU64, RatioU64Bp16, _, _, _, _>(
+                starting_indexes.height,
+                &source.cumulative.height,
+                &count_total.cumulative.height,
+                source.sum.as_array().map(|w| &w.height),
+                count_total.rolling.sum.as_array().map(|w| &w.height),
+                exit,
+            )?;
+    }
+    Ok(())
+}

crates/brk_computer/src/internal/mod.rs

@@ -1,5 +1,6 @@
 pub(crate) mod algo;
 mod amount;
+mod by_type_counts;
 mod cache_budget;
 mod containers;
 pub(crate) mod db_utils;
@@ -10,6 +11,7 @@ mod traits;
 mod transform;
 
 pub(crate) use amount::*;
+pub(crate) use by_type_counts::*;
 pub(crate) use cache_budget::*;
 pub(crate) use containers::*;
 pub(crate) use indexes::*;

crates/brk_computer/src/internal/per_block/computed/distribution.rs

@@ -2,7 +2,7 @@ use std::collections::VecDeque;
 
 use brk_error::Result;
 use brk_traversable::Traversable;
-use brk_types::{Height, get_percentile};
+use brk_types::{Height, VSize, get_percentile, get_weighted_percentile};
 use derive_more::{Deref, DerefMut};
 use schemars::JsonSchema;
 use vecdb::{
@@ -154,6 +154,141 @@ impl<T: NumericValue + JsonSchema> PerBlockDistribution<T> {
         Ok(())
     }
 
+    /// Like `compute_with_skip` but uses vsize-weighted percentiles.
+    /// Each transaction's contribution to percentile rank is proportional to its vsize.
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn compute_with_skip_weighted<A>(
+        &mut self,
+        max_from: Height,
+        source: &impl ReadableVec<A, T>,
+        vsize_source: &impl ReadableVec<A, VSize>,
+        first_indexes: &impl ReadableVec<Height, A>,
+        count_indexes: &impl ReadableVec<Height, brk_types::StoredU64>,
+        exit: &Exit,
+        skip_count: usize,
+    ) -> Result<()>
+    where
+        A: VecIndex + VecValue + brk_types::CheckedSub<A>,
+    {
+        let DistributionStats {
+            min,
+            max,
+            pct10,
+            pct25,
+            median,
+            pct75,
+            pct90,
+        } = &mut self.0;
+
+        let min = &mut min.height;
+        let max = &mut max.height;
+        let pct10 = &mut pct10.height;
+        let pct25 = &mut pct25.height;
+        let median = &mut median.height;
+        let pct75 = &mut pct75.height;
+        let pct90 = &mut pct90.height;
+
+        let combined_version = source.version()
+            + vsize_source.version()
+            + first_indexes.version()
+            + count_indexes.version();
+
+        let mut index = max_from;
+        for vec in [
+            &mut *min,
+            &mut *max,
+            &mut *median,
+            &mut *pct10,
+            &mut *pct25,
+            &mut *pct75,
+            &mut *pct90,
+        ] {
+            vec.validate_computed_version_or_reset(combined_version)?;
+            index = index.min(Height::from(vec.len()));
+        }
+
+        let start = index.to_usize();
+
+        for vec in [
+            &mut *min,
+            &mut *max,
+            &mut *median,
+            &mut *pct10,
+            &mut *pct25,
+            &mut *pct75,
+            &mut *pct90,
+        ] {
+            vec.truncate_if_needed_at(start)?;
+        }
+
+        let fi_len = first_indexes.len();
+        let first_indexes_batch: Vec<A> = first_indexes.collect_range_at(start, fi_len);
+        let count_indexes_batch: Vec<brk_types::StoredU64> =
+            count_indexes.collect_range_at(start, fi_len);
+
+        let zero = T::from(0_usize);
+        let mut values: Vec<T> = Vec::new();
+        let mut vsizes: Vec<VSize> = Vec::new();
+        let mut weighted: Vec<(T, VSize)> = Vec::new();
+
+        first_indexes_batch
+            .into_iter()
+            .zip(count_indexes_batch)
+            .try_for_each(|(first_index, count_index)| -> Result<()> {
+                let count = u64::from(count_index) as usize;
+                let effective_count = count.saturating_sub(skip_count);
+                let effective_first_index = first_index + skip_count.min(count);
+
+                let start_at = effective_first_index.to_usize();
+                let end_at = start_at + effective_count;
+
+                source.collect_range_into_at(start_at, end_at, &mut values);
+                vsize_source.collect_range_into_at(start_at, end_at, &mut vsizes);
+
+                weighted.clear();
+                weighted.extend(
+                    values
+                        .iter()
+                        .copied()
+                        .zip(vsizes.iter().copied())
+                        .filter(|(v, _)| skip_count == 0 || *v > zero),
+                );
+
+                if weighted.is_empty() {
+                    for vec in [
+                        &mut *min,
+                        &mut *max,
+                        &mut *median,
+                        &mut *pct10,
+                        &mut *pct25,
+                        &mut *pct75,
+                        &mut *pct90,
+                    ] {
+                        vec.push(zero);
+                    }
+                } else {
+                    weighted.sort_unstable_by(|a, b| a.0.cmp(&b.0));
+
+                    max.push(weighted.last().unwrap().0);
+                    pct90.push(get_weighted_percentile(&weighted, 0.90));
+                    pct75.push(get_weighted_percentile(&weighted, 0.75));
+                    median.push(get_weighted_percentile(&weighted, 0.50));
+                    pct25.push(get_weighted_percentile(&weighted, 0.25));
+                    pct10.push(get_weighted_percentile(&weighted, 0.10));
+                    min.push(weighted.first().unwrap().0);
+                }
+
+                Ok(())
+            })?;
+
+        let _lock = exit.lock();
+        for vec in [min, max, median, pct10, pct25, pct75, pct90] {
+            vec.write()?;
+        }
+
+        Ok(())
+    }
+
     pub(crate) fn compute_from_nblocks<A>(
         &mut self,
         max_from: Height,

crates/brk_computer/src/internal/per_block/percent/cumulative_rolling.rs

@@ -0,0 +1,91 @@
+//! PercentCumulativeRolling - cumulative percent + 4 rolling window percents.
+//!
+//! Mirrors `PerBlockCumulativeRolling` but for percentages derived from ratios
+//! of cumulative values and rolling sums.
+
+use brk_error::Result;
+use brk_traversable::Traversable;
+use brk_types::{Height, Version};
+use vecdb::{BinaryTransform, Database, Exit, ReadableVec, Rw, StorageMode, VecValue};
+
+use crate::{
+    indexes,
+    internal::{BpsType, PercentPerBlock, PercentRollingWindows},
+};
+
+#[derive(Traversable)]
+pub struct PercentCumulativeRolling<B: BpsType, M: StorageMode = Rw> {
+    pub cumulative: PercentPerBlock<B, M>,
+    #[traversable(flatten)]
+    pub rolling: PercentRollingWindows<B, M>,
+}
+
+impl<B: BpsType> PercentCumulativeRolling<B> {
+    pub(crate) fn forced_import(
+        db: &Database,
+        name: &str,
+        version: Version,
+        indexes: &indexes::Vecs,
+    ) -> Result<Self> {
+        let cumulative =
+            PercentPerBlock::forced_import(db, &format!("{name}_cumulative"), version, indexes)?;
+        let rolling =
+            PercentRollingWindows::forced_import(db, &format!("{name}_sum"), version, indexes)?;
+        Ok(Self {
+            cumulative,
+            rolling,
+        })
+    }
+
+    /// Alternate constructor that uses the same base name for both the
+    /// cumulative `PercentPerBlock` and the `PercentRollingWindows`, relying on
+    /// the window suffix to disambiguate. Useful for preserving legacy disk
+    /// names where the two variants historically shared a prefix.
+    pub(crate) fn forced_import_flat(
+        db: &Database,
+        name: &str,
+        version: Version,
+        indexes: &indexes::Vecs,
+    ) -> Result<Self> {
+        let cumulative = PercentPerBlock::forced_import(db, name, version, indexes)?;
+        let rolling = PercentRollingWindows::forced_import(db, name, version, indexes)?;
+        Ok(Self {
+            cumulative,
+            rolling,
+        })
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn compute_binary<S1T, S2T, F, Rc1, Rc2, Rw1, Rw2>(
+        &mut self,
+        max_from: Height,
+        cumulative_numerator: &Rc1,
+        cumulative_denominator: &Rc2,
+        rolling_numerators: [&Rw1; 4],
+        rolling_denominators: [&Rw2; 4],
+        exit: &Exit,
+    ) -> Result<()>
+    where
+        S1T: VecValue,
+        S2T: VecValue,
+        Rc1: ReadableVec<Height, S1T>,
+        Rc2: ReadableVec<Height, S2T>,
+        Rw1: ReadableVec<Height, S1T>,
+        Rw2: ReadableVec<Height, S2T>,
+        F: BinaryTransform<S1T, S2T, B>,
+    {
+        self.cumulative.compute_binary::<S1T, S2T, F>(
+            max_from,
+            cumulative_numerator,
+            cumulative_denominator,
+            exit,
+        )?;
+        self.rolling.compute_binary::<S1T, S2T, F, Rw1, Rw2>(
+            max_from,
+            rolling_numerators,
+            rolling_denominators,
+            exit,
+        )?;
+        Ok(())
+    }
+}

crates/brk_computer/src/internal/per_block/percent/mod.rs

@@ -1,10 +1,12 @@
 mod base;
+mod cumulative_rolling;
 mod lazy;
 mod lazy_windows;
 mod vec;
 mod windows;
 
 pub use base::*;
+pub use cumulative_rolling::*;
 pub use lazy::*;
 pub use lazy_windows::*;
 pub use vec::*;

crates/brk_computer/src/internal/per_tx/derived.rs

@@ -2,7 +2,7 @@ use brk_error::Result;
 use brk_indexer::Indexer;
 
 use brk_traversable::Traversable;
-use brk_types::{Indexes, TxIndex};
+use brk_types::{Indexes, TxIndex, VSize};
 use schemars::JsonSchema;
 use vecdb::{Database, Exit, ReadableVec, Rw, StorageMode, Version};
 
@@ -113,4 +113,43 @@ where
 
         Ok(())
     }
+
+    /// Like `derive_from_with_skip` but uses vsize-weighted percentiles for the
+    /// per-block distribution. The rolling 6-block distribution stays count-based.
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn derive_from_with_skip_weighted(
+        &mut self,
+        indexer: &Indexer,
+        indexes: &indexes::Vecs,
+        starting_indexes: &Indexes,
+        tx_index_source: &impl ReadableVec<TxIndex, T>,
+        vsize_source: &impl ReadableVec<TxIndex, VSize>,
+        exit: &Exit,
+        skip_count: usize,
+    ) -> Result<()>
+    where
+        T: Copy + Ord + From<f64> + Default,
+        f64: From<T>,
+    {
+        self.block.compute_with_skip_weighted(
+            starting_indexes.height,
+            tx_index_source,
+            vsize_source,
+            &indexer.vecs.transactions.first_tx_index,
+            &indexes.height.tx_index_count,
+            exit,
+            skip_count,
+        )?;
+
+        self.distribution._6b.compute_from_nblocks(
+            starting_indexes.height,
+            tx_index_source,
+            &indexer.vecs.transactions.first_tx_index,
+            &indexes.height.tx_index_count,
+            6,
+            exit,
+        )?;
+
+        Ok(())
+    }
 }

crates/brk_computer/src/internal/per_tx/distribution.rs

@@ -6,9 +6,9 @@
 use brk_error::Result;
 use brk_indexer::Indexer;
 use brk_traversable::Traversable;
-use brk_types::{Indexes, TxIndex};
+use brk_types::{Indexes, TxIndex, VSize};
 use schemars::JsonSchema;
-use vecdb::{Database, EagerVec, Exit, ImportableVec, PcoVec, Rw, StorageMode, Version};
+use vecdb::{Database, EagerVec, Exit, ImportableVec, PcoVec, ReadableVec, Rw, StorageMode, Version};
 
 use crate::{
     indexes,
@@ -65,4 +65,29 @@ where
             skip_count,
         )
     }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn derive_from_with_skip_weighted(
+        &mut self,
+        indexer: &Indexer,
+        indexes: &indexes::Vecs,
+        starting_indexes: &Indexes,
+        vsize_source: &impl ReadableVec<TxIndex, VSize>,
+        exit: &Exit,
+        skip_count: usize,
+    ) -> Result<()>
+    where
+        T: Copy + Ord + From<f64> + Default,
+        f64: From<T>,
+    {
+        self.distribution.derive_from_with_skip_weighted(
+            indexer,
+            indexes,
+            starting_indexes,
+            &self.tx_index,
+            vsize_source,
+            exit,
+            skip_count,
+        )
+    }
 }