Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-07-27 02:38:10 -07:00
Code Issues Packages Projects Releases Wiki Activity

indexer: snapshot

Browse Source
This commit is contained in:
nym21
2025-11-07 15:13:01 +01:00
parent 11b323ef00
commit 3d3787a8d9
12 changed files with 262 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/brk_indexer/src/lib.rs
+4 -4
View File
@@ -18,13 +18,13 @@ use rayon::prelude::*;
use rustc_hash::{FxHashMap, FxHashSet};
use vecdb::{AnyVec, Exit, GenericStoredVec, Reader, VecIteratorExtended};
mod indexes;
// mod stores_v2;
mod stores_v3;
mod stores_v2;
// mod stores_v3;
mod vecs;
pub use indexes::*;
// pub use stores_v2::*;
pub use stores_v3::*;
pub use stores_v2::*;
// pub use stores_v3::*;
pub use vecs::*;
// One version for all data sources
crates/brk_indexer/src/stores_v2.rs
+33 -9
View File
@@ -1,13 +1,13 @@
use std::{fs, path::Path};
use brk_error::Result;
use brk_store::{AnyStore, StoreFjallV2 as Store};
use brk_store::{AnyStore, Mode, StoreFjallV2 as Store, Type};
use brk_types::{
AddressBytes, AddressBytesHash, AddressTypeAddressIndexOutPoint,
AddressTypeAddressIndexTxIndex, BlockHashPrefix, Height, OutPoint, StoredString, TxIndex,
TxOutIndex, TxidPrefix, TypeIndex, Unit, Version, Vout,
};
use fjall2::{PersistMode, TransactionalKeyspace};
use fjall2::{Keyspace, PersistMode};
use rayon::prelude::*;
use vecdb::{AnyVec, GenericStoredVec, StoredIndex, VecIterator, VecIteratorExtended};
@@ -17,7 +17,7 @@ use super::Vecs;
#[derive(Clone)]
pub struct Stores {
pub keyspace: TransactionalKeyspace,
pub keyspace: Keyspace,
pub addressbyteshash_to_typeindex: Store<AddressBytesHash, TypeIndex>,
pub blockhashprefix_to_height: Store<BlockHashPrefix, Height>,
@@ -48,23 +48,47 @@ impl Stores {
Ok(Self {
keyspace: keyspace.clone(),
height_to_coinbase_tag: Store::import(keyspace_ref, path, "h2c", version, true)?,
addressbyteshash_to_typeindex: Store::import(keyspace_ref, path, "a2t", version, true)?,
blockhashprefix_to_height: Store::import(keyspace_ref, path, "b2h", version, true)?,
txidprefix_to_txindex: Store::import(keyspace_ref, path, "t2t", version, true)?,
height_to_coinbase_tag: Store::import(
keyspace_ref,
path,
"h2c",
version,
Mode::UniquePushOnly(Type::Sequential),
)?,
addressbyteshash_to_typeindex: Store::import(
keyspace_ref,
path,
"a2t",
version,
Mode::UniquePushOnly(Type::Random),
)?,
blockhashprefix_to_height: Store::import(
keyspace_ref,
path,
"b2h",
version,
Mode::UniquePushOnly(Type::Random),
)?,
txidprefix_to_txindex: Store::import(
keyspace_ref,
path,
"t2t",
version,
Mode::UniquePushOnly(Type::Random),
)?,
addresstype_to_addressindex_and_txindex: Store::import(
keyspace_ref,
path,
"aat",
version,
false,
Mode::VecLike,
)?,
addresstype_to_addressindex_and_unspentoutpoint: Store::import(
keyspace_ref,
path,
"aau",
version,
false,
Mode::VecLike,
)?,
})
}
crates/brk_indexer/src/stores_v3.rs
+41 -41
View File
@@ -1,7 +1,7 @@
use std::{fs, path::Path};
use brk_error::Result;
use brk_store::{AnyStore, StoreFjallV3 as Store};
use brk_store::{AnyStore, Kind3, Mode3, StoreFjallV3 as Store};
use brk_types::{
AddressBytes, AddressBytesHash, AddressTypeAddressIndexOutPoint,
AddressTypeAddressIndexTxIndex, BlockHashPrefix, Height, OutPoint, StoredString, TxIndex,
@@ -24,8 +24,8 @@ pub struct Stores {
pub height_to_coinbase_tag: Store<Height, StoredString>,
pub txidprefix_to_txindex: Store<TxidPrefix, TxIndex>,
pub addresstype_to_addressindex_and_txindex: Store<AddressTypeAddressIndexTxIndex, Unit>,
pub addresstype_to_addressindex_and_unspentoutpoint:
Store<AddressTypeAddressIndexOutPoint, Unit>,
// pub addresstype_to_addressindex_and_unspentoutpoint:
// Store<AddressTypeAddressIndexOutPoint, Unit>,
}
impl Stores {
@@ -53,49 +53,49 @@ impl Stores {
path,
"height_to_coinbase_tag",
version,
true,
true,
Mode3::PushOnly,
Kind3::Sequential,
)?,
addressbyteshash_to_typeindex: Store::import(
database_ref,
path,
"addressbyteshash_to_typeindex",
version,
true,
false,
Mode3::PushOnly,
Kind3::Random,
)?,
blockhashprefix_to_height: Store::import(
database_ref,
path,
"blockhashprefix_to_height",
version,
true,
false,
Mode3::PushOnly,
Kind3::Random,
)?,
txidprefix_to_txindex: Store::import(
database_ref,
path,
"txidprefix_to_txindex",
version,
true,
false,
Mode3::PushOnly,
Kind3::Random,
)?,
addresstype_to_addressindex_and_txindex: Store::import(
database_ref,
path,
"addresstype_to_addressindex_and_txindex",
version,
false,
false,
)?,
addresstype_to_addressindex_and_unspentoutpoint: Store::import(
database_ref,
path,
"addresstype_to_addressindex_and_unspentoutpoint",
version,
false,
false,
Mode3::PushOnly,
Kind3::Vec,
)?,
// addresstype_to_addressindex_and_unspentoutpoint: Store::import(
// database_ref,
// path,
// "addresstype_to_addressindex_and_unspentoutpoint",
// version,
// Mode3::Any,
// Kind3::Vec,
// )?,
})
}
@@ -117,7 +117,7 @@ impl Stores {
&mut self.height_to_coinbase_tag,
&mut self.txidprefix_to_txindex,
&mut self.addresstype_to_addressindex_and_txindex,
&mut self.addresstype_to_addressindex_and_unspentoutpoint,
// &mut self.addresstype_to_addressindex_and_unspentoutpoint,
]
.into_par_iter() // Changed from par_iter_mut()
.try_for_each(|store| store.commit(height))?;
@@ -134,7 +134,7 @@ impl Stores {
&self.height_to_coinbase_tag,
&self.txidprefix_to_txindex,
&self.addresstype_to_addressindex_and_txindex,
&self.addresstype_to_addressindex_and_unspentoutpoint,
// &self.addresstype_to_addressindex_and_unspentoutpoint,
]
.into_iter()
}
@@ -149,9 +149,9 @@ impl Stores {
&& self.txidprefix_to_txindex.is_empty()?
&& self.height_to_coinbase_tag.is_empty()?
&& self.addresstype_to_addressindex_and_txindex.is_empty()?
&& self
.addresstype_to_addressindex_and_unspentoutpoint
.is_empty()?
// && self
// .addresstype_to_addressindex_and_unspentoutpoint
// .is_empty()?
{
return Ok(());
}
@@ -351,13 +351,13 @@ impl Stores {
);
let outpoint = OutPoint::new(txindex, vout);
self.addresstype_to_addressindex_and_unspentoutpoint.remove(
AddressTypeAddressIndexOutPoint::from((
addresstype,
addressindex,
outpoint,
)),
);
// self.addresstype_to_addressindex_and_unspentoutpoint.remove(
// AddressTypeAddressIndexOutPoint::from((
// addresstype,
// addressindex,
// outpoint,
// )),
// );
});
// Add back outputs that were spent after the rollback point
@@ -394,14 +394,14 @@ impl Stores {
)),
);
self.addresstype_to_addressindex_and_unspentoutpoint.insert(
AddressTypeAddressIndexOutPoint::from((
addresstype,
addressindex,
outpoint,
)),
Unit,
);
// self.addresstype_to_addressindex_and_unspentoutpoint.insert(
// AddressTypeAddressIndexOutPoint::from((
// addresstype,
// addressindex,
// outpoint,
// )),
// Unit,
// );
}
}
});
crates/brk_logger/src/lib.rs
+2 -2
View File
@@ -23,8 +23,8 @@ pub fn init(path: Option<&Path>) -> io::Result<()> {
});
Builder::from_env(Env::default().default_filter_or(
// "info,bitcoin=off,bitcoincore-rpc=off,fjall=off,lsm-tree=off,rolldown=off,rolldown=off,rmcp=off,brk_rmcp=off,tracing=off,aide=off,brk_aide=off",
"debug,bitcoin=off,bitcoincore-rpc=off,rolldown=off,rolldown=off,rmcp=off,brk_rmcp=off,tracing=off,aide=off,brk_aide=off",
"info,bitcoin=off,bitcoincore-rpc=off,fjall=off,lsm-tree=off,rolldown=off,rolldown=off,rmcp=off,brk_rmcp=off,tracing=off,aide=off,brk_aide=off",
// "debug,bitcoin=off,bitcoincore-rpc=off,rolldown=off,rolldown=off,rmcp=off,brk_rmcp=off,tracing=off,aide=off,brk_aide=off",
))
.format(move |buf, record| {
let date_time = Timestamp::now()
crates/brk_store/src/fjall_v2/meta.rs
+4 -4
View File
@@ -5,7 +5,7 @@ use std::{
use brk_error::Result;
use brk_types::Version;
use fjall2::{PersistMode, TransactionalKeyspace, TransactionalPartitionHandle};
use fjall2::{Keyspace, PartitionHandle, PersistMode};
use super::Height;
@@ -18,13 +18,13 @@ pub struct StoreMeta {
impl StoreMeta {
pub fn checked_open<F>(
keyspace: &TransactionalKeyspace,
keyspace: &Keyspace,
path: &Path,
version: Version,
open_partition_handle: F,
) -> Result<(Self, TransactionalPartitionHandle)>
) -> Result<(Self, PartitionHandle)>
where
F: Fn() -> Result<TransactionalPartitionHandle>,
F: Fn() -> Result<PartitionHandle>,
{
fs::create_dir_all(path)?;
crates/brk_store/src/fjall_v2/mod.rs
+55 -25
View File
@@ -3,10 +3,7 @@ use std::{borrow::Cow, cmp, fmt::Debug, fs, hash::Hash, mem, path::Path};
use brk_error::Result;
use brk_types::{Height, Version};
use byteview6::ByteView;
use fjall2::{
InnerItem, PartitionCreateOptions, TransactionalKeyspace, TransactionalPartitionHandle,
ValueType,
};
use fjall2::{InnerItem, Keyspace, PartitionCreateOptions, PartitionHandle, ValueType};
use rustc_hash::{FxHashMap, FxHashSet};
use crate::any::AnyStore;
@@ -19,16 +16,17 @@ use meta::*;
pub struct StoreFjallV2<Key, Value> {
meta: StoreMeta,
name: &'static str,
keyspace: TransactionalKeyspace,
partition: TransactionalPartitionHandle,
keyspace: Keyspace,
partition: PartitionHandle,
puts: FxHashMap<Key, Value>,
dels: FxHashSet<Key>,
mode: Mode,
}
const MAJOR_FJALL_VERSION: Version = Version::TWO;
pub fn open_keyspace(path: &Path) -> fjall2::Result<TransactionalKeyspace> {
fjall2::Config::new(path.join("fjall")).open_transactional()
pub fn open_keyspace(path: &Path) -> fjall2::Result<Keyspace> {
fjall2::Config::new(path.join("fjall")).open()
}
impl<K, V> StoreFjallV2<K, V>
@@ -38,14 +36,14 @@ where
ByteView: From<K> + From<V>,
{
fn open_partition_handle(
keyspace: &TransactionalKeyspace,
keyspace: &Keyspace,
name: &str,
bloom_filters: bool,
) -> Result<TransactionalPartitionHandle> {
mode: Mode,
) -> Result<PartitionHandle> {
let mut options = PartitionCreateOptions::default().manual_journal_persist(true);
if bloom_filters {
options = options.bloom_filter_bits(Some(5));
if mode.is_unique_push_only() {
options = options.bloom_filter_bits(Some(7));
} else {
options = options.bloom_filter_bits(None);
}
@@ -54,11 +52,11 @@ where
}
pub fn import(
keyspace: &TransactionalKeyspace,
keyspace: &Keyspace,
path: &Path,
name: &str,
version: Version,
bloom_filters: bool,
mode: Mode,
) -> Result<Self> {
fs::create_dir_all(path)?;
@@ -67,7 +65,7 @@ where
&path.join(format!("meta/{name}")),
MAJOR_FJALL_VERSION + version,
|| {
Self::open_partition_handle(keyspace, name, bloom_filters).inspect_err(|e| {
Self::open_partition_handle(keyspace, name, mode).inspect_err(|e| {
eprintln!("{e}");
eprintln!("Delete {path:?} and try again");
})
@@ -81,6 +79,7 @@ where
partition,
puts: FxHashMap::default(),
dels: FxHashSet::default(),
mode,
})
}
@@ -99,16 +98,12 @@ where
}
pub fn is_empty(&self) -> Result<bool> {
self.keyspace
.read_tx()
.is_empty(&self.partition)
.map_err(|e| e.into())
self.partition.is_empty().map_err(|e| e.into())
}
pub fn iter(&self) -> impl Iterator<Item = (K, V)> {
self.keyspace
.read_tx()
.iter(&self.partition)
self.partition
.iter()
.map(|res| res.unwrap())
.map(|(k, v)| (K::from(ByteView::from(&*k)), V::from(ByteView::from(&*v))))
}
@@ -177,6 +172,18 @@ where
return Ok(());
}
// if self.mode.is_unique_push_only() {
// if !self.dels.is_empty() {
// unreachable!();
// }
// let mut puts = mem::take(&mut self.puts).into_iter().collect::<Vec<_>>();
// puts.sort_unstable_by(|(k1, _), (k2, _)| k1.cmp(k2));
// dbg!(&puts);
// self.partition.ingest(
// puts.into_iter()
// .map(|(k, v)| (ByteView::from(k), ByteView::from(v))),
// )?;
// } else {
let mut items = mem::take(&mut self.puts)
.into_iter()
.map(|(key, value)| Item::Value { key, value })
@@ -188,10 +195,11 @@ where
.collect::<Vec<_>>();
items.sort_unstable();
self.keyspace.inner().batch().commit_partition(
self.partition.inner(),
self.keyspace.batch().commit_partition(
&self.partition,
items.into_iter().map(InnerItem::from).collect::<Vec<_>>(),
)?;
// }
Ok(())
}
@@ -271,3 +279,25 @@ where
}
}
}
#[derive(Debug, Clone, Copy)]
pub enum Mode {
VecLike,
UniquePushOnly(Type),
}
#[derive(Debug, Clone, Copy)]
pub enum Type {
Random,
Sequential,
}
impl Mode {
pub fn is_vec_like(&self) -> bool {
matches!(*self, Self::VecLike)
}
pub fn is_unique_push_only(&self) -> bool {
matches!(*self, Self::UniquePushOnly(_))
}
}
crates/brk_store/src/fjall_v3/mod.rs
+82 -36
View File
@@ -26,6 +26,8 @@ pub struct StoreFjallV3<Key, Value> {
keyspace: Keyspace,
puts: FxHashMap<Key, Value>,
dels: FxHashSet<Key>,
mode: Mode3,
kind: Kind3,
}
const MAJOR_FJALL_VERSION: Version = Version::new(3);
@@ -47,8 +49,8 @@ where
path: &Path,
name: &str,
version: Version,
bloom_filters: bool,
sequential: bool,
mode: Mode3,
kind: Kind3,
) -> Result<Self> {
fs::create_dir_all(path)?;
@@ -57,7 +59,7 @@ where
&path.join(format!("meta/{name}")),
MAJOR_FJALL_VERSION + version,
|| {
Self::open_keyspace(database, name, bloom_filters, sequential).inspect_err(|e| {
Self::open_keyspace(database, name, mode, kind).inspect_err(|e| {
eprintln!("{e}");
eprintln!("Delete {path:?} and try again");
})
@@ -71,29 +73,31 @@ where
keyspace,
puts: FxHashMap::default(),
dels: FxHashSet::default(),
mode,
kind,
})
}
fn open_keyspace(
database: &Database,
name: &str,
bloom_filters: bool,
sequential: bool,
_mode: Mode3,
kind: Kind3,
) -> Result<Keyspace> {
let mut options = KeyspaceCreateOptions::default().manual_journal_persist(true);
if bloom_filters {
if kind.is_not_vec() {
options = options.filter_policy(FilterPolicy::new(&[
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(10.0)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(7.5)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(5.0)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(2.5)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(10.0)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(8.0)),
FilterPolicyEntry::Bloom(BloomConstructionPolicy::BitsPerKey(7.0)),
]));
} else {
options = options.filter_policy(FilterPolicy::disabled());
}
if sequential {
if kind.is_sequential() {
options = options
.filter_block_pinning_policy(PinningPolicy::all(false))
.index_block_pinning_policy(PinningPolicy::all(false));
@@ -109,11 +113,6 @@ where
{
if let Some(v) = self.puts.get(key) {
Ok(Some(Cow::Borrowed(v)))
// } else if let Some(slice) = self
// .database
// .read_tx()
// .get(&self.keyspace, ByteView::from(key))?
// {
} else if let Some(slice) = self.keyspace.get(ByteView::from(key))? {
Ok(Some(Cow::Owned(V::from(ByteView::from(slice)))))
} else {
@@ -123,9 +122,6 @@ where
#[inline]
pub fn is_empty(&self) -> Result<bool> {
// self.database
// .read_tx()
// .is_empty(&self.keyspace)
self.keyspace.is_empty().map_err(|e| e.into())
}
@@ -189,25 +185,38 @@ where
return Ok(());
}
let mut batch = self.database.batch();
// let mut batch = self.database.inner().batch();
let mut items = mem::take(&mut self.puts)
.into_iter()
.map(|(key, value)| Item::Value { key, value })
.chain(
mem::take(&mut self.dels)
.into_iter()
.map(|key| Item::Tomb(key)),
)
.collect::<Vec<_>>();
items.sort_unstable();
items.into_iter().for_each(|item| match item {
Item::Value { key, value } => {
batch.insert(&self.keyspace, ByteView::from(key), ByteView::from(value))
if self.mode.is_push_only() {
if !self.dels.is_empty() {
unreachable!();
}
Item::Tomb(key) => batch.remove(&self.keyspace, ByteView::from(key)),
});
batch.commit()?;
let mut puts = mem::take(&mut self.puts).into_iter().collect::<Vec<_>>();
puts.sort_unstable_by(|(k1, _), (k2, _)| k1.cmp(k2));
// dbg!(&puts);
self.keyspace.ingest(
puts.into_iter()
.map(|(k, v)| (ByteView::from(k), ByteView::from(v))),
)?;
} else {
let mut batch = self.database.batch();
// let mut batch = self.database.inner().batch();
let mut items = mem::take(&mut self.puts)
.into_iter()
.map(|(key, value)| Item::Value { key, value })
.chain(
mem::take(&mut self.dels)
.into_iter()
.map(|key| Item::Tomb(key)),
)
.collect::<Vec<_>>();
items.sort_unstable();
items.into_iter().for_each(|item| match item {
Item::Value { key, value } => {
batch.insert(&self.keyspace, ByteView::from(key), ByteView::from(value))
}
Item::Tomb(key) => batch.remove(&self.keyspace, ByteView::from(key)),
});
batch.commit()?;
}
// batch.ingest(
// items
@@ -304,3 +313,40 @@ impl<K, V> Item<K, V> {
// }
// }
}
#[derive(Debug, Clone, Copy)]
pub enum Mode3 {
Any,
PushOnly,
}
impl Mode3 {
pub fn is_any(&self) -> bool {
matches!(*self, Self::Any)
}
pub fn is_push_only(&self) -> bool {
matches!(*self, Self::PushOnly)
}
}
#[derive(Debug, Clone, Copy)]
pub enum Kind3 {
Random,
Sequential,
Vec,
}
impl Kind3 {
pub fn is_sequential(&self) -> bool {
matches!(*self, Self::Sequential)
}
pub fn is_random(&self) -> bool {
matches!(*self, Self::Random)
}
pub fn is_not_vec(&self) -> bool {
!matches!(*self, Self::Vec)
}
}
crates/brk_types/src/addressbyteshash.rs
+4 -2
View File
@@ -2,6 +2,8 @@ use byteview::ByteView;
use derive_deref::Deref;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
use crate::copy_first_8bytes;
use super::AddressBytes;
#[derive(
@@ -31,7 +33,7 @@ impl From<&AddressBytes> for AddressBytesHash {
impl From<ByteView> for AddressBytesHash {
#[inline]
fn from(value: ByteView) -> Self {
Self::read_from_bytes(&value).unwrap()
Self(u64::from_be_bytes(copy_first_8bytes(&value).unwrap()))
}
}
@@ -45,6 +47,6 @@ impl From<AddressBytesHash> for ByteView {
impl From<&AddressBytesHash> for ByteView {
#[inline]
fn from(value: &AddressBytesHash) -> Self {
Self::new(value.as_bytes())
Self::new(&value.0.to_be_bytes())
}
}
crates/brk_types/src/blockhashprefix.rs
+8 -8
View File
@@ -40,14 +40,7 @@ impl From<&BlockHash> for BlockHashPrefix {
impl From<ByteView> for BlockHashPrefix {
#[inline]
fn from(value: ByteView) -> Self {
Self::read_from_bytes(&value).unwrap()
}
}
impl From<&BlockHashPrefix> for ByteView {
#[inline]
fn from(value: &BlockHashPrefix) -> Self {
Self::new(value.as_bytes())
Self(u64::from_be_bytes(copy_first_8bytes(&value).unwrap()))
}
}
@@ -57,3 +50,10 @@ impl From<BlockHashPrefix> for ByteView {
Self::from(&value)
}
}
impl From<&BlockHashPrefix> for ByteView {
#[inline]
fn from(value: &BlockHashPrefix) -> Self {
Self::from(value.to_be_bytes())
}
}
crates/brk_types/src/txidprefix.rs
+8 -8
View File
@@ -40,14 +40,7 @@ impl From<&Txid> for TxidPrefix {
impl From<ByteView> for TxidPrefix {
#[inline]
fn from(value: ByteView) -> Self {
Self::read_from_bytes(&value).unwrap()
}
}
impl From<&TxidPrefix> for ByteView {
#[inline]
fn from(value: &TxidPrefix) -> Self {
Self::new(value.as_bytes())
Self(u64::from_be_bytes(copy_first_8bytes(&value).unwrap()))
}
}
@@ -58,6 +51,13 @@ impl From<TxidPrefix> for ByteView {
}
}
impl From<&TxidPrefix> for ByteView {
#[inline]
fn from(value: &TxidPrefix) -> Self {
Self::from(value.to_be_bytes())
}
}
impl From<[u8; 8]> for TxidPrefix {
#[inline]
fn from(value: [u8; 8]) -> Self {