From 2dbf9d4e86a983e3b10e516c6990ac83a9bbea97 Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Tue, 18 Aug 2026 12:09:01 +0200 Subject: [PATCH 1/2] fix(minibf): fix ordering and 404 edge cases in account addresses The /accounts/{stake_address}/addresses endpoint diverged from Blockfrost in two cases: - order=desc sorted addresses by their latest on-chain appearance. Blockfrost returns the exact reverse of the asc list, which orders addresses by first appearance. Reused addresses came out in the wrong position. - Accounts that only appear inside pool registrations (reward account or pool owner) returned 404. Blockfrost knows these credentials and returns an empty list. Fixes #1140 --- crates/minibf/src/routes/accounts.rs | 204 ++++++++++++++++++++++----- 1 file changed, 166 insertions(+), 38 deletions(-) diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index fff0281da..b87a3306f 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -22,7 +22,7 @@ use blockfrost_openapi::models::{ use dolos_cardano::{ indexes::{AsyncCardanoQueryExt, CardanoIndexExt, SlotOrder}, - model::{AccountState, DRepState}, + model::{AccountState, DRepState, PoolState}, pallas_extras, ChainSummary, FixedNamespace, LeaderRewardLog, MemberRewardLog, PoolDepositRefundLog, }; @@ -229,6 +229,61 @@ where Ok(Json(model)) } +/// Tell if any pool registration names the account as reward account or +/// pool owner. +/// +/// Blockfrost treats these credentials as known accounts even when they +/// never appear in an address or certificate. The scan runs only on the +/// 404 path, so the full pool iteration stays off the hot path. +fn account_appears_in_pool_registrations( + domain: &Facade, + account: &StakeAddress, +) -> Result +where + Option: From, + D: Domain + Clone + Send + Sync + 'static, +{ + let reward_account = account.to_vec(); + + // Pool owners are always key hashes, so a script account can only + // match through the reward account. + let owner_hash = match account.payload() { + StakePayload::Stake(hash) => Some(*hash), + StakePayload::Script(_) => None, + }; + + for item in domain.iter_cardano_entities::(None)? { + let (_, pool) = item.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + // Blockfrost knows a credential from the moment its certificate + // lands on chain and never forgets it. Check every snapshot slot + // the entity still holds to get as close as the state allows: + // - `live` is the only slot a brand-new pool writes. + // - `next` holds a mid-epoch re-registration until the boundary. + // - `mark`/`set`/`go` still hold a credential that a recent re-registration + // replaced. + let snapshots = [ + pool.snapshot.live(), + pool.snapshot.next(), + pool.snapshot.mark(), + pool.snapshot.set(), + pool.snapshot.go(), + ]; + + for snapshot in snapshots.into_iter().flatten() { + if snapshot.params.reward_account == reward_account { + return Ok(true); + } + + if owner_hash.is_some_and(|hash| snapshot.params.pool_owners.contains(&hash)) { + return Ok(true); + } + } + } + + Ok(false) +} + pub async fn by_stake_addresses( Path(stake_address): Path, Query(params): Query, @@ -236,35 +291,48 @@ pub async fn by_stake_addresses( ) -> Result>, Error> where Option: From, + Option: From, D: Domain + Clone + Send + Sync + 'static, { let pagination = Pagination::try_from(params)?; pagination.enforce_max_scan_limit(domain.config.max_scan_items())?; let network = domain.get_network_id()?; let account_key = parse_account_key_param(&stake_address, network)?; - if !domain.cardano_entity_exists::(account_key.entity_key.as_slice())? { + + if !domain.cardano_entity_exists::(account_key.entity_key.as_slice())? + && !account_appears_in_pool_registrations(&domain, &account_key.address)? + { return Err(StatusCode::NOT_FOUND.into()); } let (start_slot, end_slot) = pagination.start_and_end_slots(&domain).await?; + + // Blockfrost orders addresses by first on-chain appearance, and `desc` + // is the exact reverse of the `asc` list. Scan ascending in both cases; + // a descending scan would order reused addresses by their latest + // appearance instead of their first one. let stream = domain.query().blocks_by_stake_stream( &account_key.address.to_vec(), start_slot, end_slot, - SlotOrder::from(pagination.order), + SlotOrder::Asc, ); - let mut items = vec![]; - let mut skipped = 0; + // `asc` can stop once the requested page is full. `desc` needs the + // complete list before the reversal. + let scan_target = match pagination.order { + Order::Asc => Some(pagination.to()), + Order::Desc => None, + }; + + let account = account_key.address.to_vec(); + + let mut ordered = vec![]; let mut seen = BTreeSet::new(); let mut stream = Box::pin(stream); - while let Some(res) = stream.next().await { - if items.len() >= pagination.count { - break; - } - + 'scan: while let Some(res) = stream.next().await { let (_slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let Some(block) = block else { @@ -272,37 +340,39 @@ where }; let block = MultiEraBlock::decode(&block).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + for (_, utxo) in block.txs().iter().flat_map(|tx| tx.produces()) { let address = utxo .address() .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - if match &address { - Address::Shelley(shelley) => { - pallas_extras::shelley_address_to_stake_address(shelley) - .map(|x| x.to_vec() == account_key.address.to_vec()) - .unwrap_or(false) - } - Address::Stake(stake) => stake.to_vec() == account_key.address.to_vec(), - Address::Byron(_) => false, - } && seen.insert(address.to_string()) - { - if skipped < (pagination.page as usize - 1) * pagination.count { - skipped += 1; - } else { - items.push(AccountAddressesContentInner { - address: address.to_string(), - }); - if items.len() >= pagination.count { - break; - } + + if !address_belongs_to_account(&address, &account) { + continue; + } + + let address = address.to_string(); + + if seen.insert(address.clone()) { + ordered.push(address); + + if scan_target.is_some_and(|target| ordered.len() >= target) { + break 'scan; } } } - if items.len() >= pagination.count { - break; - } } + if matches!(pagination.order, Order::Desc) { + ordered.reverse(); + } + + let items = ordered + .into_iter() + .skip(pagination.skip()) + .take(pagination.count) + .map(|address| AccountAddressesContentInner { address }) + .collect(); + Ok(Json(items)) } @@ -1551,15 +1621,47 @@ mod tests { async fn accounts_by_stake_addresses_order_desc() { let app = TestApp::new(); let stake_address = app.vectors().stake_address.as_str(); - let path = format!("/accounts/{stake_address}/addresses?order=desc&count=5"); - let (status, bytes) = app.get_bytes(&path).await; + + let (status, bytes) = app + .get_bytes(&format!( + "/accounts/{stake_address}/addresses?order=asc&count=100" + )) + .await; assert_eq!(status, StatusCode::OK); + let asc: Vec = + serde_json::from_slice(&bytes).expect("failed to parse addresses asc"); + let (status, bytes) = app + .get_bytes(&format!( + "/accounts/{stake_address}/addresses?order=desc&count=100" + )) + .await; + assert_eq!(status, StatusCode::OK); let desc: Vec = serde_json::from_slice(&bytes).expect("failed to parse addresses desc"); - if desc.is_empty() { - return; - } + + assert!(!asc.is_empty()); + + // The synthetic chain reuses the primary address in every block. A + // last-appearance ordering would move that address to the front of + // `desc`. Blockfrost defines `desc` as the reverse of `asc`. + let mut reversed: Vec<_> = asc.iter().map(|x| x.address.clone()).collect(); + reversed.reverse(); + let desc_addresses: Vec<_> = desc.iter().map(|x| x.address.clone()).collect(); + assert_eq!(desc_addresses, reversed); + + // A `desc` page must be a window into the reversed list. + let (status, bytes) = app + .get_bytes(&format!( + "/accounts/{stake_address}/addresses?order=desc&count=2&page=2" + )) + .await; + assert_eq!(status, StatusCode::OK); + let page: Vec = + serde_json::from_slice(&bytes).expect("failed to parse addresses desc page"); + let page_addresses: Vec<_> = page.iter().map(|x| x.address.clone()).collect(); + assert_eq!(page_addresses, reversed[2..4].to_vec()); + let address_bounds = |addr: &str| { app.vectors() .account_address_bounds @@ -1568,11 +1670,37 @@ mod tests { .expect("missing address in vectors") }; - let desc_blocks: Vec<_> = desc.iter().map(|x| address_bounds(&x.address).1).collect(); + // first on-chain appearance must not increase along `desc` + let desc_blocks: Vec<_> = desc.iter().map(|x| address_bounds(&x.address).0).collect(); assert!(desc_blocks.windows(2).all(|w| w[0] >= w[1])); } + #[tokio::test] + async fn accounts_by_stake_addresses_pool_only_account_returns_empty_list() { + let app = TestApp::new(); + + // The synthetic chain registers a pool owned by key hash [2u8; 28]. + // That credential never appears in an address or certificate, so no + // account state exists for it. Blockfrost still answers with an + // empty list because the credential is known through the pool. + let owner = StakeAddress::new(Network::Testnet, StakePayload::Stake(Hash::from([2u8; 28]))); + let owner = owner.to_bech32().expect("failed to encode owner address"); + + let path = format!("/accounts/{owner}/addresses"); + let (status, bytes) = app.get_bytes(&path).await; + assert_eq!( + status, + StatusCode::OK, + "unexpected status {status} with body: {}", + String::from_utf8_lossy(&bytes) + ); + + let addresses: Vec = + serde_json::from_slice(&bytes).expect("failed to parse account addresses"); + assert!(addresses.is_empty()); + } + #[tokio::test] async fn accounts_by_stake_addresses_bad_request() { let app = TestApp::new(); From dcf459f08ed1dc6575d1c5d2f84142b21ca43bcb Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Tue, 18 Aug 2026 14:22:43 +0200 Subject: [PATCH 2/2] feat(indexes): add a stake address log for account address queries The /accounts/{stake_address}/addresses endpoint scans every archive block that touches the account. With correct first-appearance ordering, a desc request must scan the account's full history: 305 seconds measured on mainnet for an exchange account with 400k+ addresses. The stake address log stores each (stake credential, address) pair once, at its first on-chain appearance, ordered by slot, transaction order, and output order. Both orders become one page read. - fjall: new stake-log keyspace with membership entries (the write probe and undo key) and ordered entries (the page read). - memory: same semantics, so ToyDomain tests exercise the log. - redb3 and noop answer None; redb3 is deprecated for index stores. - The apply path emits appearances from index_block. The undo path derives them from the same function, so a rollback removes exactly what apply inserted, and only when the undone block was the pair's first appearance. - A ready marker gates reads. Genesis bootstrap sets it, so stores synced from scratch serve the log. Existing stores answer None and the endpoint falls back to the archive scan until a resync. --- crates/cardano/src/genesis/mod.rs | 5 + crates/cardano/src/indexes/delta.rs | 51 ++++- crates/cardano/src/indexes/mod.rs | 4 +- crates/cardano/src/lib.rs | 7 +- crates/core/src/builtin/memory/index.rs | 237 +++++++++++++++++++++++- crates/core/src/builtin/noop.rs | 14 ++ crates/core/src/indexes.rs | 43 +++++ crates/fjall/src/index/mod.rs | 78 +++++++- crates/fjall/src/index/stake_log.rs | 199 ++++++++++++++++++++ crates/minibf/src/routes/accounts.rs | 78 +++++++- crates/minibf/src/test_support.rs | 34 +++- crates/redb3/src/indexes/mod.rs | 17 ++ crates/redb3/src/state/utxoset.rs | 1 + crates/testing/src/faults.rs | 21 +++ src/adapters/storage.rs | 24 +++ tests/index_roundtrip.rs | 112 ++++++++++- tests/memory.rs | 1 + 17 files changed, 915 insertions(+), 11 deletions(-) create mode 100644 crates/fjall/src/index/stake_log.rs diff --git a/crates/cardano/src/genesis/mod.rs b/crates/cardano/src/genesis/mod.rs index e2f967a0e..a02f9f6e0 100644 --- a/crates/cardano/src/genesis/mod.rs +++ b/crates/cardano/src/genesis/mod.rs @@ -170,6 +170,11 @@ pub fn bootstrap_utxos( state_writer.commit()?; index_writer.commit()?; + // A genesis bootstrap only happens on a fresh store, so every block the + // store will ever index flows through the apply path from here on. That + // makes the stake address log complete by construction. + indexes.mark_stake_log_ready()?; + Ok(()) } diff --git a/crates/cardano/src/indexes/delta.rs b/crates/cardano/src/indexes/delta.rs index e4fd16da0..852b42bfd 100644 --- a/crates/cardano/src/indexes/delta.rs +++ b/crates/cardano/src/indexes/delta.rs @@ -4,8 +4,8 @@ //! `IndexDelta` structures from Cardano block data. use dolos_core::{ - ArchiveIndexDelta, BlockSlot, ChainPoint, EraCbor, IndexDelta, Tag, TxoRef, UtxoIndexDelta, - UtxoSetDelta, + ArchiveIndexDelta, BlockSlot, ChainPoint, EraCbor, IndexDelta, StakeAddressAppearance, Tag, + TxoRef, UtxoIndexDelta, UtxoSetDelta, }; use pallas::{ codec::minicbor, @@ -283,6 +283,10 @@ impl CardanoIndexDeltaBuilder { self.start_block(block.slot(), block.hash().to_vec(), Some(block.number())); + self.delta + .stake_addresses + .extend(stake_appearances_from_block(block)); + for tx in block.txs() { self.add_tx_hash(tx.hash().to_vec()); @@ -412,6 +416,49 @@ impl CardanoIndexDeltaBuilder { } } +/// Stake address log candidates in one block: every produced output that +/// carries a stake credential, ordered by transaction and output index. +/// +/// Both the apply path (`index_block`) and the rollback path +/// (`compute_undo`) derive their log entries from this one function, so an +/// undo removes exactly what an apply inserted. +pub fn stake_appearances_from_block( + block: &pallas::ledger::traverse::MultiEraBlock<'_>, +) -> Vec { + let mut out = Vec::new(); + + for (tx_order, tx) in block.txs().iter().enumerate() { + for (output_order, output) in tx.produces() { + let Ok(address) = output.address() else { + continue; + }; + + let stake = match &address { + Address::Shelley(x) => { + pallas_extras::shelley_address_to_stake_address(x).map(|s| s.to_vec()) + } + Address::Stake(x) => Some(x.to_vec()), + Address::Byron(_) => None, + }; + + let Some(stake) = stake else { + continue; + }; + + let order = ((tx_order as u32) << 16) | (output_order as u32 & 0xffff); + + out.push(StakeAddressAppearance { + slot: block.slot(), + order, + stake, + address: address.to_vec(), + }); + } + } + + out +} + /// Build an `IndexDelta` from a `UtxoSetDelta` (for genesis/bulk import). /// /// This creates an `IndexDelta` containing only UTxO filter changes, diff --git a/crates/cardano/src/indexes/mod.rs b/crates/cardano/src/indexes/mod.rs index 237576d4e..763619fb1 100644 --- a/crates/cardano/src/indexes/mod.rs +++ b/crates/cardano/src/indexes/mod.rs @@ -12,7 +12,9 @@ mod dimensions; mod ext; mod query; -pub use delta::{index_delta_from_utxo_delta, CardanoIndexDeltaBuilder}; +pub use delta::{ + index_delta_from_utxo_delta, stake_appearances_from_block, CardanoIndexDeltaBuilder, +}; pub use dimensions::{archive as archive_dimensions, utxo as utxo_dimensions}; pub use ext::CardanoIndexExt; pub use query::{AsyncCardanoQueryExt, ScriptData, ScriptLanguage, SlotOrder}; diff --git a/crates/cardano/src/lib.rs b/crates/cardano/src/lib.rs index dfdbd33da..fcd8d5729 100644 --- a/crates/cardano/src/lib.rs +++ b/crates/cardano/src/lib.rs @@ -503,7 +503,12 @@ impl dolos_core::ChainLogic for CardanoLogic { let utxo_delta = crate::utxoset::compute_undo_delta(blockv, &decoded_inputs) .map_err(ChainError::from)?; - let index_delta = crate::indexes::index_delta_from_utxo_delta(point, &utxo_delta); + let mut index_delta = crate::indexes::index_delta_from_utxo_delta(point, &utxo_delta); + + // The undo path never re-runs `index_block`, so the stake address + // log entries to remove are derived here from the same function the + // apply path used to insert them. + index_delta.stake_addresses = crate::indexes::stake_appearances_from_block(blockv); let tx_hashes = blockv.txs().iter().map(|tx| tx.hash()).collect(); diff --git a/crates/core/src/builtin/memory/index.rs b/crates/core/src/builtin/memory/index.rs index cacb689b9..0015cfc5a 100644 --- a/crates/core/src/builtin/memory/index.rs +++ b/crates/core/src/builtin/memory/index.rs @@ -32,7 +32,8 @@ use std::sync::{Arc, Mutex, RwLock}; use crate::indexes::{ key_hash, ArchiveIndexDelta, ExactKind, ExactRecord, IndexDelta, IndexError, IndexRecord, - IndexStore, IndexWriter, KeyHash, TagDimension, TagRecord, MAX_EXACT_KEY_LEN, + IndexStore, IndexWriter, KeyHash, StakeAddressAppearance, TagDimension, TagRecord, + MAX_EXACT_KEY_LEN, }; use crate::{BlockSlot, ChainPoint, TxoRef, UtxoSet}; @@ -44,6 +45,12 @@ type ArchiveTag = (Cow<'static, str>, KeyHash, BlockSlot); /// heap allocation on either side. type ExactKey = [u8; MAX_EXACT_KEY_LEN]; +/// One ordered entry of the stake address log: `(slot, order, address)`. +type StakeLogEntry = (BlockSlot, u32, Vec); + +/// A `(stake, address)` pair, the membership key of the stake address log. +type StakeLogPair = (Vec, Vec); + /// A key's stored form, or `None` unless it is exactly the width its kind /// requires. /// @@ -74,6 +81,15 @@ struct Tables { /// Keyed on the record's own inline key rather than a `Vec`, so /// `iter_exact_records` copies rather than allocates per record. exact: BTreeMap<(ExactKind, ExactKey), BlockSlot>, + /// Stake address log: appearances ordered by `(slot, order, address)` + /// per stake credential. The set holds only first appearances. + stake_log: BTreeMap, BTreeSet>, + /// Membership map for the log: each pair's first appearance, which is + /// also what an undo has to match before it may remove the pair. + stake_log_pairs: BTreeMap, + /// Set once the log is complete from genesis; queries answer `None` + /// until then. + stake_log_ready: bool, } /// A single mutation, recorded by a writer and replayed at commit. See the @@ -86,6 +102,8 @@ enum Op { RemoveArchiveTag(ArchiveTag), InsertExact(ExactKind, ExactKey, BlockSlot), RemoveExact(ExactKind, ExactKey), + InsertStakeAddress(StakeAddressAppearance), + RemoveStakeAddress(StakeAddressAppearance), } fn poisoned() -> IndexError { @@ -206,6 +224,10 @@ impl IndexWriter for MemoryIndexWriter { } } + for appearance in &delta.stake_addresses { + ops.push(Op::InsertStakeAddress(appearance.clone())); + } + ops.push(Op::SetCursor(delta.cursor.clone())); Ok(()) @@ -248,6 +270,10 @@ impl IndexWriter for MemoryIndexWriter { } } + for appearance in &delta.stake_addresses { + ops.push(Op::RemoveStakeAddress(appearance.clone())); + } + Ok(()) } @@ -282,6 +308,9 @@ impl IndexWriter for MemoryIndexWriter { let mut tables = self.store.tables.write().map_err(|_| poisoned())?; + // Reborrow so the match arms can hold disjoint field borrows. + let tables = &mut *tables; + for op in ops { match op { Op::SetCursor(cursor) => tables.cursor = Some(cursor), @@ -314,6 +343,38 @@ impl IndexWriter for MemoryIndexWriter { Op::RemoveExact(kind, key) => { tables.exact.remove(&(kind, key)); } + Op::InsertStakeAddress(app) => { + let pair = (app.stake.clone(), app.address.clone()); + // Only the first appearance of a pair is kept; the + // membership map is the probe. + if let std::collections::btree_map::Entry::Vacant(entry) = + tables.stake_log_pairs.entry(pair) + { + entry.insert((app.slot, app.order)); + tables.stake_log.entry(app.stake).or_default().insert(( + app.slot, + app.order, + app.address, + )); + } + } + Op::RemoveStakeAddress(app) => { + let pair = (app.stake.clone(), app.address.clone()); + // Remove only when the undone block is the pair's first + // appearance; a pair seen earlier stays untouched. + let stored = tables.stake_log_pairs.get(&pair).copied(); + if let Some((slot, order)) = stored { + if slot == app.slot { + tables.stake_log_pairs.remove(&pair); + if let Some(set) = tables.stake_log.get_mut(&app.stake) { + set.remove(&(slot, order, app.address)); + if set.is_empty() { + tables.stake_log.remove(&app.stake); + } + } + } + } + } } } @@ -404,6 +465,14 @@ impl IndexStore for MemoryIndexStore { target.archive_tags.extend(source.archive_tags); target.exact.extend(source.exact); + for (stake, set) in source.stake_log { + target.stake_log.entry(stake).or_default().extend(set); + } + for (pair, first) in source.stake_log_pairs { + target.stake_log_pairs.entry(pair).or_insert(first); + } + target.stake_log_ready |= source.stake_log_ready; + Ok(()) } @@ -424,6 +493,45 @@ impl IndexStore for MemoryIndexStore { Ok(found) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, IndexError> { + let tables = self.tables.read().map_err(|_| poisoned())?; + + if !tables.stake_log_ready { + return Ok(None); + } + + let Some(set) = tables.stake_log.get(stake) else { + return Ok(Some(Vec::new())); + }; + + let pick = |entry: &(BlockSlot, u32, Vec)| entry.2.clone(); + + let page = if reverse { + set.iter() + .rev() + .skip(offset) + .take(limit) + .map(pick) + .collect() + } else { + set.iter().skip(offset).take(limit).map(pick).collect() + }; + + Ok(Some(page)) + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + let mut tables = self.tables.write().map_err(|_| poisoned())?; + tables.stake_log_ready = true; + Ok(()) + } + fn slot_by_block_hash(&self, hash: &[u8]) -> Result, IndexError> { // A key that could not have been stored cannot be found, so a // wrong-width query is a miss rather than an error. @@ -529,3 +637,130 @@ impl IndexStore for MemoryIndexStore { Ok(MemoryExactIter(records.into_iter())) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn appearance(slot: BlockSlot, order: u32, stake: u8, address: u8) -> StakeAddressAppearance { + StakeAddressAppearance { + slot, + order, + stake: vec![stake; 29], + address: vec![address; 57], + } + } + + fn apply_appearances(store: &MemoryIndexStore, items: &[StakeAddressAppearance]) { + let writer = store.start_writer().unwrap(); + let delta = IndexDelta { + stake_addresses: items.to_vec(), + ..Default::default() + }; + writer.apply(&delta).unwrap(); + writer.commit().unwrap(); + } + + fn undo_appearances(store: &MemoryIndexStore, items: &[StakeAddressAppearance]) { + let writer = store.start_writer().unwrap(); + let delta = IndexDelta { + stake_addresses: items.to_vec(), + ..Default::default() + }; + writer.undo(&delta).unwrap(); + writer.commit().unwrap(); + } + + fn page(store: &MemoryIndexStore, stake: u8, reverse: bool) -> Option>> { + store + .addresses_by_stake_log(&[stake; 29], 0, 100, reverse) + .unwrap() + } + + #[test] + fn stake_log_answers_none_until_marked_ready() { + let store = MemoryIndexStore::new(); + apply_appearances(&store, &[appearance(1, 0, 0xaa, 0x01)]); + + assert_eq!(page(&store, 0xaa, false), None); + + store.mark_stake_log_ready().unwrap(); + + assert_eq!(page(&store, 0xaa, false), Some(vec![vec![0x01; 57]])); + } + + #[test] + fn stake_log_keeps_first_appearance_only() { + let store = MemoryIndexStore::new(); + store.mark_stake_log_ready().unwrap(); + + // address 0x01 first at slot 1, reused at slot 3; + // address 0x02 first at slot 2 + apply_appearances(&store, &[appearance(1, 0, 0xaa, 0x01)]); + apply_appearances(&store, &[appearance(2, 0, 0xaa, 0x02)]); + apply_appearances(&store, &[appearance(3, 0, 0xaa, 0x01)]); + + let asc = page(&store, 0xaa, false).unwrap(); + assert_eq!(asc, vec![vec![0x01; 57], vec![0x02; 57]]); + + // desc is the exact reverse: the reused address stays last + let desc = page(&store, 0xaa, true).unwrap(); + assert_eq!(desc, vec![vec![0x02; 57], vec![0x01; 57]]); + } + + #[test] + fn stake_log_undo_removes_only_the_first_appearance() { + let store = MemoryIndexStore::new(); + store.mark_stake_log_ready().unwrap(); + + apply_appearances(&store, &[appearance(1, 0, 0xaa, 0x01)]); + apply_appearances(&store, &[appearance(2, 0, 0xaa, 0x01)]); + + // rolling back the repeat leaves the pair in place + undo_appearances(&store, &[appearance(2, 0, 0xaa, 0x01)]); + assert_eq!(page(&store, 0xaa, false), Some(vec![vec![0x01; 57]])); + + // rolling back the first appearance removes it + undo_appearances(&store, &[appearance(1, 0, 0xaa, 0x01)]); + assert_eq!(page(&store, 0xaa, false), Some(vec![])); + } + + #[test] + fn stake_log_orders_within_a_block_by_order_field() { + let store = MemoryIndexStore::new(); + store.mark_stake_log_ready().unwrap(); + + apply_appearances( + &store, + &[appearance(1, 7, 0xaa, 0x02), appearance(1, 3, 0xaa, 0x01)], + ); + + let asc = page(&store, 0xaa, false).unwrap(); + assert_eq!(asc, vec![vec![0x01; 57], vec![0x02; 57]]); + } + + #[test] + fn stake_log_pages_and_windows() { + let store = MemoryIndexStore::new(); + store.mark_stake_log_ready().unwrap(); + + for i in 0u8..5 { + apply_appearances(&store, &[appearance(i as u64 + 1, 0, 0xaa, i + 1)]); + } + + let window = store + .addresses_by_stake_log(&[0xaa; 29], 2, 2, false) + .unwrap() + .unwrap(); + assert_eq!(window, vec![vec![3; 57], vec![4; 57]]); + + let reversed = store + .addresses_by_stake_log(&[0xaa; 29], 2, 2, true) + .unwrap() + .unwrap(); + assert_eq!(reversed, vec![vec![3; 57], vec![2; 57]]); + + // an unknown stake answers an empty page, not `None` + assert_eq!(page(&store, 0xbb, false), Some(vec![])); + } +} diff --git a/crates/core/src/builtin/noop.rs b/crates/core/src/builtin/noop.rs index d5e98e1d2..fae9dc170 100644 --- a/crates/core/src/builtin/noop.rs +++ b/crates/core/src/builtin/noop.rs @@ -105,6 +105,20 @@ impl IndexStore for NoOpIndexStore { Ok(UtxoSet::default()) } + fn addresses_by_stake_log( + &self, + _stake: &[u8], + _offset: usize, + _limit: usize, + _reverse: bool, + ) -> Result>>, IndexError> { + Ok(None) + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + Ok(()) + } + fn slot_by_block_hash(&self, _hash: &[u8]) -> Result, IndexError> { Ok(None) } diff --git a/crates/core/src/indexes.rs b/crates/core/src/indexes.rs index 0252ae968..0171c2ef9 100644 --- a/crates/core/src/indexes.rs +++ b/crates/core/src/indexes.rs @@ -69,6 +69,20 @@ pub struct ArchiveIndexDelta { pub tags: Vec, } +/// One address appearing under a stake credential inside a block. +/// +/// These records feed the stake address log: the per-account list of +/// addresses ordered by first on-chain appearance. `order` breaks ties +/// inside one block (transaction index, then output index). Stores keep +/// only the first appearance of each `(stake, address)` pair. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StakeAddressAppearance { + pub slot: BlockSlot, + pub order: u32, + pub stake: Vec, + pub address: Vec, +} + /// Unified index delta for a batch of operations. /// /// This structure contains all index changes for a batch of blocks, @@ -81,6 +95,9 @@ pub struct IndexDelta { pub utxo: UtxoIndexDelta, /// Archive index changes (one per block in batch). pub archive: Vec, + /// First-appearance candidates for the stake address log, one per + /// produced output that carries a stake credential. + pub stake_addresses: Vec, } impl Default for IndexDelta { @@ -89,6 +106,7 @@ impl Default for IndexDelta { cursor: ChainPoint::Origin, utxo: UtxoIndexDelta::default(), archive: Vec::new(), + stake_addresses: Vec::new(), } } } @@ -462,6 +480,31 @@ pub trait IndexStore: Clone + Send + Sync + 'static { /// dimension and key and have not been consumed. fn utxos_by_tag(&self, dimension: TagDimension, key: &[u8]) -> Result; + // ============ Stake Address Log ============ + + /// Read one page of the stake address log: the addresses seen under the + /// stake credential, ordered by first on-chain appearance. + /// + /// `offset` and `limit` window the ordered list; `reverse` reads the + /// exact reverse of it. Returns `None` when the log is not authoritative + /// on this store — the backend does not maintain it, or the store + /// predates the log and was never rebuilt. Callers fall back to an + /// archive scan in that case. + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, IndexError>; + + /// Declare the stake address log complete from genesis. + /// + /// A genesis bootstrap calls this once on a fresh store. Until it runs, + /// [`IndexStore::addresses_by_stake_log`] answers `None`. Backends that + /// do not maintain the log treat this as a no-op. + fn mark_stake_log_ready(&self) -> Result<(), IndexError>; + // ============ Archive Queries (Exact Lookups) ============ /// Get the slot for a block by its hash (exact lookup). diff --git a/crates/fjall/src/index/mod.rs b/crates/fjall/src/index/mod.rs index 316d1cc3a..de7eb0e36 100644 --- a/crates/fjall/src/index/mod.rs +++ b/crates/fjall/src/index/mod.rs @@ -50,6 +50,7 @@ use fjall::{ pub mod archive_tags; pub mod exact; pub mod scan; +pub mod stake_log; pub mod state_tags; use crate::keys::{dim_prefix, hash_dimension, DIM_HASH_SIZE}; @@ -80,11 +81,17 @@ mod keyspace_names { pub const UTXO_TAGS: &str = "state-tags"; /// Archive tags keyspace (append-only, never deleted) pub const BLOCK_TAGS: &str = "archive-tags"; + /// Stake address log keyspace (insert-once pairs, removed on rollback) + pub const STAKE_LOG: &str = "stake-log"; } /// Key for the cursor entry const CURSOR_KEY: &[u8] = &[0u8]; +/// Key in the cursor keyspace marking the stake address log as complete +/// from genesis. Queries answer `None` until it is written. +const STAKE_LOG_READY_KEY: &[u8] = &[1u8]; + /// Fjall-based index store implementation with four keyspaces. /// /// Uses 4 keyspaces split by workload class: @@ -104,6 +111,8 @@ pub struct IndexStore { utxo_tags: Keyspace, /// Block tags keyspace (append-only, never deleted) block_tags: Keyspace, + /// Stake address log keyspace (insert-once pairs, removed on rollback) + stake_log: Keyspace, /// Configuration flush_on_commit: bool, } @@ -163,11 +172,12 @@ impl IndexStore { opts }; - // 4 keyspaces: cursor, exact, utxo_tags, block_tags + // 5 keyspaces: cursor, exact, utxo_tags, block_tags, stake_log let cursor = db.keyspace(keyspace_names::CURSOR, build_opts)?; let exact = db.keyspace(keyspace_names::EXACT, build_opts)?; let utxo_tags = db.keyspace(keyspace_names::UTXO_TAGS, build_opts)?; let block_tags = db.keyspace(keyspace_names::BLOCK_TAGS, build_opts)?; + let stake_log = db.keyspace(keyspace_names::STAKE_LOG, build_opts)?; Ok(Self { db, @@ -175,6 +185,7 @@ impl IndexStore { exact, utxo_tags, block_tags, + stake_log, flush_on_commit, }) } @@ -249,6 +260,10 @@ impl IndexStore { pub struct IndexStoreWriter { batch: Mutex, store: IndexStore, + /// Pairs already inserted into the stake log by this batch. The batch + /// cannot read its own pending writes, so cross-block dedup inside one + /// writer lives here. + stake_pairs_seen: Mutex>>, } impl CoreIndexWriter for IndexStoreWriter { @@ -264,6 +279,22 @@ impl CoreIndexWriter for IndexStoreWriter { // Apply archive tag changes to archive-tags keyspace archive_tags::apply(&mut batch, &self.store.block_tags, delta).map_err(IndexError::from)?; + // Apply stake address log first appearances + let mut seen = self + .stake_pairs_seen + .lock() + .map_err(|_| Error::LockPoisoned)?; + let snapshot = self.store.db.snapshot(); + stake_log::apply( + &mut batch, + &self.store.stake_log, + &snapshot, + &mut seen, + &delta.stake_addresses, + ) + .map_err(IndexError::from)?; + drop(seen); + // Set cursor let cursor_bytes = bincode::serialize(&delta.cursor).map_err(|e| Error::Codec(e.to_string()))?; @@ -284,6 +315,16 @@ impl CoreIndexWriter for IndexStoreWriter { // Undo archive tag changes archive_tags::undo(&mut batch, &self.store.block_tags, delta).map_err(IndexError::from)?; + // Undo stake address log first appearances + let snapshot = self.store.db.snapshot(); + stake_log::undo( + &mut batch, + &self.store.stake_log, + &snapshot, + &delta.stake_addresses, + ) + .map_err(IndexError::from)?; + Ok(()) } @@ -369,6 +410,7 @@ impl CoreIndexStore for IndexStore { Ok(IndexStoreWriter { batch: Mutex::new(batch), store: self.clone(), + stake_pairs_seen: Mutex::new(std::collections::HashSet::new()), }) } @@ -409,6 +451,40 @@ impl CoreIndexStore for IndexStore { state_tags::get_by_key(&snapshot, &self.utxo_tags, dimension, key).map_err(IndexError::from) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, IndexError> { + // Use snapshot for MVCC reads to avoid deadlocks with concurrent writes + let snapshot = self.db.snapshot(); + + // Without the ready marker the log is not authoritative: the store + // may predate the log entirely. Callers fall back to an archive scan. + let ready = snapshot + .get(&self.cursor, STAKE_LOG_READY_KEY) + .map_err(Error::from)? + .is_some(); + + if !ready { + return Ok(None); + } + + let page = stake_log::page(&snapshot, &self.stake_log, stake, offset, limit, reverse) + .map_err(IndexError::from)?; + + Ok(Some(page)) + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + let mut batch = self.db.batch().durability(Some(PersistMode::Buffer)); + batch.insert(&self.cursor, STAKE_LOG_READY_KEY, [1u8]); + batch.commit().map_err(Error::Fjall)?; + Ok(()) + } + fn slot_by_block_hash(&self, block_hash: &[u8]) -> Result, IndexError> { // Use snapshot for MVCC reads to avoid deadlocks with concurrent writes let snapshot = self.db.snapshot(); diff --git a/crates/fjall/src/index/stake_log.rs b/crates/fjall/src/index/stake_log.rs new file mode 100644 index 000000000..c574f94d8 --- /dev/null +++ b/crates/fjall/src/index/stake_log.rs @@ -0,0 +1,199 @@ +//! Stake address log operations for the `stake-log` keyspace. +//! +//! The log answers one query: the addresses seen under a stake credential, +//! ordered by first on-chain appearance. Two entry shapes share the +//! keyspace, discriminated by a tag byte: +//! +//! - Pair entry: `[0x00][stake_len:1][stake][address]` -> `[slot:8][order:4]`. +//! One per known `(stake, address)` pair. This is the membership probe on the +//! write path and the undo key on rollback. +//! - Ordered entry: `[0x01][stake_len:1][stake][slot:8][order:4][address]` -> +//! empty. Lexicographic key order is chronological order, so a page read is a +//! prefix scan windowed from either end. +//! +//! Only the first appearance of a pair is stored. The write batch cannot +//! read its own pending inserts, so the caller threads a `seen` set through +//! `apply` to dedup pairs inside one batch; the pair entry dedups across +//! batches. + +use std::collections::HashSet; + +use dolos_core::{BlockSlot, StakeAddressAppearance}; +use fjall::{Keyspace, OwnedWriteBatch, Readable}; + +use crate::Error; + +/// Tag byte for pair (membership) entries. +const PAIR_TAG: u8 = 0x00; + +/// Tag byte for ordered (page-read) entries. +const ORDERED_TAG: u8 = 0x01; + +/// Width of the `[slot:8][order:4]` sort key. +const SORT_KEY_SIZE: usize = 12; + +fn build_pair_key(stake: &[u8], address: &[u8]) -> Vec { + let mut key = Vec::with_capacity(2 + stake.len() + address.len()); + key.push(PAIR_TAG); + key.push(stake.len() as u8); + key.extend_from_slice(stake); + key.extend_from_slice(address); + key +} + +fn build_ordered_key(stake: &[u8], slot: BlockSlot, order: u32, address: &[u8]) -> Vec { + let mut key = Vec::with_capacity(2 + stake.len() + SORT_KEY_SIZE + address.len()); + key.push(ORDERED_TAG); + key.push(stake.len() as u8); + key.extend_from_slice(stake); + key.extend_from_slice(&slot.to_be_bytes()); + key.extend_from_slice(&order.to_be_bytes()); + key.extend_from_slice(address); + key +} + +/// Prefix covering every ordered entry of one stake credential. +fn build_ordered_prefix(stake: &[u8]) -> Vec { + let mut prefix = Vec::with_capacity(2 + stake.len()); + prefix.push(ORDERED_TAG); + prefix.push(stake.len() as u8); + prefix.extend_from_slice(stake); + prefix +} + +fn encode_sort_key(slot: BlockSlot, order: u32) -> [u8; SORT_KEY_SIZE] { + let mut value = [0u8; SORT_KEY_SIZE]; + value[..8].copy_from_slice(&slot.to_be_bytes()); + value[8..].copy_from_slice(&order.to_be_bytes()); + value +} + +fn decode_sort_key(value: &[u8]) -> Option<(BlockSlot, u32)> { + if value.len() != SORT_KEY_SIZE { + return None; + } + + let slot = BlockSlot::from_be_bytes(value[..8].try_into().ok()?); + let order = u32::from_be_bytes(value[8..].try_into().ok()?); + Some((slot, order)) +} + +/// Insert the first appearance of each pair the delta carries. +/// +/// `seen` dedups pairs inside the current write batch: the batch cannot +/// read its own pending inserts, and one batch can span many blocks (WAL +/// catch-up applies a whole range through a single writer). +pub fn apply( + batch: &mut OwnedWriteBatch, + keyspace: &Keyspace, + readable: &R, + seen: &mut HashSet>, + appearances: &[StakeAddressAppearance], +) -> Result<(), Error> { + for appearance in appearances { + let pair_key = build_pair_key(&appearance.stake, &appearance.address); + + if seen.contains(&pair_key) { + continue; + } + + if readable.get(keyspace, &pair_key)?.is_some() { + seen.insert(pair_key); + continue; + } + + batch.insert( + keyspace, + pair_key.clone(), + encode_sort_key(appearance.slot, appearance.order), + ); + + batch.insert( + keyspace, + build_ordered_key( + &appearance.stake, + appearance.slot, + appearance.order, + &appearance.address, + ), + [], + ); + + seen.insert(pair_key); + } + + Ok(()) +} + +/// Remove pairs whose stored first appearance is the undone block. +/// +/// A pair first seen in an earlier block stays untouched: the undone block +/// merely repeated an address the account already had. +pub fn undo( + batch: &mut OwnedWriteBatch, + keyspace: &Keyspace, + readable: &R, + appearances: &[StakeAddressAppearance], +) -> Result<(), Error> { + for appearance in appearances { + let pair_key = build_pair_key(&appearance.stake, &appearance.address); + + let Some(value) = readable.get(keyspace, &pair_key)? else { + continue; + }; + + let Some((slot, order)) = decode_sort_key(&value) else { + continue; + }; + + if slot != appearance.slot { + continue; + } + + batch.remove(keyspace, pair_key); + batch.remove( + keyspace, + build_ordered_key(&appearance.stake, slot, order, &appearance.address), + ); + } + + Ok(()) +} + +/// Read one page of addresses for a stake credential, ordered by first +/// appearance (or its exact reverse). +pub fn page( + readable: &R, + keyspace: &Keyspace, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, +) -> Result>, Error> { + let prefix = build_ordered_prefix(stake); + let header = prefix.len() + SORT_KEY_SIZE; + + let iter = readable.prefix(keyspace, prefix); + + let mut page = Vec::new(); + + if reverse { + for guard in iter.rev().skip(offset).take(limit) { + let key = guard.key()?; + + if key.len() > header { + page.push(key[header..].to_vec()); + } + } + } else { + for guard in iter.skip(offset).take(limit) { + let key = guard.key()?; + + if key.len() > header { + page.push(key[header..].to_vec()); + } + } + } + + Ok(page) +} diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index b87a3306f..b0f6b1e2b 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -27,8 +27,8 @@ use dolos_cardano::{ PoolDepositRefundLog, }; use dolos_core::{ - async_query::BlockRefMeta, ArchiveStore as _, Domain, EntityKey, LogKey, StateStore as _, - TemporalKey, TxHash, + async_query::BlockRefMeta, ArchiveStore as _, Domain, EntityKey, IndexStore as _, LogKey, + StateStore as _, TemporalKey, TxHash, }; use futures::future::join_all; use futures_util::StreamExt; @@ -305,6 +305,37 @@ where return Err(StatusCode::NOT_FOUND.into()); } + // The stake address log answers both orders with one page read. It only + // exists on stores synced since the log was introduced; `None` falls + // through to the archive scan below. The scan honors from/to filters, + // the log does not, so range-filtered requests always scan. + if pagination.from.is_none() && pagination.to.is_none() { + let page = domain + .indexes() + .addresses_by_stake_log( + &account_key.address.to_vec(), + pagination.skip(), + pagination.count, + matches!(pagination.order, Order::Desc), + ) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + if let Some(addresses) = page { + let items = addresses + .into_iter() + .map(|bytes| { + Address::from_bytes(&bytes) + .map(|address| AccountAddressesContentInner { + address: address.to_string(), + }) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) + }) + .collect::, _>>()?; + + return Ok(Json(items)); + } + } + let (start_slot, end_slot) = pagination.start_and_end_slots(&domain).await?; // Blockfrost orders addresses by first on-chain appearance, and `desc` @@ -1701,6 +1732,49 @@ mod tests { assert!(addresses.is_empty()); } + #[tokio::test] + async fn accounts_by_stake_addresses_log_matches_archive_scan() { + // Two identical synthetic chains: one serves from the stake address + // log, the other from the archive-scan fallback. Every query shape + // must produce identical responses on both. + let logged = TestApp::new(); + let scanned = TestApp::new_scan_fallback(); + + let stake_address = logged.vectors().stake_address.clone(); + assert_eq!(stake_address, scanned.vectors().stake_address); + + let queries = [ + "order=asc&count=100", + "order=desc&count=100", + "order=asc&count=2&page=2", + "order=desc&count=2&page=2", + "count=1&page=3", + ]; + + for query in queries { + let path = format!("/accounts/{stake_address}/addresses?{query}"); + + let (status, bytes) = logged.get_bytes(&path).await; + assert_eq!(status, StatusCode::OK, "log path failed for {query}"); + let from_log: Vec = + serde_json::from_slice(&bytes).expect("failed to parse log response"); + + let (status, bytes) = scanned.get_bytes(&path).await; + assert_eq!(status, StatusCode::OK, "scan path failed for {query}"); + let from_scan: Vec = + serde_json::from_slice(&bytes).expect("failed to parse scan response"); + + let log_addresses: Vec<_> = from_log.iter().map(|x| x.address.clone()).collect(); + let scan_addresses: Vec<_> = from_scan.iter().map(|x| x.address.clone()).collect(); + + assert!(!log_addresses.is_empty(), "empty response for {query}"); + assert_eq!( + log_addresses, scan_addresses, + "log and scan disagree for {query}" + ); + } + } + #[tokio::test] async fn accounts_by_stake_addresses_bad_request() { let app = TestApp::new(); diff --git a/crates/minibf/src/test_support.rs b/crates/minibf/src/test_support.rs index 290dc095a..8cad42750 100644 --- a/crates/minibf/src/test_support.rs +++ b/crates/minibf/src/test_support.rs @@ -8,7 +8,7 @@ use axum::{ use dolos_core::{ config::{CardanoConfig, MinibfConfig}, import::ImportExt as _, - Domain, StateStore, + Domain, IndexStore as _, StateStore, }; use dolos_testing::{ synthetic::{ @@ -29,7 +29,17 @@ pub struct TestDomainBuilder { } impl TestDomainBuilder { - pub fn new_with_synthetic(mut cfg: SyntheticBlockConfig) -> Self { + pub fn new_with_synthetic(cfg: SyntheticBlockConfig) -> Self { + Self::new_with_synthetic_and_stake_log(cfg, true) + } + + /// `mark_stake_log` controls whether the domain declares its stake + /// address log authoritative. Leave it unset to force the archive-scan + /// fallback in endpoint tests. + pub fn new_with_synthetic_and_stake_log( + mut cfg: SyntheticBlockConfig, + mark_stake_log: bool, + ) -> Self { let genesis = Arc::new(dolos_cardano::include::preview::load()); let min_slot = { let temp = ToyDomain::new_with_genesis_and_config( @@ -84,6 +94,13 @@ impl TestDomainBuilder { .expect("failed to seed account stake logs"); } + if mark_stake_log { + domain + .indexes() + .mark_stake_log_ready() + .expect("failed to mark stake log ready"); + } + Self { domain, vectors } } @@ -121,6 +138,19 @@ impl TestApp { Self::from_domain(domain, vectors, fault) } + /// Like [`TestApp::new`], but without an authoritative stake address + /// log, so account-address requests take the archive-scan fallback. + pub fn new_scan_fallback() -> Self { + let cfg = SyntheticBlockConfig { + block_count: 5, + txs_per_block: 3, + ..Default::default() + }; + let (domain, vectors) = + TestDomainBuilder::new_with_synthetic_and_stake_log(cfg, false).finish(); + Self::from_domain(domain, vectors, None) + } + pub fn new_with_cfg_and_setup( cfg: SyntheticBlockConfig, setup: impl FnOnce(&ToyDomain, &SyntheticVectors), diff --git a/crates/redb3/src/indexes/mod.rs b/crates/redb3/src/indexes/mod.rs index 910c3380a..d46e217da 100644 --- a/crates/redb3/src/indexes/mod.rs +++ b/crates/redb3/src/indexes/mod.rs @@ -810,6 +810,23 @@ impl CoreIndexStore for IndexStore { self.initialize_schema_internal().map_err(IndexError::from) } + /// This backend is deprecated for real nodes and does not maintain the + /// stake address log. `None` sends callers to their archive-scan + /// fallback. + fn addresses_by_stake_log( + &self, + _stake: &[u8], + _offset: usize, + _limit: usize, + _reverse: bool, + ) -> Result>>, IndexError> { + Ok(None) + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + Ok(()) + } + fn copy(&self, target: &Self) -> Result<(), IndexError> { let rx = self.db.begin_read().map_err(map_db_error)?; let wx = target.db.begin_write().map_err(map_db_error)?; diff --git a/crates/redb3/src/state/utxoset.rs b/crates/redb3/src/state/utxoset.rs index 75107fa13..a7e48d77a 100644 --- a/crates/redb3/src/state/utxoset.rs +++ b/crates/redb3/src/state/utxoset.rs @@ -217,6 +217,7 @@ mod tests { cursor, utxo: UtxoIndexDelta { produced, consumed }, archive: Vec::new(), + stake_addresses: Vec::new(), } } diff --git a/crates/testing/src/faults.rs b/crates/testing/src/faults.rs index 8742a8789..5eb43e100 100644 --- a/crates/testing/src/faults.rs +++ b/crates/testing/src/faults.rs @@ -320,6 +320,27 @@ impl IndexStore for FaultyIndexStore { self.inner.utxos_by_tag(dimension, key) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, IndexError> { + if self.should_fault() { + return Err(self.fault_err()); + } + self.inner + .addresses_by_stake_log(stake, offset, limit, reverse) + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + if self.should_fault() { + return Err(self.fault_err()); + } + self.inner.mark_stake_log_ready() + } + fn slot_by_block_hash(&self, hash: &[u8]) -> Result, IndexError> { if self.should_fault() { return Err(self.fault_err()); diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index b27115cba..cec267a16 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -1158,6 +1158,30 @@ impl CoreIndexStore for IndexStoreBackend { } } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, IndexError> { + match self { + Self::Redb(s) => s.addresses_by_stake_log(stake, offset, limit, reverse), + Self::Fjall(s) => s.addresses_by_stake_log(stake, offset, limit, reverse), + Self::Memory(s) => s.addresses_by_stake_log(stake, offset, limit, reverse), + Self::NoOp(s) => s.addresses_by_stake_log(stake, offset, limit, reverse), + } + } + + fn mark_stake_log_ready(&self) -> Result<(), IndexError> { + match self { + Self::Redb(s) => s.mark_stake_log_ready(), + Self::Fjall(s) => s.mark_stake_log_ready(), + Self::Memory(s) => s.mark_stake_log_ready(), + Self::NoOp(s) => s.mark_stake_log_ready(), + } + } + fn slot_by_block_hash(&self, hash: &[u8]) -> Result, IndexError> { match self { Self::Redb(s) => s.slot_by_block_hash(hash), diff --git a/tests/index_roundtrip.rs b/tests/index_roundtrip.rs index d9c082248..e63787272 100644 --- a/tests/index_roundtrip.rs +++ b/tests/index_roundtrip.rs @@ -28,7 +28,7 @@ use dolos_cardano::indexes::{archive_dimensions, CardanoIndexExt}; use dolos_core::{ builtin::MemoryIndexStore, config::FjallIndexConfig, ArchiveIndexDelta, BlockSlot, ChainPoint, ExactKind, ExactRecord, IndexDelta, IndexRecord, IndexStore as CoreIndexStore, - IndexWriter as CoreIndexWriter, Tag, TagDimension, TagRecord, + IndexWriter as CoreIndexWriter, StakeAddressAppearance, Tag, TagDimension, TagRecord, }; const EPOCH_LEN: BlockSlot = 432_000; @@ -212,6 +212,7 @@ fn seed_deltas(spec: &SeedSpec, sink: &mut impl FnMut(IndexDelta)) -> SeedCounts cursor, utxo: Default::default(), archive, + stake_addresses: Vec::new(), }); block = batch_end; @@ -353,6 +354,11 @@ macro_rules! conformance_suite { fn slots_by_tag_are_ordered_in_both_directions() { super::slots_by_tag_are_ordered_in_both_directions::<$backend>(); } + + #[test] + fn stake_log_round_trips_and_pages() { + super::stake_log_round_trips_and_pages::<$backend>(); + } } }; } @@ -463,6 +469,7 @@ fn slots_by_tag_are_ordered_in_both_directions() { cursor: ChainPoint::Slot(30), utxo: Default::default(), archive, + stake_addresses: Vec::new(), }, ); @@ -1112,6 +1119,7 @@ fn malformed_exact_keys_are_refused() { tx_hashes: vec![vec![0xCD; 32]], tags: Vec::new(), }], + stake_addresses: Vec::new(), }; let writer = store.start_writer().expect("start_writer failed"); @@ -1153,6 +1161,7 @@ fn malformed_exact_keys_are_refused() { tx_hashes: vec![vec![0xEF; width]], tags: Vec::new(), }], + stake_addresses: Vec::new(), }; let writer = store.start_writer().expect("start_writer failed"); @@ -1202,3 +1211,104 @@ fn seeded_block() -> (Vec, u64, BlockSlot) { .expect("the seed writes blocks") .clone() } + +/// The stake address log conformance check: gated by the ready marker, +/// first-appearance dedup (also inside one writer spanning blocks), ordered +/// paging in both directions, and undo of only the first appearance. +fn stake_log_round_trips_and_pages() { + let (store, _guard) = B::open(); + + let stake_a = vec![0xAA; 29]; + let stake_b = vec![0xBB; 29]; + let addr = |b: u8| vec![b; 57]; + + let appearance = + |slot: BlockSlot, order: u32, stake: &[u8], address: u8| StakeAddressAppearance { + slot, + order, + stake: stake.to_vec(), + address: addr(address), + }; + + let delta = |slot: BlockSlot, items: Vec| IndexDelta { + cursor: ChainPoint::Slot(slot), + stake_addresses: items, + ..Default::default() + }; + + let apply_one = |d: IndexDelta| { + let writer = store.start_writer().expect("start_writer failed"); + writer.apply(&d).expect("apply failed"); + writer.commit().expect("commit failed"); + }; + + let page = |stake: &[u8], offset: usize, limit: usize, reverse: bool| { + store + .addresses_by_stake_log(stake, offset, limit, reverse) + .expect("addresses_by_stake_log failed") + }; + + // before the marker the log is not authoritative, but writes still land + apply_one(delta(1, vec![appearance(1, 0, &stake_a, 0x01)])); + assert_eq!(page(&stake_a, 0, 10, false), None); + + store + .mark_stake_log_ready() + .expect("mark_stake_log_ready failed"); + + // one writer spanning two blocks dedups the pair the later block repeats + { + let writer = store.start_writer().expect("start_writer failed"); + writer + .apply(&delta(2, vec![appearance(2, 0, &stake_a, 0x02)])) + .expect("apply failed"); + writer + .apply(&delta( + 3, + vec![ + appearance(3, 0, &stake_a, 0x02), + appearance(3, 1, &stake_b, 0x03), + ], + )) + .expect("apply failed"); + writer.commit().expect("commit failed"); + } + + // a repeat in a later writer is deduped against the committed store + apply_one(delta(4, vec![appearance(4, 0, &stake_a, 0x01)])); + + let asc = page(&stake_a, 0, 10, false).expect("log should be ready"); + assert_eq!(asc, vec![addr(0x01), addr(0x02)]); + + // desc is the exact reverse of asc + let desc = page(&stake_a, 0, 10, true).expect("log should be ready"); + assert_eq!(desc, vec![addr(0x02), addr(0x01)]); + + // offset windows work from both ends + assert_eq!(page(&stake_a, 1, 1, false).unwrap(), vec![addr(0x02)]); + assert_eq!(page(&stake_a, 1, 1, true).unwrap(), vec![addr(0x01)]); + + // stakes are isolated, and an unknown stake is an empty page, not None + assert_eq!(page(&stake_b, 0, 10, false).unwrap(), vec![addr(0x03)]); + assert_eq!( + page(&[0xCC; 29], 0, 10, false).unwrap(), + Vec::>::new() + ); + + let undo_one = |d: IndexDelta| { + let writer = store.start_writer().expect("start_writer failed"); + writer.undo(&d).expect("undo failed"); + writer.commit().expect("commit failed"); + }; + + // undoing the repeat leaves the pair in place + undo_one(delta(4, vec![appearance(4, 0, &stake_a, 0x01)])); + assert_eq!( + page(&stake_a, 0, 10, false).unwrap(), + vec![addr(0x01), addr(0x02)] + ); + + // undoing the first appearance removes it + undo_one(delta(1, vec![appearance(1, 0, &stake_a, 0x01)])); + assert_eq!(page(&stake_a, 0, 10, false).unwrap(), vec![addr(0x02)]); +} diff --git a/tests/memory.rs b/tests/memory.rs index 84d637fa5..10a171c0c 100644 --- a/tests/memory.rs +++ b/tests/memory.rs @@ -386,6 +386,7 @@ fn seed_archive_tags(store: &S) { cursor, utxo: Default::default(), archive, + stake_addresses: Vec::new(), }; let writer = store.start_writer().expect("start_writer failed");