From 65a5d615c8c1b8afbbb71bb0664d0dcd3750bf57 Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Tue, 18 Aug 2026 12:09:01 +0200 Subject: [PATCH 1/6] 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 970f36eef..82b8e8100 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, CardanoStateIndexExt, SlotOrder}, - model::{AccountState, DRepState}, + model::{AccountState, DRepState, PoolState}, pallas_extras, AccountEpochLog, ChainSummary, FixedNamespace, PoolHash, }; use dolos_core::async_query::BlockMetaResolver; @@ -224,6 +224,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, @@ -231,35 +286,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 { @@ -267,37 +335,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)) } @@ -1760,15 +1830,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 @@ -1777,11 +1879,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 977e0cd8523fb4c68ec05f157ecb0cc3e91946b3 Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Fri, 11 Sep 2026 11:03:52 +0200 Subject: [PATCH 2/6] feat(archive): 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. The log is a projection of the block history, so it lives in the archive store beside the archive tags and the exact lookups, and it is written through the same ArchiveWriter::apply_index call in the same batch as the blocks. - core: `ArchiveIndexDelta::stake_addresses` carries the candidates; `ArchiveStore::addresses_by_stake_log` reads a page and `mark_stake_log_ready` declares the log complete. - fjall: new `archive-stake-log` keyspace with pair entries (the write probe and undo key), ordered entries (the page read) and the ready marker. The writer keeps a batch-local seen-set because a batch cannot read its own pending inserts. The keyspace is not swept by prune_history: its entries are first appearances. - memory: same semantics, so ToyDomain tests exercise the log. - noop answers None. - The apply path emits appearances from `index_block`. The undo path rebuilds the same deltas, so a rollback removes exactly what apply inserted, and only when the undone block was the pair's first appearance. - Genesis bootstrap marks the log ready, before the state cursor, so a crash in between re-runs genesis. Stores restored from a stele or synced before the log answer None and the endpoint falls back to the archive scan until a resync. --- AGENTS.md | 6 +- crates/cardano/src/genesis/work_unit.rs | 10 +- crates/cardano/src/indexes/delta.rs | 89 +++++++- crates/cardano/src/indexes/mod.rs | 4 +- crates/core/src/archive.rs | 28 +++ crates/core/src/builtin/memory/archive.rs | 132 ++++++++++- crates/core/src/builtin/noop.rs | 14 ++ crates/core/src/indexes.rs | 23 +- crates/core/tests/block_meta.rs | 1 + crates/fjall/src/archive/README.md | 28 ++- crates/fjall/src/archive/mod.rs | 97 +++++++- crates/fjall/src/archive/stake_log.rs | 266 ++++++++++++++++++++++ crates/minibf/src/routes/accounts.rs | 108 +++++++++ crates/minibf/src/test_support.rs | 18 ++ crates/snapshot/tests/export.rs | 16 ++ crates/testing/src/faults.rs | 21 ++ crates/testing/src/measured.rs | 26 +++ docs/content/architecture/data-layer.mdx | 7 +- src/adapters/storage.rs | 36 +++ tests/archive_conformance.rs | 3 + tests/archive_index_roundtrip.rs | 115 +++++++++- tests/memory.rs | 1 + 22 files changed, 1023 insertions(+), 26 deletions(-) create mode 100644 crates/fjall/src/archive/stake_log.rs diff --git a/AGENTS.md b/AGENTS.md index 13d449c48..bfa930fc3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,7 @@ Dolos uses three distinct storage backends, each serving a specific purpose: ### ArchiveStore - **Purpose**: Historical block storage with temporal indexing -- **Contents**: Block bodies indexed by slot (one zstd frame per body in flat segment files, compressed with the dictionary bundled in `dolos-flatfiles`), entity logs keyed by `LogKey` (slot + entity key), and the lookups over them: archive tags (by address, payment, stake, policy, asset, datum, …) and exact lookups (by block hash, block number, tx hash), written in the same batch as the blocks they project +- **Contents**: Block bodies indexed by slot (one zstd frame per body in flat segment files, compressed with the dictionary bundled in `dolos-flatfiles`), entity logs keyed by `LogKey` (slot + entity key), and the lookups over them: archive tags (by address, payment, stake, policy, asset, datum, …), exact lookups (by block hash, block number, tx hash) and the stake address log (each account's addresses by first appearance), written in the same batch as the blocks they project - **Traits**: `ArchiveStore` (reads) + `ArchiveWriter` (batched writes) - **Database**: `/archive` (index plus flat block segment files) @@ -31,6 +31,7 @@ There is no standalone index store. Every index is a projection and lives in the store that holds what it projects: - the live-UTxO tags (by address, payment, stake, policy, asset, script ref) project the UTxO set and live in the `StateStore` (`StateStore::utxos_by_tag`, written through `StateWriter::apply_utxo_tags` in the same batch as the set) - the archive tags and the exact lookups (by block hash, block number, tx hash) project the block history and live in the `ArchiveStore` (`ArchiveStore::slots_by_tag` / `slot_by_*`, written through `ArchiveWriter::apply_index` in the same batch as the blocks) +- the stake address log (each `(stake, address)` pair once, at its first on-chain appearance) projects the block history too and lives in the `ArchiveStore` (`ArchiveStore::addresses_by_stake_log`, written through the same `ArchiveWriter::apply_index`); it is authoritative only on stores synced from genesis, which is what `mark_stake_log_ready` records ### Database File Organization @@ -105,11 +106,12 @@ The project follows a modular workspace architecture with clear separation of co - **`state-utxos`**: UTxO set storage with `[tx_hash:32][index:4]` keys - **`state-entities`**: All entity types with `[ns_hash:8][entity_key:32]` keys - **`state-tags`**: Live-UTxO tags with `[dim_hash:8][lookup_key:var][txo_ref:36]` keys - - `archive`: `ArchiveStore` implementation with four-keyspace design: + - `archive`: `ArchiveStore` implementation with five-keyspace design: - **`archive-blocks`**: Slot -> packed physical frame locations in the flat segment files - **`archive-logs`**: All log namespaces with `[ns_hash:8][log_key:40]` keys - **`archive-tags`**: Tag-based prefix scans for block tags with `[dim_hash:8][key_hash:8][slot:8]` keys - **`index-exact`**: Exact-match lookups with `[dim_hash:8][key_data:var]` -> `[slot:8]` + - **`archive-stake-log`**: Stake address log, `(stake, address)` pairs at first appearance, page reads by stake prefix from either end - `keys`: Shared key encoding utilities - **Key Advantages**: - Reduced segment files compared to per-entity keyspaces diff --git a/crates/cardano/src/genesis/work_unit.rs b/crates/cardano/src/genesis/work_unit.rs index 279f47f68..142f7b64c 100644 --- a/crates/cardano/src/genesis/work_unit.rs +++ b/crates/cardano/src/genesis/work_unit.rs @@ -5,7 +5,8 @@ use std::sync::Arc; use dolos_core::{ - config::CardanoConfig, ChainPoint, Domain, DomainError, Genesis, WalStore as _, WorkUnit, + config::CardanoConfig, ArchiveStore as _, ChainPoint, Domain, DomainError, Genesis, + WalStore as _, WorkUnit, }; use tracing::{debug, info}; @@ -53,6 +54,13 @@ where fn commit_state(&mut self, domain: &D, _shard_index: u32) -> Result<(), DomainError> { info!("bootstrapping chain from genesis"); + // A store that starts here sees every block it will ever hold flow + // through the apply path, so its stake address log is complete by + // construction. Marked before the state cursor — the flag that says + // genesis is done — so a crash in between re-runs genesis instead of + // leaving a store that never marks. + domain.archive().mark_stake_log_ready()?; + // Execute the genesis bootstrap super::execute::(domain.state(), &self.genesis, &self.config)?; diff --git a/crates/cardano/src/indexes/delta.rs b/crates/cardano/src/indexes/delta.rs index 0968a8635..e087dfde8 100644 --- a/crates/cardano/src/indexes/delta.rs +++ b/crates/cardano/src/indexes/delta.rs @@ -7,7 +7,8 @@ //! store). use dolos_core::{ - ArchiveIndexDelta, BlockSlot, EraCbor, Tag, TxoRef, UtxoIndexDelta, UtxoMap, UtxoSetDelta, + ArchiveIndexDelta, BlockSlot, EraCbor, StakeAddressAppearance, Tag, TxoRef, UtxoIndexDelta, + UtxoMap, UtxoSetDelta, }; use pallas::{ codec::minicbor, @@ -133,6 +134,7 @@ impl CardanoIndexDeltaBuilder { block_number: number, tx_hashes: Vec::new(), tags: Vec::new(), + stake_addresses: Vec::new(), }); } @@ -171,6 +173,34 @@ impl CardanoIndexDeltaBuilder { } } + /// Record an address appearing under its stake credential in the current + /// block, for the stake address log. Addresses without a stake credential + /// (Byron, enterprise) are not logged. + /// + /// `order` places the appearance inside the block; see + /// [`stake_appearance_order`]. + pub fn add_stake_appearance(&mut self, order: u32, addr: &Address) { + let stake = match addr { + 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 { + return; + }; + + self.current_block() + .stake_addresses + .push(StakeAddressAppearance { + order, + stake, + address: addr.to_vec(), + }); + } + /// Add asset tags to the current block. pub fn add_assets(&mut self, assets: &MultiEraValue) { let block = self.current_block(); @@ -291,7 +321,7 @@ impl CardanoIndexDeltaBuilder { self.start_block(block.slot(), block.hash().to_vec(), Some(block.number())); - for tx in block.txs() { + for (tx_order, tx) in block.txs().iter().enumerate() { self.add_tx_hash(tx.hash().to_vec()); for (label, _) in tx.metadata().collect::>() { @@ -315,9 +345,13 @@ impl CardanoIndexDeltaBuilder { } } - for (_, output) in tx.produces() { + for (output_order, output) in tx.produces() { if let Ok(addr) = output.address() { self.add_address(&addr); + self.add_stake_appearance( + stake_appearance_order(tx_order, output_order), + &addr, + ); } self.add_assets(&output.value()); if let Some(datum) = output.datum() { @@ -437,6 +471,13 @@ impl CardanoIndexDeltaBuilder { } } +/// The position of a produced output inside its block, as the stake address +/// log orders appearances within one slot: transaction index in the high 16 +/// bits, output index in the low 16. +pub fn stake_appearance_order(tx_order: usize, output_order: usize) -> u32 { + ((tx_order as u32) << 16) | (output_order as u32 & 0xffff) +} + /// The live-UTxO tag changes a `UtxoSetDelta` implies. /// /// Pairs with `StateWriter::apply_utxoset`: every ref that delta puts into the @@ -485,6 +526,48 @@ mod tests { assert_eq!(archive[0].tags.len(), 3); } + /// Only addresses with a stake credential feed the stake address log, and + /// the appearance carries the credential the endpoint queries by. + #[test] + fn stake_appearances_carry_the_stake_credential() { + use pallas::ledger::addresses::ByronAddress; + + let mut builder = CardanoIndexDeltaBuilder::new(); + builder.start_block(100, vec![0; 32], Some(50)); + + let shelley = test_shelley_address(); + builder.add_stake_appearance(stake_appearance_order(1, 2), &shelley); + + let enterprise = Address::Shelley(ShelleyAddress::new( + Network::Testnet, + ShelleyPaymentPart::Key([1; 28].as_slice().into()), + ShelleyDelegationPart::Null, + )); + builder.add_stake_appearance(stake_appearance_order(1, 3), &enterprise); + + let byron = Address::Byron(ByronAddress::new(&[0x82; 20], 0)); + builder.add_stake_appearance(stake_appearance_order(2, 0), &byron); + + let archive = builder.build(); + let appearances = &archive[0].stake_addresses; + + let Address::Shelley(shelley_inner) = &shelley else { + unreachable!() + }; + let stake = pallas_extras::shelley_address_to_stake_address(shelley_inner) + .expect("test address delegates to a key") + .to_vec(); + + assert_eq!( + appearances, + &[StakeAddressAppearance { + order: (1 << 16) | 2, + stake, + address: shelley.to_vec(), + }] + ); + } + /// An output that carries a reference script gets a `script_ref` tag whose /// key is the script's on-chain hash. #[test] diff --git a/crates/cardano/src/indexes/mod.rs b/crates/cardano/src/indexes/mod.rs index 9ea28411e..cdcec6eaf 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::{utxo_index_delta_from_utxo_delta, CardanoIndexDeltaBuilder}; +pub use delta::{ + stake_appearance_order, utxo_index_delta_from_utxo_delta, CardanoIndexDeltaBuilder, +}; pub use dimensions::{archive as archive_dimensions, utxo as utxo_dimensions}; pub use ext::{CardanoArchiveIndexExt, CardanoStateIndexExt}; pub use query::{AsyncCardanoQueryExt, ScriptData, ScriptLanguage, SlotOrder}; diff --git a/crates/core/src/archive.rs b/crates/core/src/archive.rs index b4570e780..991762d40 100644 --- a/crates/core/src/archive.rs +++ b/crates/core/src/archive.rs @@ -402,6 +402,34 @@ pub trait ArchiveStore: Clone + Send + Sync + 'static { end: BlockSlot, ) -> Result; + /// 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 was not + /// synced from genesis with the log in place (a stele restore, or a store + /// that predates the log). Callers fall back to an archive scan then. + /// + /// Entries come from `ArchiveIndexDelta::stake_addresses` through + /// [`ArchiveWriter::apply_index`]. [`ArchiveWriter::undo_index`] removes a + /// pair only when the undone block is its stored first appearance. + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError>; + + /// Declare the stake address log complete from genesis. + /// + /// Genesis bootstrap calls this once on a fresh store, before it writes + /// the state cursor that marks genesis as done. Until it runs, + /// [`ArchiveStore::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<(), ArchiveError>; + /// Iterate every archive tag record whose slot falls in `slots`. /// /// `slots` is **half-open** (`start..end`), unlike diff --git a/crates/core/src/builtin/memory/archive.rs b/crates/core/src/builtin/memory/archive.rs index 8aad1e8d4..5229d836f 100644 --- a/crates/core/src/builtin/memory/archive.rs +++ b/crates/core/src/builtin/memory/archive.rs @@ -18,7 +18,11 @@ //! keyed `(kind, key)`. The disk backend encodes `[dim_hash][key][slot]` //! triples into a flat keyspace and can only reach the contract's order by //! walking a caller-supplied dimension list in name order; keyed on the -//! triple, that order is the map's own and holds by construction. +//! triple, that order is the map's own and holds by construction; +//! - the stake address log lives in two maps: the ordered entries per stake +//! credential, keyed `(slot, order, address)` so a page read is a walk from +//! either end, and a membership map from `(stake, address)` to the pair's +//! first appearance, which is the write-path probe and the undo check. //! //! ## Stored key form //! @@ -46,8 +50,8 @@ use pallas::ledger::traverse::MultiEraBlock; use crate::archive::{ArchiveError, ArchiveStore, ArchiveWriter, LogKey, Skippable}; use crate::indexes::{ - key_hash, ArchiveIndexDelta, ExactKind, ExactRecord, IndexRecord, KeyHash, TagDimension, - TagRecord, MAX_EXACT_KEY_LEN, + key_hash, ArchiveIndexDelta, ExactKind, ExactRecord, IndexRecord, KeyHash, + StakeAddressAppearance, TagDimension, TagRecord, MAX_EXACT_KEY_LEN, }; use crate::{ BlockBody, BlockSlot, ChainPoint, EntityValue, Namespace, RawBlock, StateSchema, TemporalKey, @@ -61,6 +65,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. /// @@ -91,6 +101,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: first appearances ordered `(slot, order, address)` + /// per stake credential. + 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. @@ -106,6 +125,8 @@ enum Op { RemoveArchiveTag(ArchiveTag), InsertExact(ExactKind, ExactKey, BlockSlot), RemoveExact(ExactKind, ExactKey), + InsertStakeAddress(BlockSlot, StakeAddressAppearance), + RemoveStakeAddress(BlockSlot, StakeAddressAppearance), } fn poisoned() -> ArchiveError { @@ -150,6 +171,18 @@ impl MemoryArchiveStore { Ok(()) } + /// Forget the stake address log's ready marker, so queries answer `None` + /// and callers take their archive-scan fallback. + /// + /// For tests only: the toy domain runs genesis, which marks the log, and + /// a test of the fallback path needs a store where it is not marked. The + /// disk backend has no such switch — there the marker is genesis's alone. + pub fn clear_stake_log_ready(&self) -> Result<(), ArchiveError> { + let mut tables = self.tables.write().map_err(|_| poisoned())?; + tables.stake_log_ready = false; + Ok(()) + } + fn check_namespace(&self, ns: Namespace) -> Result<(), ArchiveError> { if !self.schema.contains_key(ns) { return Err(ArchiveError::NamespaceNotFound(ns)); @@ -244,6 +277,10 @@ impl ArchiveWriter for MemoryArchiveWriter { for tag in Self::archive_tags_of(block) { ops.push(Op::InsertArchiveTag(tag)); } + + for appearance in &block.stake_addresses { + ops.push(Op::InsertStakeAddress(block.slot, appearance.clone())); + } } Ok(()) @@ -263,6 +300,10 @@ impl ArchiveWriter for MemoryArchiveWriter { for tag in Self::archive_tags_of(block) { ops.push(Op::RemoveArchiveTag(tag)); } + + for appearance in block.stake_addresses.iter().rev() { + ops.push(Op::RemoveStakeAddress(block.slot, appearance.clone())); + } } Ok(()) @@ -316,6 +357,9 @@ impl ArchiveWriter for MemoryArchiveWriter { 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 { // An identical body means this block is being written again @@ -359,6 +403,45 @@ impl ArchiveWriter for MemoryArchiveWriter { Op::RemoveExact(kind, key) => { tables.exact.remove(&(kind, key)); } + // Only the first appearance of a pair is kept; the membership + // map is the probe, and it sees this batch's earlier inserts. + Op::InsertStakeAddress(slot, app) => { + let pair = (app.stake.clone(), app.address.clone()); + + if let std::collections::btree_map::Entry::Vacant(entry) = + tables.stake_log_pairs.entry(pair) + { + entry.insert((slot, app.order)); + tables.stake_log.entry(app.stake).or_default().insert(( + slot, + app.order, + app.address, + )); + } + } + // Removed only when the undone block is the pair's stored + // first appearance; a pair seen earlier stays untouched. + Op::RemoveStakeAddress(slot, app) => { + let pair = (app.stake.clone(), app.address.clone()); + + let Some(&(first_slot, order)) = tables.stake_log_pairs.get(&pair) else { + continue; + }; + + if first_slot != slot { + continue; + } + + 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); + } + } + } } } @@ -633,6 +716,10 @@ impl ArchiveStore for MemoryArchiveStore { .retain(|(_, _, slot)| *slot >= prune_before); tables.exact.retain(|_, slot| *slot >= prune_before); + // The stake address log is left alone, as on the disk backend: its + // entries are first appearances, so removing one below the cutoff + // would drop an address the account may still use. + Ok(done) } @@ -717,6 +804,45 @@ impl ArchiveStore for MemoryArchiveStore { Ok(MemorySlotIter(slots.into_iter())) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + 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: &StakeLogEntry| 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<(), ArchiveError> { + let mut tables = self.tables.write().map_err(|_| poisoned())?; + tables.stake_log_ready = true; + Ok(()) + } + fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/crates/core/src/builtin/noop.rs b/crates/core/src/builtin/noop.rs index 62709d42a..d3ff14cc7 100644 --- a/crates/core/src/builtin/noop.rs +++ b/crates/core/src/builtin/noop.rs @@ -224,6 +224,20 @@ impl ArchiveStore for NoOpArchiveStore { Ok(EmptySlotIter) } + fn addresses_by_stake_log( + &self, + _stake: &[u8], + _offset: usize, + _limit: usize, + _reverse: bool, + ) -> Result>>, ArchiveError> { + Ok(None) + } + + fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { + Ok(()) + } + /// Errors rather than yielding an empty iteration: this seam's callers /// publish the iterated records as a signed snapshot layer, and a /// well-formed *empty* layer from an index-less node is indistinguishable diff --git a/crates/core/src/indexes.rs b/crates/core/src/indexes.rs index 042f29602..67f2e7b08 100644 --- a/crates/core/src/indexes.rs +++ b/crates/core/src/indexes.rs @@ -13,7 +13,11 @@ //! - the archive tags and the exact lookups project the block history and live //! in the archive store (`ArchiveStore::slots_by_tag`, //! `ArchiveStore::slot_by_*`, written through `ArchiveWriter::apply_index`), -//! where the `indexes` stele layer is produced from and restored into. +//! where the `indexes` stele layer is produced from and restored into; +//! - the stake address log projects the block history too and lives beside them +//! (`ArchiveStore::addresses_by_stake_log`, written through the same +//! `ArchiveWriter::apply_index`). It is not part of the `indexes` stele +//! layer: a restored store answers `None` until it is synced from genesis. use std::borrow::Cow; @@ -75,6 +79,23 @@ pub struct ArchiveIndexDelta { pub block_number: Option, pub tx_hashes: Vec>, pub tags: Vec, + /// First-appearance candidates for the stake address log, one per + /// produced output that carries a stake credential, in block order. + pub stake_addresses: 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. The slot is the block's +/// (`ArchiveIndexDelta::slot`); `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 order: u32, + pub stake: Vec, + pub address: Vec, } /// What can go wrong building an index record. diff --git a/crates/core/tests/block_meta.rs b/crates/core/tests/block_meta.rs index c38b05094..ef2b01726 100644 --- a/crates/core/tests/block_meta.rs +++ b/crates/core/tests/block_meta.rs @@ -30,6 +30,7 @@ fn domain_with_block() -> (ToyDomain, TxHash) { block_number: Some(block.number()), tx_hashes: block.txs().iter().map(|tx| tx.hash().to_vec()).collect(), tags: Vec::new(), + stake_addresses: Vec::new(), }]) .unwrap(); writer.commit().unwrap(); diff --git a/crates/fjall/src/archive/README.md b/crates/fjall/src/archive/README.md index 6ba486a06..dc093e1d4 100644 --- a/crates/fjall/src/archive/README.md +++ b/crates/fjall/src/archive/README.md @@ -10,9 +10,10 @@ This module implements the `ArchiveStore` trait using [Fjall](https://github.com | 2 | `archive-logs` | All log namespaces | Range scans within a namespace | | 3 | `archive-tags` | Block tags, append-only | Prefix scans by dimension and key | | 4 | `index-exact` | Block hash / block number / tx hash → slot | Point lookups | +| 5 | `archive-stake-log` | Stake credential → addresses by first appearance | Prefix scans by stake, from either end | -The last two are projections of the blocks. Keeping them here lets the history -and its lookups commit in the same batch. Both keyspaces use +The last three are projections of the blocks. Keeping them here lets the history +and its lookups commit in the same batch. All three keyspaces use `l0_threshold = 8` and `memtable_size_mb = 128`. ## Key Schemas @@ -54,6 +55,27 @@ Value: (empty) Exact keys are stored verbatim and fixed-width per kind, so these records are lossless — `ExactRecord::new` is the single width-validation site. +### Stake address log (`archive-stake-log`) + +Three entry shapes, discriminated by a tag byte: + +| Entry | Key | Value | +|-------|-----|-------| +| Pair | `[0x00][stake_len:1][stake][address]` | `[slot:8][order:4]` | +| Ordered | `[0x01][stake_len:1][stake][slot:8][order:4][address]` | (empty) | +| Ready marker | `[0xff]` | `[1]` | + +Each `(stake, address)` pair is stored once, at its first on-chain appearance; +`order` is the transaction index in the high 16 bits and the output index in +the low 16. The pair entry is the write-path probe and the undo key; the +ordered entry is what a page read walks, from either end. Full address bytes +are stored so pointer addresses round-trip. + +Genesis bootstrap writes the ready marker. Until it exists +`addresses_by_stake_log` answers `None`, because a store restored from a stele +or synced before the log existed holds an incomplete log, and callers fall back +to an archive scan. + ## Dimension Hashing `hash_dimension()` in `keys.rs` computes the dimension hashes every index keyspace uses: @@ -84,4 +106,4 @@ both traversals share. ## Pruning -Neither `prune_history` nor `truncate_front` touches the two index keyspaces today: a rollback removes their entries through `ArchiveWriter::undo_index`, and pruning them under a sliding history window is not done yet. +`prune_history` sweeps `archive-tags` and `index-exact` (see `mod.rs`). It does not touch `archive-stake-log`: its entries are first appearances, so removing one below the cutoff would drop an address the account may still use. `truncate_front` touches none of the three; a rollback removes their entries through `ArchiveWriter::undo_index`. diff --git a/crates/fjall/src/archive/mod.rs b/crates/fjall/src/archive/mod.rs index c1802213e..ed855e219 100644 --- a/crates/fjall/src/archive/mod.rs +++ b/crates/fjall/src/archive/mod.rs @@ -3,12 +3,12 @@ //! The archive keeps block bodies in flat segment files ([`dolos_flatfiles`], //! one zstd frame per body, named by the frame's physical location) and //! holds the rows that point into the history — a blocks location table, -//! the derived-log namespaces, and the two index projections of the blocks +//! the derived-log namespaces, and the three index projections of the blocks //! — in an LSM tree. Behavior is //! pinned by the shared conformance suite (`tests/archive_conformance.rs`), //! with the builtin memory archive as the oracle. //! -//! ## Four Keyspace Design +//! ## Five Keyspace Design //! //! 1. **`archive-blocks`**: slot → packed 16-byte [`BlockLocation`]s, newest //! first. Key is the 8-byte big-endian slot; the value encoding is @@ -28,10 +28,14 @@ //! 4. **`index-exact`**: exact-match lookups (block hash, block number, tx hash //! → slot). Key: `[dim_hash:8][key_data:var]` → `[slot:8]`. See [`exact`]. //! -//! The two index keyspaces are projections of the blocks and are written in +//! 5. **`archive-stake-log`**: the stake address log, each `(stake, address)` +//! pair once at its first on-chain appearance, plus the ready marker genesis +//! writes. See [`stake_log`]. +//! +//! The three index keyspaces are projections of the blocks and are written in //! the same batch as the block locations, so the history and its lookups //! commit together. They keep the compaction settings the standalone index -//! store gave them. A rollback removes its entries through +//! store gave them. A rollback removes their entries through //! [`CoreArchiveWriter::undo_index`]; `truncate_front` does not touch them. //! //! ## Pruning the index keyspaces @@ -46,12 +50,16 @@ //! a sixteenth of the window since the last one. A node restored from a //! full-history stele pays one whole-keyspace scan on its first sweep. //! +//! The stake address log is not swept: its entries are first appearances, +//! so removing one below the cutoff would drop an address the account may +//! still use. +//! //! Unlike the redb writer, log batches are not reordered before insertion: //! shuffling exists to work around redb's half-split of ascending B-tree //! leaves, and an LSM memtable sorts its batch regardless of arrival order. use std::borrow::Cow; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::ops::{Bound, Range}; use std::path::Path; use std::sync::atomic::{AtomicU64, Ordering}; @@ -77,6 +85,7 @@ use crate::Error; pub mod exact; pub mod log_keys; pub mod scan; +pub mod stake_log; pub mod tags; use log_keys::{ @@ -115,6 +124,8 @@ mod keyspace_names { pub const TAGS: &str = "archive-tags"; /// Exact-match keyspace (block hash, tx hash, block number -> slot) pub const EXACT: &str = "index-exact"; + /// Stake address log keyspace (insert-once pairs, removed on rollback) + pub const STAKE_LOG: &str = "archive-stake-log"; } fn io_err(e: std::io::Error) -> ArchiveError { @@ -128,7 +139,7 @@ fn fjall_err(e: fjall::Error) -> ArchiveError { /// Fjall-based archive store. /// /// Block bodies live in flat segment files; the location table, the log -/// namespaces and the two index keyspaces live in the LSM tree. +/// namespaces and the three index keyspaces live in the LSM tree. #[derive(Clone)] pub struct ArchiveStore { db: Database, @@ -136,6 +147,7 @@ pub struct ArchiveStore { logs: Keyspace, tags: Keyspace, exact: Keyspace, + stake_log: Keyspace, flatfiles: Arc, schema: Arc, flush_on_commit: bool, @@ -246,6 +258,7 @@ impl ArchiveStore { }; let tags = db.keyspace(keyspace_names::TAGS, index_opts)?; let exact = db.keyspace(keyspace_names::EXACT, index_opts)?; + let stake_log = db.keyspace(keyspace_names::STAKE_LOG, index_opts)?; Ok(Self { db, @@ -253,6 +266,7 @@ impl ArchiveStore { logs, tags, exact, + stake_log, flatfiles: Arc::new(flatfiles), schema: Arc::new(schema), flush_on_commit, @@ -278,6 +292,7 @@ impl ArchiveStore { (keyspace_names::LOGS, &self.logs), (keyspace_names::TAGS, &self.tags), (keyspace_names::EXACT, &self.exact), + (keyspace_names::STAKE_LOG, &self.stake_log), ] .map(|(name, ks)| (name, ks.disk_space(), ks.path().to_path_buf())) .to_vec() @@ -293,6 +308,7 @@ impl ArchiveStore { self.logs.major_compact()?; self.tags.major_compact()?; self.exact.major_compact()?; + self.stake_log.major_compact()?; self.db.persist(PersistMode::SyncAll)?; Ok(()) @@ -352,8 +368,9 @@ impl ArchiveStore { } } - /// Remove every index entry below `prune_before` if the cutoff has moved - /// far enough since the last sweep to be worth a walk of both keyspaces. + /// Remove every tag and exact entry below `prune_before` if the cutoff + /// has moved far enough since the last sweep to be worth a walk of both + /// keyspaces. The stake address log is not swept (see the module docs). /// /// The threshold is a sixteenth of the retained window (at least one /// slot), so a sliding node walks its window-sized index about sixteen @@ -458,11 +475,16 @@ impl ArchiveStore { /// boundary) and consecutive `undo`s at one slot resolve against the /// overlay first and the committed state second — the reads redb gets for /// free from its transaction seeing its own writes. +/// +/// `stake_pairs_seen` plays the same role for the stake address log: the +/// pairs this writer has already inserted, since the batch cannot read its +/// own pending inserts and one batch spans many blocks. pub struct ArchiveWriter { store: ArchiveStore, batch: Mutex, pending_blocks: Mutex>, overlay: Mutex>>, + stake_pairs_seen: Mutex>>, #[cfg(test)] fail_index_commit: bool, } @@ -474,6 +496,7 @@ impl ArchiveWriter { store: store.clone(), pending_blocks: Mutex::new(Vec::new()), overlay: Mutex::new(HashMap::new()), + stake_pairs_seen: Mutex::new(HashSet::new()), #[cfg(test)] fail_index_commit: false, } @@ -564,6 +587,20 @@ impl CoreArchiveWriter for ArchiveWriter { exact::apply(&mut batch, &self.store.exact, deltas)?; tags::apply(&mut batch, &self.store.tags, deltas)?; + let snapshot = self.store.db.snapshot(); + let mut seen = self.stake_pairs_seen.lock().unwrap(); + + for block in deltas { + stake_log::apply( + &mut batch, + &self.store.stake_log, + &snapshot, + &mut seen, + block.slot, + &block.stake_addresses, + )?; + } + Ok(()) } @@ -573,6 +610,18 @@ impl CoreArchiveWriter for ArchiveWriter { exact::undo(&mut batch, &self.store.exact, deltas)?; tags::undo(&mut batch, &self.store.tags, deltas)?; + let snapshot = self.store.db.snapshot(); + + for block in deltas.iter().rev() { + stake_log::undo( + &mut batch, + &self.store.stake_log, + &snapshot, + block.slot, + &block.stake_addresses, + )?; + } + Ok(()) } @@ -1211,6 +1260,38 @@ impl CoreArchiveStore for ArchiveStore { .map_err(ArchiveError::from) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + let snapshot = self.db.snapshot(); + + // Without the ready marker the log is not authoritative: the store + // was not synced from genesis with the log in place. + if !stake_log::is_ready(&snapshot, &self.stake_log)? { + return Ok(None); + } + + let page = stake_log::page(&snapshot, &self.stake_log, stake, offset, limit, reverse)?; + + Ok(Some(page)) + } + + fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { + let mut batch = self.db.batch(); + stake_log::mark_ready(&mut batch, &self.stake_log); + + batch + .durability(Some(PersistMode::Buffer)) + .commit() + .map_err(fjall_err)?; + + Ok(()) + } + fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/crates/fjall/src/archive/stake_log.rs b/crates/fjall/src/archive/stake_log.rs new file mode 100644 index 000000000..e3b34f921 --- /dev/null +++ b/crates/fjall/src/archive/stake_log.rs @@ -0,0 +1,266 @@ +//! Stake address log operations for the `archive-stake-log` keyspace of the +//! archive store. +//! +//! The log answers one query: the addresses seen under a stake credential, +//! ordered by first on-chain appearance. Three 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. +//! - Ready marker: `[0xff]` -> `[1]`. Written once by genesis bootstrap. Reads +//! answer `None` until it exists, because a store that was not synced from +//! genesis with the log in place holds an incomplete log. +//! +//! Only the first appearance of a pair is stored. The write batch cannot read +//! its own pending inserts, so the writer threads a `seen` set through +//! [`apply`] to dedup pairs inside one batch; the pair entry dedups across +//! batches. +//! +//! The keyspace is not swept by `prune_history`: its entries are first +//! appearances, so removing one below the cutoff would drop an address the +//! account may still use. A rollback removes entries through [`undo`]. + +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; + +/// Key of the ready marker: a tag byte no pair or ordered entry can start +/// with, so it never lands inside a prefix scan. +const READY_KEY: &[u8] = &[0xff]; + +/// 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 in one block. +/// +/// `seen` dedups pairs inside the current write batch: the batch cannot +/// read its own pending inserts, and one batch spans many blocks. +pub fn apply( + batch: &mut OwnedWriteBatch, + keyspace: &Keyspace, + readable: &R, + seen: &mut HashSet>, + slot: BlockSlot, + 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(slot, appearance.order), + ); + + batch.insert( + keyspace, + build_ordered_key( + &appearance.stake, + slot, + appearance.order, + &appearance.address, + ), + [], + ); + + seen.insert(pair_key); + } + + Ok(()) +} + +/// Remove the 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, + slot: BlockSlot, + 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((first_slot, order)) = decode_sort_key(&value) else { + continue; + }; + + if first_slot != 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(); + + let mut push = |guard: fjall::Guard| -> Result<(), Error> { + let key = guard.key()?; + + if key.len() > header { + page.push(key[header..].to_vec()); + } + + Ok(()) + }; + + if reverse { + for guard in iter.rev().skip(offset).take(limit) { + push(guard)?; + } + } else { + for guard in iter.skip(offset).take(limit) { + push(guard)?; + } + } + + Ok(page) +} + +/// Whether genesis has declared the log complete. +pub fn is_ready(readable: &R, keyspace: &Keyspace) -> Result { + Ok(readable.get(keyspace, READY_KEY)?.is_some()) +} + +/// Write the ready marker. +pub fn mark_ready(batch: &mut OwnedWriteBatch, keyspace: &Keyspace) { + batch.insert(keyspace, READY_KEY, [1u8]); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ordered_keys_sort_by_slot_then_order() { + let stake = [0xaa; 29]; + let a = build_ordered_key(&stake, 1, 7, &[1; 57]); + let b = build_ordered_key(&stake, 2, 0, &[0; 57]); + let c = build_ordered_key(&stake, 2, 1, &[0; 57]); + + assert!(a < b, "slot dominates order"); + assert!(b < c, "order breaks ties within a slot"); + } + + #[test] + fn ordered_keys_stay_under_their_stake_prefix() { + let stake = [0xaa; 29]; + let other = [0xab; 29]; + let key = build_ordered_key(&stake, 5, 0, &[1; 57]); + + assert!(key.starts_with(&build_ordered_prefix(&stake))); + assert!(!key.starts_with(&build_ordered_prefix(&other))); + } + + #[test] + fn pair_and_ordered_entries_never_share_a_prefix() { + let stake = [0xaa; 29]; + let pair = build_pair_key(&stake, &[1; 57]); + + assert!(!pair.starts_with(&build_ordered_prefix(&stake))); + assert_ne!(pair.as_slice(), READY_KEY); + } + + #[test] + fn sort_key_round_trips() { + let encoded = encode_sort_key(141_868_807, 0x0003_0002); + assert_eq!(decode_sort_key(&encoded), Some((141_868_807, 0x0003_0002))); + assert_eq!(decode_sort_key(&encoded[..11]), None); + } +} diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index 82b8e8100..a9dd791c9 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -300,6 +300,38 @@ where return Err(StatusCode::NOT_FOUND.into()); } + // The stake address log answers both orders with one page read. It is + // authoritative only on stores synced from genesis with the log in place; + // `None` falls through to the archive scan below. The scan honors the + // 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 + .archive() + .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` @@ -1910,6 +1942,82 @@ 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_scan_fallback_orders_desc_by_first_appearance() { + // The fallback keeps Blockfrost's ordering on its own: `desc` is the + // reverse of `asc`, not a latest-appearance ordering. + let app = TestApp::new_scan_fallback(); + let stake_address = app.vectors().stake_address.as_str(); + + 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"); + + assert!(!asc.is_empty()); + + 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); + } + #[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 d3f5c8e26..67508c37e 100644 --- a/crates/minibf/src/test_support.rs +++ b/crates/minibf/src/test_support.rs @@ -116,6 +116,24 @@ impl TestApp { Self::new_with_cfg_and_fault(cfg, None) } + /// Like [`TestApp::new`], but with the stake address log marked not + /// authoritative, 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() + }; + + Self::new_with_cfg_and_setup(cfg, |domain, _| { + domain + .archive() + .clear_stake_log_ready() + .expect("failed to clear the stake log marker"); + }) + } + pub fn new_with_cfg_and_fault(cfg: SyntheticBlockConfig, fault: Option) -> Self { let (domain, vectors) = TestDomainBuilder::new_with_synthetic(cfg).finish(); Self::from_domain(domain, vectors, fault, None) diff --git a/crates/snapshot/tests/export.rs b/crates/snapshot/tests/export.rs index fc772c2c1..610780e0e 100644 --- a/crates/snapshot/tests/export.rs +++ b/crates/snapshot/tests/export.rs @@ -1592,6 +1592,21 @@ impl ArchiveStore for Counted { self.inner.iter_archive_tags(dimensions, slots) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + self.inner + .addresses_by_stake_log(stake, offset, limit, reverse) + } + + fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { + self.inner.mark_stake_log_ready() + } + fn iter_exact_records( &self, slots: std::ops::Range, @@ -1627,6 +1642,7 @@ fn index_across_the_skeleton() -> MemoryArchiveStore { .into_iter() .map(|dimension| dolos_core::Tag::new(dimension, vec![slot as u8; 28])) .collect(), + stake_addresses: Vec::new(), }); } } diff --git a/crates/testing/src/faults.rs b/crates/testing/src/faults.rs index 680d89317..8f71f4c12 100644 --- a/crates/testing/src/faults.rs +++ b/crates/testing/src/faults.rs @@ -355,6 +355,27 @@ impl ArchiveStore for FaultyArchiveStore { self.inner.slots_by_tag(dimension, key, start, end) } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + 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<(), ArchiveError> { + if self.should_fault() { + return Err(self.fault_err()); + } + self.inner.mark_stake_log_ready() + } + fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/crates/testing/src/measured.rs b/crates/testing/src/measured.rs index 0d05e4b54..5b4c2e3ec 100644 --- a/crates/testing/src/measured.rs +++ b/crates/testing/src/measured.rs @@ -248,6 +248,32 @@ impl ArchiveStore for MeasuredArchive { }) } + /// One tag candidate per address the page yields: the log stands in for + /// the tag scan the caller would otherwise run. + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + let page = self + .inner + .addresses_by_stake_log(stake, offset, limit, reverse)?; + + if let Some(page) = &page { + self.counters + .tag_candidates + .fetch_add(page.len() as u64, Ordering::Relaxed); + } + + Ok(page) + } + + fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { + self.inner.mark_stake_log_ready() + } + fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/docs/content/architecture/data-layer.mdx b/docs/content/architecture/data-layer.mdx index 3939a8d16..477ec04fc 100644 --- a/docs/content/architecture/data-layer.mdx +++ b/docs/content/architecture/data-layer.mdx @@ -47,7 +47,7 @@ The three storage modes exposed in configuration are really combinations of back `archive.backend = "no_op"` is the ledger-only switch, and it drops the historical lookups with the archive that hosts them. The live-UTxO tags are not affected: they project the UTxO set, so they stay in the state store in every mode (about 3.5 GB on mainnet). -Pruning is driven by `prune_history()` on the WAL and archive stores. In the archive it covers the block bodies, the logs, and the `archive-tags` and `index-exact` keyspaces, so a sliding-history node's index footprint is window-sized too. See the [configuration schema](../configuration/schema#storage-section) for the exact keys. +Pruning is driven by `prune_history()` on the WAL and archive stores. In the archive it covers the block bodies, the logs, and the `archive-tags` and `index-exact` keyspaces, so a sliding-history node's index footprint is window-sized too. The `archive-stake-log` keyspace is not pruned: it records each account address once, at its first appearance, and that appearance may lie before the window while the address is still in use. See the [configuration schema](../configuration/schema#storage-section) for the exact keys. ## Data model primitives @@ -58,12 +58,13 @@ A few concepts recur across the stores: - **`EpochValue` snapshot window.** Cardano staking reads state as it was several epochs ago. State entities that participate in staking are stored as a rotating window of snapshots (live / mark / set / go / next). The details are covered in the [Ledger Model](./ledger-model). - **Archive dual storage.** Block bodies are written to append-only flatfile segments (one segment per Cardano epoch) as one zstd frame per block, compressed with the dictionary bundled in `dolos-flatfiles`, while a compact index maps each slot to its `(segment, offset, length)` frame location. Every frame is written and synced before the location that names it is committed, so a crash leaves at most dead space at a segment's end and never a location pointing at bytes that are not there. Historical entity changes are stored as `(slot, entity-key) → value` logs, enabling range queries over time. - **Index dimensions.** Tag multimaps keyed by address, payment credential, stake credential, policy id, and asset id. The live ones point at the matching UTxOs and live in the state store's `state-tags` keyspace, written in the same batch as the UTxO set; the historical ones point at slots and live in the archive store's `archive-tags` keyspace, beside the `index-exact` keyspace that resolves a block hash, block number or transaction hash to its slot — both written in the same batch as the blocks they project. +- **Stake address log.** Each `(stake credential, address)` pair once, at its first on-chain appearance, in the archive store's `archive-stake-log` keyspace and written in the same batch as the blocks. It serves an account's address list in either order as one page read. The log is authoritative only on a node synced from genesis with it in place; a node restored from a stele answers from a scan of the archive instead until it is resynced. -Two fjall databases, eight keyspaces: +Two fjall databases, nine keyspaces: | Store | Keyspaces | One batch commits | |---|---|---| | state | `state-cursor`, `state-utxos`, `state-entities`, `state-tags` | entities + UTxO set + live-UTxO tags + cursor | -| archive | `archive-blocks`, `archive-logs`, `archive-tags`, `index-exact` | block locations + logs + archive tags + exact lookups | +| archive | `archive-blocks`, `archive-logs`, `archive-tags`, `index-exact`, `archive-stake-log` | block locations + logs + archive tags + exact lookups + stake address log | Source: `crates/core/src/{state,archive,wal,indexes,mempool}.rs`, `crates/redb3/src/*`, `crates/fjall/src/*`, and storage configuration in `crates/core/src/config.rs`. diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index 3c65bb9a5..84dc9bf40 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -1195,6 +1195,42 @@ impl CoreArchiveStore for ArchiveStoreBackend { } } + fn addresses_by_stake_log( + &self, + stake: &[u8], + offset: usize, + limit: usize, + reverse: bool, + ) -> Result>>, ArchiveError> { + match self { + Self::Memory(s) => { + CoreArchiveStore::addresses_by_stake_log(s, stake, offset, limit, reverse) + } + Self::LogsOnly(inner) => CoreArchiveStore::addresses_by_stake_log( + inner.as_ref(), + stake, + offset, + limit, + reverse, + ), + Self::Fjall(s) => { + CoreArchiveStore::addresses_by_stake_log(s, stake, offset, limit, reverse) + } + Self::NoOp(s) => { + CoreArchiveStore::addresses_by_stake_log(s, stake, offset, limit, reverse) + } + } + } + + fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { + match self { + Self::Memory(s) => CoreArchiveStore::mark_stake_log_ready(s), + Self::LogsOnly(inner) => CoreArchiveStore::mark_stake_log_ready(inner.as_ref()), + Self::Fjall(s) => CoreArchiveStore::mark_stake_log_ready(s), + Self::NoOp(s) => CoreArchiveStore::mark_stake_log_ready(s), + } + } + fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/tests/archive_conformance.rs b/tests/archive_conformance.rs index f9d5243ec..3118191ec 100644 --- a/tests/archive_conformance.rs +++ b/tests/archive_conformance.rs @@ -1048,6 +1048,7 @@ fn index_delta(slot: u64, seed: u8) -> ArchiveIndexDelta { block_number: Some(slot), tx_hashes: vec![vec![0x80 | seed; 32]], tags: vec![Tag::new(archive_dimensions::ADDRESS, vec![seed; 28])], + stake_addresses: Vec::new(), } } @@ -1144,6 +1145,7 @@ fn slots_by_tag_bounds_are_inclusive() { block_number: Some(slot), tx_hashes: Vec::new(), tags: vec![Tag::new(archive_dimensions::ADDRESS, key.clone())], + stake_addresses: Vec::new(), }) .collect(); @@ -1185,6 +1187,7 @@ fn slot_delta(slot: u64) -> ArchiveIndexDelta { archive_dimensions::ADDRESS, SHARED_TAG_KEY.to_vec(), )], + stake_addresses: Vec::new(), } } diff --git a/tests/archive_index_roundtrip.rs b/tests/archive_index_roundtrip.rs index 15623d68e..ca321ad34 100644 --- a/tests/archive_index_roundtrip.rs +++ b/tests/archive_index_roundtrip.rs @@ -34,7 +34,7 @@ use dolos_cardano::indexes::{archive_dimensions, CardanoArchiveIndexExt}; use dolos_core::{ builtin::MemoryArchiveStore, ArchiveIndexDelta, ArchiveStore as CoreArchiveStore, ArchiveWriter as CoreArchiveWriter, BlockSlot, ExactKind, ExactRecord, IndexRecord, - StateSchema, Tag, TagDimension, TagRecord, + StakeAddressAppearance, StateSchema, Tag, TagDimension, TagRecord, }; const EPOCH_LEN: BlockSlot = 432_000; @@ -210,6 +210,7 @@ fn seed_deltas(spec: &SeedSpec, sink: &mut impl FnMut(Vec)) - block_number: Some(epoch * 1_000_000 + b), tx_hashes, tags, + stake_addresses: Vec::new(), }); } @@ -330,6 +331,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>(); + } } }; } @@ -426,6 +432,7 @@ fn slots_by_tag_are_ordered_in_both_directions() { block_number: Some(slot), tx_hashes: Vec::new(), tags: vec![Tag::new(archive_dimensions::POLICY, policy.clone())], + stake_addresses: Vec::new(), }) .collect(); @@ -1341,6 +1348,7 @@ fn malformed_exact_keys_are_refused() { block_number: Some(1), tx_hashes: vec![vec![0xCD; 32]], tags: Vec::new(), + stake_addresses: Vec::new(), }]; let writer = store.start_writer().expect("start_writer failed"); @@ -1378,6 +1386,7 @@ fn malformed_exact_keys_are_refused() { block_number: Some(1), tx_hashes: vec![vec![0xEF; width]], tags: Vec::new(), + stake_addresses: Vec::new(), }]; let writer = store.start_writer().expect("start_writer failed"); @@ -1427,3 +1436,107 @@ 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 = |order: u32, stake: &[u8], address: u8| StakeAddressAppearance { + order, + stake: stake.to_vec(), + address: addr(address), + }; + + let delta = |slot: BlockSlot, items: Vec| ArchiveIndexDelta { + slot, + block_hash: hash32(0x0B, slot, 0), + block_number: Some(slot), + stake_addresses: items, + ..Default::default() + }; + + 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(&store, &[delta(1, vec![appearance(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 + apply( + &store, + &[ + delta(2, vec![appearance(0, &stake_a, 0x02)]), + delta( + 3, + vec![appearance(0, &stake_a, 0x02), appearance(1, &stake_b, 0x03)], + ), + ], + ); + + // a repeat in a later writer is deduped against the committed store + apply(&store, &[delta(4, vec![appearance(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() + ); + + // within one block, `order` decides: output 3 before output 7 + apply( + &store, + &[delta( + 5, + vec![appearance(7, &stake_b, 0x05), appearance(3, &stake_b, 0x04)], + )], + ); + assert_eq!( + page(&stake_b, 0, 10, false).unwrap(), + vec![addr(0x03), addr(0x04), addr(0x05)] + ); + + let undo_one = |d: ArchiveIndexDelta| { + let writer = store.start_writer().expect("start_writer failed"); + writer + .undo_index(std::slice::from_ref(&d)) + .expect("undo_index failed"); + writer.commit().expect("commit failed"); + }; + + // undoing the repeat leaves the pair in place + undo_one(delta(4, vec![appearance(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(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 1bcdba43e..180829726 100644 --- a/tests/memory.rs +++ b/tests/memory.rs @@ -378,6 +378,7 @@ fn seed_archive_tags(store: &S) { block_number: Some(b), tx_hashes: Vec::new(), tags, + stake_addresses: Vec::new(), }); } From aa347ecf004d890e5e92c7e34b386923227bf800 Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Fri, 11 Sep 2026 11:36:18 +0200 Subject: [PATCH 3/6] fix(minibf): ignore from/to on account addresses Blockfrost declares only count, page and order for /accounts/{stake_address}/addresses. Neither ryo nor mimicry binds from/to for it, so a windowed request answers the full list there. MiniBF honored them through the shared pagination struct and narrowed the scan, a quiet divergence. Scan the whole history like the sibling handlers that take no window, and drop the guard that kept windowed requests off the stake address log. A test checks that a window starting at the newest first appearance changes nothing on either the log path or the scan fallback. --- crates/minibf/src/routes/accounts.rs | 103 +++++++++++++++++++-------- 1 file changed, 75 insertions(+), 28 deletions(-) diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index a9dd791c9..916eecba2 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -279,6 +279,12 @@ where Ok(false) } +/// `GET /accounts/{stake_address}/addresses`. +/// +/// Blockfrost declares only `count`, `page` and `order` for this endpoint. +/// The shared query struct also accepts `from` and `to`; they are ignored +/// here, as Blockfrost ignores them, so the list always covers the account's +/// whole history. pub async fn by_stake_addresses( Path(stake_address): Path, Query(params): Query, @@ -302,37 +308,33 @@ where // The stake address log answers both orders with one page read. It is // authoritative only on stores synced from genesis with the log in place; - // `None` falls through to the archive scan below. The scan honors the - // 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 - .archive() - .addresses_by_stake_log( - &account_key.address.to_vec(), - pagination.skip(), - pagination.count, - matches!(pagination.order, Order::Desc), - ) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + // `None` falls through to the archive scan below. + let page = domain + .archive() + .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::, _>>()?; + 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)); - } + return Ok(Json(items)); } - let (start_slot, end_slot) = pagination.start_and_end_slots(&domain).await?; + let end_slot = domain.get_tip_slot()?; // Blockfrost orders addresses by first on-chain appearance, and `desc` // is the exact reverse of the `asc` list. Scan ascending in both cases; @@ -340,7 +342,7 @@ where // appearance instead of their first one. let stream = domain.query().blocks_by_stake_stream( &account_key.address.to_vec(), - start_slot, + 0, end_slot, SlotOrder::Asc, ); @@ -1985,6 +1987,51 @@ mod tests { } } + #[tokio::test] + async fn accounts_by_stake_addresses_ignores_from_and_to() { + // Blockfrost declares no `from`/`to` for this endpoint and its query + // binds neither, so a windowed request answers the full list. The + // window below starts at the newest first appearance: honoring it + // would drop every older address. + for app in [TestApp::new(), TestApp::new_scan_fallback()] { + let stake_address = app.vectors().stake_address.as_str(); + + let newest_first_appearance = app + .vectors() + .account_address_bounds + .iter() + .map(|(_, min, _)| *min) + .max() + .expect("vectors carry account addresses"); + let last_block = app + .vectors() + .account_address_bounds + .iter() + .map(|(_, _, max)| *max) + .max() + .expect("vectors carry account addresses"); + + let (status, bytes) = app + .get_bytes(&format!("/accounts/{stake_address}/addresses")) + .await; + assert_eq!(status, StatusCode::OK); + let full: Vec = + serde_json::from_slice(&bytes).expect("failed to parse addresses"); + assert!(full.len() > 1, "the window must have something to drop"); + + let (status, bytes) = app + .get_bytes(&format!( + "/accounts/{stake_address}/addresses?from={newest_first_appearance}&to={last_block}" + )) + .await; + assert_eq!(status, StatusCode::OK); + let windowed: Vec = + serde_json::from_slice(&bytes).expect("failed to parse windowed addresses"); + + assert_eq!(windowed, full); + } + } + #[tokio::test] async fn accounts_by_stake_addresses_scan_fallback_orders_desc_by_first_appearance() { // The fallback keeps Blockfrost's ordering on its own: `desc` is the From e4232ddf2969cd4c5a3db1067ab38f1a73a7167a Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Fri, 11 Sep 2026 11:48:36 +0200 Subject: [PATCH 4/6] fix(minibf): scan account addresses newest-first for desc The fallback scan for stores without the stake address log walked the account's whole history ascending and reversed it, because a descending walk meets a reused address at its latest appearance first. Measured on a preview store: 20s for an account with 185k tagged blocks whose asc page answers in 70ms. Walk newest-first instead and ask the archive address tag where each address really belongs: its earliest tagged block is its first production, since an address is spent only after it is produced. Each address is looked up once, when the scan first meets it, and emitted when the scan reaches that block. A page of a many-address account now fills from recent blocks and stops; the two preview accounts above drop from 15-21s to under 0.12s with byte-identical bodies. Accounts with a handful of reused addresses still read their whole history in either order, as asc does on main; the stake address log is what answers those in one page read. --- crates/minibf/src/routes/accounts.rs | 187 ++++++++++++++++++++++----- 1 file changed, 155 insertions(+), 32 deletions(-) diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index 916eecba2..bb9a1a4fe 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -1,4 +1,7 @@ -use std::{collections::BTreeSet, ops::Deref}; +use std::{ + collections::{BTreeSet, HashMap, HashSet}, + ops::Deref, +}; use axum::{ extract::{Path, Query, State}, @@ -21,12 +24,14 @@ use blockfrost_openapi::models::{ }; use dolos_cardano::{ - indexes::{AsyncCardanoQueryExt, CardanoStateIndexExt, SlotOrder}, + indexes::{AsyncCardanoQueryExt, CardanoArchiveIndexExt, CardanoStateIndexExt, SlotOrder}, model::{AccountState, DRepState, PoolState}, pallas_extras, AccountEpochLog, ChainSummary, FixedNamespace, PoolHash, }; use dolos_core::async_query::BlockMetaResolver; -use dolos_core::{ArchiveStore as _, Domain, EntityKey, LogKey, StateStore as _, TemporalKey}; +use dolos_core::{ + ArchiveStore as _, BlockSlot, Domain, EntityKey, LogKey, StateStore as _, TemporalKey, +}; use futures_util::StreamExt; use pallas::{ codec::minicbor, @@ -335,32 +340,48 @@ where } let end_slot = domain.get_tip_slot()?; + let account = account_key.address.to_vec(); // 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(), - 0, - end_slot, - SlotOrder::Asc, - ); - - // `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, + // is the exact reverse of the `asc` list. Either scan stops once it + // holds the page window; `enforce_max_scan_limit` bounds that window. + let ordered = match pagination.order { + Order::Asc => first_appearances_asc(&domain, &account, end_slot, pagination.to()).await?, + Order::Desc => first_appearances_desc(&domain, &account, end_slot, pagination.to()).await?, }; - let account = account_key.address.to_vec(); + let items = ordered + .into_iter() + .skip(pagination.skip()) + .take(pagination.count) + .map(|address| AccountAddressesContentInner { address }) + .collect(); - let mut ordered = vec![]; - let mut seen = BTreeSet::new(); + Ok(Json(items)) +} +/// The account's addresses by first appearance, oldest first, at most +/// `target` of them. +/// +/// An ascending scan meets every address at its first appearance, so the +/// first time it sees one is the answer. +async fn first_appearances_asc( + domain: &Facade, + account: &[u8], + end_slot: BlockSlot, + target: usize, +) -> Result, StatusCode> +where + D: Domain + Clone + Send + Sync + 'static, +{ + let stream = domain + .query() + .blocks_by_stake_stream(account, 0, end_slot, SlotOrder::Asc); let mut stream = Box::pin(stream); + let mut ordered = vec![]; + let mut seen = BTreeSet::new(); + 'scan: while let Some(res) = stream.next().await { let (_slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; @@ -375,7 +396,7 @@ where .address() .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - if !address_belongs_to_account(&address, &account) { + if !address_belongs_to_account(&address, account) { continue; } @@ -384,25 +405,127 @@ where if seen.insert(address.clone()) { ordered.push(address); - if scan_target.is_some_and(|target| ordered.len() >= target) { + if ordered.len() >= target { break 'scan; } } } } - if matches!(pagination.order, Order::Desc) { - ordered.reverse(); + Ok(ordered) +} + +/// The account's addresses by first appearance, newest first, at most +/// `target` of them. +/// +/// A descending scan meets a reused address at its latest appearance first. +/// The archive `address` tag tells where the address really belongs: its +/// earliest tagged block is its first production, because an address can +/// only be spent after it was produced. Each address is looked up once, when +/// the scan first meets it, and emitted when the scan reaches that block. +/// +/// The tag index keys on a 64-bit hash of the address, so a colliding hash +/// could hide an address. The odds per address are the number of tagged +/// addresses over 2^64. +async fn first_appearances_desc( + domain: &Facade, + account: &[u8], + end_slot: BlockSlot, + target: usize, +) -> Result, StatusCode> +where + D: Domain + Clone + Send + Sync + 'static, +{ + let stream = domain + .query() + .blocks_by_stake_stream(account, 0, end_slot, SlotOrder::Desc); + let mut stream = Box::pin(stream); + + let mut ordered = vec![]; + + // Every address the scan has met so far, with the slot of its first + // appearance: looked up once, then answered from here. + let mut first_seen: HashMap, BlockSlot> = HashMap::new(); + + 'scan: while let Some(res) = stream.next().await { + let (slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + let Some(block) = block else { + continue; + }; + + let block = MultiEraBlock::decode(&block).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + let mut fresh = vec![]; + let mut in_block: HashSet> = HashSet::new(); + + for (_, utxo) in block.txs().iter().flat_map(|tx| tx.produces()) { + let address = utxo + .address() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + if !address_belongs_to_account(&address, account) { + continue; + } + + let bytes = address.to_vec(); + + if !in_block.insert(bytes.clone()) { + continue; + } + + let first = match first_seen.get(&bytes) { + Some(first) => *first, + None => { + let first = first_appearance(domain, &bytes, slot)?; + first_seen.insert(bytes, first); + first + } + }; + + if first == slot { + fresh.push(address.to_string()); + } + } + + // Within one block the ascending order is output order; `desc` + // reverses it like everything else. + for address in fresh.into_iter().rev() { + ordered.push(address); + + if ordered.len() >= target { + break 'scan; + } + } } - let items = ordered - .into_iter() - .skip(pagination.skip()) - .take(pagination.count) - .map(|address| AccountAddressesContentInner { address }) - .collect(); + Ok(ordered) +} - Ok(Json(items)) +/// The slot of the earliest block that carries the archive `address` tag for +/// `address`, given that the block at `slot` does. +fn first_appearance( + domain: &Facade, + address: &[u8], + slot: BlockSlot, +) -> Result +where + D: Domain + Clone + Send + Sync + 'static, +{ + let Some(before) = slot.checked_sub(1) else { + return Ok(slot); + }; + + let mut slots = domain + .archive() + .slots_by_address(address, 0, before) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + match slots.next() { + Some(Ok(first)) => Ok(first), + Some(Err(_)) => Err(StatusCode::INTERNAL_SERVER_ERROR), + None => Ok(slot), + } } /// Fold one block's txs into an account's lifetime totals. From 32ce4c3bb20fcaa58dac136cc27954368f9bc190 Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Fri, 11 Sep 2026 15:09:05 +0200 Subject: [PATCH 5/6] refactor(minibf): serve account addresses from the stake log only The endpoint kept an archive-scan fallback for stores without the stake address log, gated by a ready marker that genesis wrote. The fallback is what the log exists to replace: an ascending scan that reverses for desc reads an account's whole history, and even the newest-first variant does so for accounts with a handful of reused addresses (226s for the heaviest preview account). Either the index is there or the endpoint is not worth serving from a scan. Drop the fallback and the marker: the handler is one log read in either order, `addresses_by_stake_log` answers a page rather than an `Option`, and `mark_stake_log_ready` goes with its seven implementations and the genesis hook. The log is populated by every path that applies blocks from genesis: relay sync, Mithril bootstrap, WAL catch-up, `doctor rebuild-state` and the snapshot backfill. A stele restore does not write it, because the `indexes` stele layer carries tag and exact records only; that is a known gap to close in the stele profile. --- AGENTS.md | 2 +- crates/cardano/src/genesis/work_unit.rs | 10 +- crates/core/src/archive.rs | 21 +- crates/core/src/builtin/memory/archive.rs | 31 +-- crates/core/src/builtin/noop.rs | 8 +- crates/core/src/indexes.rs | 3 +- crates/fjall/src/archive/README.md | 8 +- crates/fjall/src/archive/mod.rs | 25 +- crates/fjall/src/archive/stake_log.rs | 20 +- crates/minibf/src/routes/accounts.rs | 306 ++-------------------- crates/minibf/src/test_support.rs | 18 -- crates/snapshot/tests/export.rs | 6 +- crates/testing/src/faults.rs | 9 +- crates/testing/src/measured.rs | 14 +- docs/content/architecture/data-layer.mdx | 2 +- src/adapters/storage.rs | 11 +- tests/archive_index_roundtrip.rs | 39 +-- 17 files changed, 59 insertions(+), 474 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index bfa930fc3..aac5966de 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,7 @@ There is no standalone index store. Every index is a projection and lives in the store that holds what it projects: - the live-UTxO tags (by address, payment, stake, policy, asset, script ref) project the UTxO set and live in the `StateStore` (`StateStore::utxos_by_tag`, written through `StateWriter::apply_utxo_tags` in the same batch as the set) - the archive tags and the exact lookups (by block hash, block number, tx hash) project the block history and live in the `ArchiveStore` (`ArchiveStore::slots_by_tag` / `slot_by_*`, written through `ArchiveWriter::apply_index` in the same batch as the blocks) -- the stake address log (each `(stake, address)` pair once, at its first on-chain appearance) projects the block history too and lives in the `ArchiveStore` (`ArchiveStore::addresses_by_stake_log`, written through the same `ArchiveWriter::apply_index`); it is authoritative only on stores synced from genesis, which is what `mark_stake_log_ready` records +- the stake address log (each `(stake, address)` pair once, at its first on-chain appearance) projects the block history too and lives in the `ArchiveStore` (`ArchiveStore::addresses_by_stake_log`, written through the same `ArchiveWriter::apply_index`) ### Database File Organization diff --git a/crates/cardano/src/genesis/work_unit.rs b/crates/cardano/src/genesis/work_unit.rs index 142f7b64c..279f47f68 100644 --- a/crates/cardano/src/genesis/work_unit.rs +++ b/crates/cardano/src/genesis/work_unit.rs @@ -5,8 +5,7 @@ use std::sync::Arc; use dolos_core::{ - config::CardanoConfig, ArchiveStore as _, ChainPoint, Domain, DomainError, Genesis, - WalStore as _, WorkUnit, + config::CardanoConfig, ChainPoint, Domain, DomainError, Genesis, WalStore as _, WorkUnit, }; use tracing::{debug, info}; @@ -54,13 +53,6 @@ where fn commit_state(&mut self, domain: &D, _shard_index: u32) -> Result<(), DomainError> { info!("bootstrapping chain from genesis"); - // A store that starts here sees every block it will ever hold flow - // through the apply path, so its stake address log is complete by - // construction. Marked before the state cursor — the flag that says - // genesis is done — so a crash in between re-runs genesis instead of - // leaving a store that never marks. - domain.archive().mark_stake_log_ready()?; - // Execute the genesis bootstrap super::execute::(domain.state(), &self.genesis, &self.config)?; diff --git a/crates/core/src/archive.rs b/crates/core/src/archive.rs index 991762d40..e6fa417e3 100644 --- a/crates/core/src/archive.rs +++ b/crates/core/src/archive.rs @@ -406,29 +406,20 @@ pub trait ArchiveStore: Clone + Send + Sync + 'static { /// 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 was not - /// synced from genesis with the log in place (a stele restore, or a store - /// that predates the log). Callers fall back to an archive scan then. + /// exact reverse of it. An unknown credential is an empty page. /// /// Entries come from `ArchiveIndexDelta::stake_addresses` through - /// [`ArchiveWriter::apply_index`]. [`ArchiveWriter::undo_index`] removes a - /// pair only when the undone block is its stored first appearance. + /// [`ArchiveWriter::apply_index`], in the same batch as the blocks they + /// project. [`ArchiveWriter::undo_index`] removes a pair only when the + /// undone block is its stored first appearance. The log is as complete + /// as the history that was applied through that path. fn addresses_by_stake_log( &self, stake: &[u8], offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError>; - - /// Declare the stake address log complete from genesis. - /// - /// Genesis bootstrap calls this once on a fresh store, before it writes - /// the state cursor that marks genesis as done. Until it runs, - /// [`ArchiveStore::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<(), ArchiveError>; + ) -> Result>, ArchiveError>; /// Iterate every archive tag record whose slot falls in `slots`. /// diff --git a/crates/core/src/builtin/memory/archive.rs b/crates/core/src/builtin/memory/archive.rs index 5229d836f..b394f6d17 100644 --- a/crates/core/src/builtin/memory/archive.rs +++ b/crates/core/src/builtin/memory/archive.rs @@ -107,9 +107,6 @@ struct Tables { /// 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. @@ -171,18 +168,6 @@ impl MemoryArchiveStore { Ok(()) } - /// Forget the stake address log's ready marker, so queries answer `None` - /// and callers take their archive-scan fallback. - /// - /// For tests only: the toy domain runs genesis, which marks the log, and - /// a test of the fallback path needs a store where it is not marked. The - /// disk backend has no such switch — there the marker is genesis's alone. - pub fn clear_stake_log_ready(&self) -> Result<(), ArchiveError> { - let mut tables = self.tables.write().map_err(|_| poisoned())?; - tables.stake_log_ready = false; - Ok(()) - } - fn check_namespace(&self, ns: Namespace) -> Result<(), ArchiveError> { if !self.schema.contains_key(ns) { return Err(ArchiveError::NamespaceNotFound(ns)); @@ -810,15 +795,11 @@ impl ArchiveStore for MemoryArchiveStore { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { 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())); + return Ok(Vec::new()); }; let pick = |entry: &StakeLogEntry| entry.2.clone(); @@ -834,13 +815,7 @@ impl ArchiveStore for MemoryArchiveStore { set.iter().skip(offset).take(limit).map(pick).collect() }; - Ok(Some(page)) - } - - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - let mut tables = self.tables.write().map_err(|_| poisoned())?; - tables.stake_log_ready = true; - Ok(()) + Ok(page) } fn iter_archive_tags( diff --git a/crates/core/src/builtin/noop.rs b/crates/core/src/builtin/noop.rs index d3ff14cc7..6e539297a 100644 --- a/crates/core/src/builtin/noop.rs +++ b/crates/core/src/builtin/noop.rs @@ -230,12 +230,8 @@ impl ArchiveStore for NoOpArchiveStore { _offset: usize, _limit: usize, _reverse: bool, - ) -> Result>>, ArchiveError> { - Ok(None) - } - - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - Ok(()) + ) -> Result>, ArchiveError> { + Ok(Vec::new()) } /// Errors rather than yielding an empty iteration: this seam's callers diff --git a/crates/core/src/indexes.rs b/crates/core/src/indexes.rs index 67f2e7b08..0b7ae2c84 100644 --- a/crates/core/src/indexes.rs +++ b/crates/core/src/indexes.rs @@ -16,8 +16,7 @@ //! where the `indexes` stele layer is produced from and restored into; //! - the stake address log projects the block history too and lives beside them //! (`ArchiveStore::addresses_by_stake_log`, written through the same -//! `ArchiveWriter::apply_index`). It is not part of the `indexes` stele -//! layer: a restored store answers `None` until it is synced from genesis. +//! `ArchiveWriter::apply_index`). use std::borrow::Cow; diff --git a/crates/fjall/src/archive/README.md b/crates/fjall/src/archive/README.md index dc093e1d4..7b7c9ab99 100644 --- a/crates/fjall/src/archive/README.md +++ b/crates/fjall/src/archive/README.md @@ -57,13 +57,12 @@ Exact keys are stored verbatim and fixed-width per kind, so these records are lo ### Stake address log (`archive-stake-log`) -Three entry shapes, discriminated by a tag byte: +Two entry shapes, discriminated by a tag byte: | Entry | Key | Value | |-------|-----|-------| | Pair | `[0x00][stake_len:1][stake][address]` | `[slot:8][order:4]` | | Ordered | `[0x01][stake_len:1][stake][slot:8][order:4][address]` | (empty) | -| Ready marker | `[0xff]` | `[1]` | Each `(stake, address)` pair is stored once, at its first on-chain appearance; `order` is the transaction index in the high 16 bits and the output index in @@ -71,11 +70,6 @@ the low 16. The pair entry is the write-path probe and the undo key; the ordered entry is what a page read walks, from either end. Full address bytes are stored so pointer addresses round-trip. -Genesis bootstrap writes the ready marker. Until it exists -`addresses_by_stake_log` answers `None`, because a store restored from a stele -or synced before the log existed holds an incomplete log, and callers fall back -to an archive scan. - ## Dimension Hashing `hash_dimension()` in `keys.rs` computes the dimension hashes every index keyspace uses: diff --git a/crates/fjall/src/archive/mod.rs b/crates/fjall/src/archive/mod.rs index ed855e219..240717e58 100644 --- a/crates/fjall/src/archive/mod.rs +++ b/crates/fjall/src/archive/mod.rs @@ -29,8 +29,7 @@ //! → slot). Key: `[dim_hash:8][key_data:var]` → `[slot:8]`. See [`exact`]. //! //! 5. **`archive-stake-log`**: the stake address log, each `(stake, address)` -//! pair once at its first on-chain appearance, plus the ready marker genesis -//! writes. See [`stake_log`]. +//! pair once at its first on-chain appearance. See [`stake_log`]. //! //! The three index keyspaces are projections of the blocks and are written in //! the same batch as the block locations, so the history and its lookups @@ -1266,30 +1265,12 @@ impl CoreArchiveStore for ArchiveStore { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { let snapshot = self.db.snapshot(); - // Without the ready marker the log is not authoritative: the store - // was not synced from genesis with the log in place. - if !stake_log::is_ready(&snapshot, &self.stake_log)? { - return Ok(None); - } - let page = stake_log::page(&snapshot, &self.stake_log, stake, offset, limit, reverse)?; - Ok(Some(page)) - } - - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - let mut batch = self.db.batch(); - stake_log::mark_ready(&mut batch, &self.stake_log); - - batch - .durability(Some(PersistMode::Buffer)) - .commit() - .map_err(fjall_err)?; - - Ok(()) + Ok(page) } fn iter_archive_tags( diff --git a/crates/fjall/src/archive/stake_log.rs b/crates/fjall/src/archive/stake_log.rs index e3b34f921..e9a3b0fed 100644 --- a/crates/fjall/src/archive/stake_log.rs +++ b/crates/fjall/src/archive/stake_log.rs @@ -2,7 +2,7 @@ //! archive store. //! //! The log answers one query: the addresses seen under a stake credential, -//! ordered by first on-chain appearance. Three entry shapes share the +//! 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]`. @@ -11,9 +11,6 @@ //! - 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. -//! - Ready marker: `[0xff]` -> `[1]`. Written once by genesis bootstrap. Reads -//! answer `None` until it exists, because a store that was not synced from -//! genesis with the log in place holds an incomplete log. //! //! Only the first appearance of a pair is stored. The write batch cannot read //! its own pending inserts, so the writer threads a `seen` set through @@ -37,10 +34,6 @@ const PAIR_TAG: u8 = 0x00; /// Tag byte for ordered (page-read) entries. const ORDERED_TAG: u8 = 0x01; -/// Key of the ready marker: a tag byte no pair or ordered entry can start -/// with, so it never lands inside a prefix scan. -const READY_KEY: &[u8] = &[0xff]; - /// Width of the `[slot:8][order:4]` sort key. const SORT_KEY_SIZE: usize = 12; @@ -213,16 +206,6 @@ pub fn page( Ok(page) } -/// Whether genesis has declared the log complete. -pub fn is_ready(readable: &R, keyspace: &Keyspace) -> Result { - Ok(readable.get(keyspace, READY_KEY)?.is_some()) -} - -/// Write the ready marker. -pub fn mark_ready(batch: &mut OwnedWriteBatch, keyspace: &Keyspace) { - batch.insert(keyspace, READY_KEY, [1u8]); -} - #[cfg(test)] mod tests { use super::*; @@ -254,7 +237,6 @@ mod tests { let pair = build_pair_key(&stake, &[1; 57]); assert!(!pair.starts_with(&build_ordered_prefix(&stake))); - assert_ne!(pair.as_slice(), READY_KEY); } #[test] diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index bb9a1a4fe..55a5a54ed 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -1,7 +1,4 @@ -use std::{ - collections::{BTreeSet, HashMap, HashSet}, - ops::Deref, -}; +use std::{collections::BTreeSet, ops::Deref}; use axum::{ extract::{Path, Query, State}, @@ -24,14 +21,12 @@ use blockfrost_openapi::models::{ }; use dolos_cardano::{ - indexes::{AsyncCardanoQueryExt, CardanoArchiveIndexExt, CardanoStateIndexExt, SlotOrder}, + indexes::{AsyncCardanoQueryExt, CardanoStateIndexExt, SlotOrder}, model::{AccountState, DRepState, PoolState}, pallas_extras, AccountEpochLog, ChainSummary, FixedNamespace, PoolHash, }; use dolos_core::async_query::BlockMetaResolver; -use dolos_core::{ - ArchiveStore as _, BlockSlot, Domain, EntityKey, LogKey, StateStore as _, TemporalKey, -}; +use dolos_core::{ArchiveStore as _, Domain, EntityKey, LogKey, StateStore as _, TemporalKey}; use futures_util::StreamExt; use pallas::{ codec::minicbor, @@ -311,10 +306,10 @@ where return Err(StatusCode::NOT_FOUND.into()); } - // The stake address log answers both orders with one page read. It is - // authoritative only on stores synced from genesis with the log in place; - // `None` falls through to the archive scan below. - let page = domain + // Every page is one read of the stake address log, in either order. The + // log is written in the same batch as the blocks it projects, by every + // path that applies blocks from genesis. + let addresses = domain .archive() .addresses_by_stake_log( &account_key.address.to_vec(), @@ -324,210 +319,20 @@ where ) .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 end_slot = domain.get_tip_slot()?; - let account = account_key.address.to_vec(); - - // Blockfrost orders addresses by first on-chain appearance, and `desc` - // is the exact reverse of the `asc` list. Either scan stops once it - // holds the page window; `enforce_max_scan_limit` bounds that window. - let ordered = match pagination.order { - Order::Asc => first_appearances_asc(&domain, &account, end_slot, pagination.to()).await?, - Order::Desc => first_appearances_desc(&domain, &account, end_slot, pagination.to()).await?, - }; - - let items = ordered + let items = addresses .into_iter() - .skip(pagination.skip()) - .take(pagination.count) - .map(|address| AccountAddressesContentInner { address }) - .collect(); + .map(|bytes| { + Address::from_bytes(&bytes) + .map(|address| AccountAddressesContentInner { + address: address.to_string(), + }) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) + }) + .collect::, _>>()?; Ok(Json(items)) } -/// The account's addresses by first appearance, oldest first, at most -/// `target` of them. -/// -/// An ascending scan meets every address at its first appearance, so the -/// first time it sees one is the answer. -async fn first_appearances_asc( - domain: &Facade, - account: &[u8], - end_slot: BlockSlot, - target: usize, -) -> Result, StatusCode> -where - D: Domain + Clone + Send + Sync + 'static, -{ - let stream = domain - .query() - .blocks_by_stake_stream(account, 0, end_slot, SlotOrder::Asc); - let mut stream = Box::pin(stream); - - let mut ordered = vec![]; - let mut seen = BTreeSet::new(); - - 'scan: while let Some(res) = stream.next().await { - let (_slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - let Some(block) = block else { - continue; - }; - - 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 !address_belongs_to_account(&address, account) { - continue; - } - - let address = address.to_string(); - - if seen.insert(address.clone()) { - ordered.push(address); - - if ordered.len() >= target { - break 'scan; - } - } - } - } - - Ok(ordered) -} - -/// The account's addresses by first appearance, newest first, at most -/// `target` of them. -/// -/// A descending scan meets a reused address at its latest appearance first. -/// The archive `address` tag tells where the address really belongs: its -/// earliest tagged block is its first production, because an address can -/// only be spent after it was produced. Each address is looked up once, when -/// the scan first meets it, and emitted when the scan reaches that block. -/// -/// The tag index keys on a 64-bit hash of the address, so a colliding hash -/// could hide an address. The odds per address are the number of tagged -/// addresses over 2^64. -async fn first_appearances_desc( - domain: &Facade, - account: &[u8], - end_slot: BlockSlot, - target: usize, -) -> Result, StatusCode> -where - D: Domain + Clone + Send + Sync + 'static, -{ - let stream = domain - .query() - .blocks_by_stake_stream(account, 0, end_slot, SlotOrder::Desc); - let mut stream = Box::pin(stream); - - let mut ordered = vec![]; - - // Every address the scan has met so far, with the slot of its first - // appearance: looked up once, then answered from here. - let mut first_seen: HashMap, BlockSlot> = HashMap::new(); - - 'scan: while let Some(res) = stream.next().await { - let (slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - let Some(block) = block else { - continue; - }; - - let block = MultiEraBlock::decode(&block).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - let mut fresh = vec![]; - let mut in_block: HashSet> = HashSet::new(); - - for (_, utxo) in block.txs().iter().flat_map(|tx| tx.produces()) { - let address = utxo - .address() - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - if !address_belongs_to_account(&address, account) { - continue; - } - - let bytes = address.to_vec(); - - if !in_block.insert(bytes.clone()) { - continue; - } - - let first = match first_seen.get(&bytes) { - Some(first) => *first, - None => { - let first = first_appearance(domain, &bytes, slot)?; - first_seen.insert(bytes, first); - first - } - }; - - if first == slot { - fresh.push(address.to_string()); - } - } - - // Within one block the ascending order is output order; `desc` - // reverses it like everything else. - for address in fresh.into_iter().rev() { - ordered.push(address); - - if ordered.len() >= target { - break 'scan; - } - } - } - - Ok(ordered) -} - -/// The slot of the earliest block that carries the archive `address` tag for -/// `address`, given that the block at `slot` does. -fn first_appearance( - domain: &Facade, - address: &[u8], - slot: BlockSlot, -) -> Result -where - D: Domain + Clone + Send + Sync + 'static, -{ - let Some(before) = slot.checked_sub(1) else { - return Ok(slot); - }; - - let mut slots = domain - .archive() - .slots_by_address(address, 0, before) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - match slots.next() { - Some(Ok(first)) => Ok(first), - Some(Err(_)) => Err(StatusCode::INTERNAL_SERVER_ERROR), - None => Ok(slot), - } -} - /// Fold one block's txs into an account's lifetime totals. /// /// Produced outputs are received and resolved inputs are sent, the same split @@ -2067,56 +1872,14 @@ 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_ignores_from_and_to() { // Blockfrost declares no `from`/`to` for this endpoint and its query // binds neither, so a windowed request answers the full list. The // window below starts at the newest first appearance: honoring it // would drop every older address. - for app in [TestApp::new(), TestApp::new_scan_fallback()] { + { + let app = TestApp::new(); let stake_address = app.vectors().stake_address.as_str(); let newest_first_appearance = app @@ -2155,39 +1918,6 @@ mod tests { } } - #[tokio::test] - async fn accounts_by_stake_addresses_scan_fallback_orders_desc_by_first_appearance() { - // The fallback keeps Blockfrost's ordering on its own: `desc` is the - // reverse of `asc`, not a latest-appearance ordering. - let app = TestApp::new_scan_fallback(); - let stake_address = app.vectors().stake_address.as_str(); - - 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"); - - assert!(!asc.is_empty()); - - 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); - } - #[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 67508c37e..d3f5c8e26 100644 --- a/crates/minibf/src/test_support.rs +++ b/crates/minibf/src/test_support.rs @@ -116,24 +116,6 @@ impl TestApp { Self::new_with_cfg_and_fault(cfg, None) } - /// Like [`TestApp::new`], but with the stake address log marked not - /// authoritative, 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() - }; - - Self::new_with_cfg_and_setup(cfg, |domain, _| { - domain - .archive() - .clear_stake_log_ready() - .expect("failed to clear the stake log marker"); - }) - } - pub fn new_with_cfg_and_fault(cfg: SyntheticBlockConfig, fault: Option) -> Self { let (domain, vectors) = TestDomainBuilder::new_with_synthetic(cfg).finish(); Self::from_domain(domain, vectors, fault, None) diff --git a/crates/snapshot/tests/export.rs b/crates/snapshot/tests/export.rs index 610780e0e..799b94735 100644 --- a/crates/snapshot/tests/export.rs +++ b/crates/snapshot/tests/export.rs @@ -1598,15 +1598,11 @@ impl ArchiveStore for Counted { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { self.inner .addresses_by_stake_log(stake, offset, limit, reverse) } - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - self.inner.mark_stake_log_ready() - } - fn iter_exact_records( &self, slots: std::ops::Range, diff --git a/crates/testing/src/faults.rs b/crates/testing/src/faults.rs index 8f71f4c12..216aeb42a 100644 --- a/crates/testing/src/faults.rs +++ b/crates/testing/src/faults.rs @@ -361,7 +361,7 @@ impl ArchiveStore for FaultyArchiveStore { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { if self.should_fault() { return Err(self.fault_err()); } @@ -369,13 +369,6 @@ impl ArchiveStore for FaultyArchiveStore { .addresses_by_stake_log(stake, offset, limit, reverse) } - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - if self.should_fault() { - return Err(self.fault_err()); - } - self.inner.mark_stake_log_ready() - } - fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/crates/testing/src/measured.rs b/crates/testing/src/measured.rs index 5b4c2e3ec..7ab032841 100644 --- a/crates/testing/src/measured.rs +++ b/crates/testing/src/measured.rs @@ -256,24 +256,18 @@ impl ArchiveStore for MeasuredArchive { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { let page = self .inner .addresses_by_stake_log(stake, offset, limit, reverse)?; - if let Some(page) = &page { - self.counters - .tag_candidates - .fetch_add(page.len() as u64, Ordering::Relaxed); - } + self.counters + .tag_candidates + .fetch_add(page.len() as u64, Ordering::Relaxed); Ok(page) } - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - self.inner.mark_stake_log_ready() - } - fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/docs/content/architecture/data-layer.mdx b/docs/content/architecture/data-layer.mdx index 477ec04fc..f7f720f9b 100644 --- a/docs/content/architecture/data-layer.mdx +++ b/docs/content/architecture/data-layer.mdx @@ -58,7 +58,7 @@ A few concepts recur across the stores: - **`EpochValue` snapshot window.** Cardano staking reads state as it was several epochs ago. State entities that participate in staking are stored as a rotating window of snapshots (live / mark / set / go / next). The details are covered in the [Ledger Model](./ledger-model). - **Archive dual storage.** Block bodies are written to append-only flatfile segments (one segment per Cardano epoch) as one zstd frame per block, compressed with the dictionary bundled in `dolos-flatfiles`, while a compact index maps each slot to its `(segment, offset, length)` frame location. Every frame is written and synced before the location that names it is committed, so a crash leaves at most dead space at a segment's end and never a location pointing at bytes that are not there. Historical entity changes are stored as `(slot, entity-key) → value` logs, enabling range queries over time. - **Index dimensions.** Tag multimaps keyed by address, payment credential, stake credential, policy id, and asset id. The live ones point at the matching UTxOs and live in the state store's `state-tags` keyspace, written in the same batch as the UTxO set; the historical ones point at slots and live in the archive store's `archive-tags` keyspace, beside the `index-exact` keyspace that resolves a block hash, block number or transaction hash to its slot — both written in the same batch as the blocks they project. -- **Stake address log.** Each `(stake credential, address)` pair once, at its first on-chain appearance, in the archive store's `archive-stake-log` keyspace and written in the same batch as the blocks. It serves an account's address list in either order as one page read. The log is authoritative only on a node synced from genesis with it in place; a node restored from a stele answers from a scan of the archive instead until it is resynced. +- **Stake address log.** Each `(stake credential, address)` pair once, at its first on-chain appearance, in the archive store's `archive-stake-log` keyspace and written in the same batch as the blocks. It serves an account's address list in either order as one page read. Two fjall databases, nine keyspaces: diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index 84dc9bf40..6af230ff9 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -1201,7 +1201,7 @@ impl CoreArchiveStore for ArchiveStoreBackend { offset: usize, limit: usize, reverse: bool, - ) -> Result>>, ArchiveError> { + ) -> Result>, ArchiveError> { match self { Self::Memory(s) => { CoreArchiveStore::addresses_by_stake_log(s, stake, offset, limit, reverse) @@ -1222,15 +1222,6 @@ impl CoreArchiveStore for ArchiveStoreBackend { } } - fn mark_stake_log_ready(&self) -> Result<(), ArchiveError> { - match self { - Self::Memory(s) => CoreArchiveStore::mark_stake_log_ready(s), - Self::LogsOnly(inner) => CoreArchiveStore::mark_stake_log_ready(inner.as_ref()), - Self::Fjall(s) => CoreArchiveStore::mark_stake_log_ready(s), - Self::NoOp(s) => CoreArchiveStore::mark_stake_log_ready(s), - } - } - fn iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/tests/archive_index_roundtrip.rs b/tests/archive_index_roundtrip.rs index ca321ad34..858157de4 100644 --- a/tests/archive_index_roundtrip.rs +++ b/tests/archive_index_roundtrip.rs @@ -1437,9 +1437,9 @@ fn seeded_block() -> (Vec, u64, BlockSlot) { .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. +/// The stake address log conformance check: 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(); @@ -1467,13 +1467,8 @@ fn stake_log_round_trips_and_pages() { .expect("addresses_by_stake_log failed") }; - // before the marker the log is not authoritative, but writes still land apply(&store, &[delta(1, vec![appearance(0, &stake_a, 0x01)])]); - assert_eq!(page(&stake_a, 0, 10, false), None); - - store - .mark_stake_log_ready() - .expect("mark_stake_log_ready failed"); + assert_eq!(page(&stake_a, 0, 10, false), vec![addr(0x01)]); // one writer spanning two blocks dedups the pair the later block repeats apply( @@ -1490,23 +1485,20 @@ fn stake_log_round_trips_and_pages() { // a repeat in a later writer is deduped against the committed store apply(&store, &[delta(4, vec![appearance(0, &stake_a, 0x01)])]); - let asc = page(&stake_a, 0, 10, false).expect("log should be ready"); + let asc = page(&stake_a, 0, 10, false); 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"); + let desc = page(&stake_a, 0, 10, true); 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)]); + assert_eq!(page(&stake_a, 1, 1, false), vec![addr(0x02)]); + assert_eq!(page(&stake_a, 1, 1, true), 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() - ); + // stakes are isolated, and an unknown stake is an empty page + assert_eq!(page(&stake_b, 0, 10, false), vec![addr(0x03)]); + assert_eq!(page(&[0xCC; 29], 0, 10, false), Vec::>::new()); // within one block, `order` decides: output 3 before output 7 apply( @@ -1517,7 +1509,7 @@ fn stake_log_round_trips_and_pages() { )], ); assert_eq!( - page(&stake_b, 0, 10, false).unwrap(), + page(&stake_b, 0, 10, false), vec![addr(0x03), addr(0x04), addr(0x05)] ); @@ -1531,12 +1523,9 @@ fn stake_log_round_trips_and_pages() { // undoing the repeat leaves the pair in place undo_one(delta(4, vec![appearance(0, &stake_a, 0x01)])); - assert_eq!( - page(&stake_a, 0, 10, false).unwrap(), - vec![addr(0x01), addr(0x02)] - ); + assert_eq!(page(&stake_a, 0, 10, false), vec![addr(0x01), addr(0x02)]); // undoing the first appearance removes it undo_one(delta(1, vec![appearance(0, &stake_a, 0x01)])); - assert_eq!(page(&stake_a, 0, 10, false).unwrap(), vec![addr(0x02)]); + assert_eq!(page(&stake_a, 0, 10, false), vec![addr(0x02)]); } From a532795e8e170eb8caebf809a278de2e15f1bc6d Mon Sep 17 00:00:00 2001 From: slowbackspace Date: Fri, 11 Sep 2026 15:15:22 +0200 Subject: [PATCH 6/6] test(minibf): flatten the from/to test body --- crates/minibf/src/routes/accounts.rs | 72 ++++++++++++++-------------- 1 file changed, 35 insertions(+), 37 deletions(-) diff --git a/crates/minibf/src/routes/accounts.rs b/crates/minibf/src/routes/accounts.rs index 55a5a54ed..4d074e370 100644 --- a/crates/minibf/src/routes/accounts.rs +++ b/crates/minibf/src/routes/accounts.rs @@ -1878,44 +1878,42 @@ mod tests { // binds neither, so a windowed request answers the full list. The // window below starts at the newest first appearance: honoring it // would drop every older address. - { - let app = TestApp::new(); - let stake_address = app.vectors().stake_address.as_str(); + let app = TestApp::new(); + let stake_address = app.vectors().stake_address.as_str(); - let newest_first_appearance = app - .vectors() - .account_address_bounds - .iter() - .map(|(_, min, _)| *min) - .max() - .expect("vectors carry account addresses"); - let last_block = app - .vectors() - .account_address_bounds - .iter() - .map(|(_, _, max)| *max) - .max() - .expect("vectors carry account addresses"); - - let (status, bytes) = app - .get_bytes(&format!("/accounts/{stake_address}/addresses")) - .await; - assert_eq!(status, StatusCode::OK); - let full: Vec = - serde_json::from_slice(&bytes).expect("failed to parse addresses"); - assert!(full.len() > 1, "the window must have something to drop"); - - let (status, bytes) = app - .get_bytes(&format!( - "/accounts/{stake_address}/addresses?from={newest_first_appearance}&to={last_block}" - )) - .await; - assert_eq!(status, StatusCode::OK); - let windowed: Vec = - serde_json::from_slice(&bytes).expect("failed to parse windowed addresses"); - - assert_eq!(windowed, full); - } + let newest_first_appearance = app + .vectors() + .account_address_bounds + .iter() + .map(|(_, min, _)| *min) + .max() + .expect("vectors carry account addresses"); + let last_block = app + .vectors() + .account_address_bounds + .iter() + .map(|(_, _, max)| *max) + .max() + .expect("vectors carry account addresses"); + + let (status, bytes) = app + .get_bytes(&format!("/accounts/{stake_address}/addresses")) + .await; + assert_eq!(status, StatusCode::OK); + let full: Vec = + serde_json::from_slice(&bytes).expect("failed to parse addresses"); + assert!(full.len() > 1, "the window must have something to drop"); + + let (status, bytes) = app + .get_bytes(&format!( + "/accounts/{stake_address}/addresses?from={newest_first_appearance}&to={last_block}" + )) + .await; + assert_eq!(status, StatusCode::OK); + let windowed: Vec = + serde_json::from_slice(&bytes).expect("failed to parse windowed addresses"); + + assert_eq!(windowed, full); } #[tokio::test]