diff --git a/AGENTS.md b/AGENTS.md index 13d449c48..aac5966de 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`) ### 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/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..e6fa417e3 100644 --- a/crates/core/src/archive.rs +++ b/crates/core/src/archive.rs @@ -402,6 +402,25 @@ 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. An unknown credential is an empty page. + /// + /// Entries come from `ArchiveIndexDelta::stake_addresses` through + /// [`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>; + /// 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..b394f6d17 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,12 @@ 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, } /// A single mutation, recorded by a writer and replayed at commit. @@ -106,6 +122,8 @@ enum Op { RemoveArchiveTag(ArchiveTag), InsertExact(ExactKind, ExactKey, BlockSlot), RemoveExact(ExactKind, ExactKey), + InsertStakeAddress(BlockSlot, StakeAddressAppearance), + RemoveStakeAddress(BlockSlot, StakeAddressAppearance), } fn poisoned() -> ArchiveError { @@ -244,6 +262,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 +285,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 +342,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 +388,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 +701,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 +789,35 @@ 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())?; + + let Some(set) = tables.stake_log.get(stake) else { + return Ok(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(page) + } + 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..6e539297a 100644 --- a/crates/core/src/builtin/noop.rs +++ b/crates/core/src/builtin/noop.rs @@ -224,6 +224,16 @@ impl ArchiveStore for NoOpArchiveStore { Ok(EmptySlotIter) } + fn addresses_by_stake_log( + &self, + _stake: &[u8], + _offset: usize, + _limit: usize, + _reverse: bool, + ) -> Result>, ArchiveError> { + Ok(Vec::new()) + } + /// 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..0b7ae2c84 100644 --- a/crates/core/src/indexes.rs +++ b/crates/core/src/indexes.rs @@ -13,7 +13,10 @@ //! - 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`). use std::borrow::Cow; @@ -75,6 +78,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..7b7c9ab99 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,21 @@ 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`) + +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) | + +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. + ## Dimension Hashing `hash_dimension()` in `keys.rs` computes the dimension hashes every index keyspace uses: @@ -84,4 +100,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..240717e58 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,13 @@ //! 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. 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 +49,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 +84,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 +123,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 +138,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 +146,7 @@ pub struct ArchiveStore { logs: Keyspace, tags: Keyspace, exact: Keyspace, + stake_log: Keyspace, flatfiles: Arc, schema: Arc, flush_on_commit: bool, @@ -246,6 +257,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 +265,7 @@ impl ArchiveStore { logs, tags, exact, + stake_log, flatfiles: Arc::new(flatfiles), schema: Arc::new(schema), flush_on_commit, @@ -278,6 +291,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 +307,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 +367,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 +474,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 +495,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 +586,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 +609,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 +1259,20 @@ 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(); + + let page = stake_log::page(&snapshot, &self.stake_log, stake, offset, limit, reverse)?; + + Ok(page) + } + 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..e9a3b0fed --- /dev/null +++ b/crates/fjall/src/archive/stake_log.rs @@ -0,0 +1,248 @@ +//! 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. Two entry shapes share the +//! keyspace, discriminated by a tag byte: +//! +//! - Pair entry: `[0x00][stake_len:1][stake][address]` -> `[slot:8][order:4]`. +//! One per known `(stake, address)` pair. This is the membership probe on the +//! write path and the undo key on rollback. +//! - Ordered entry: `[0x01][stake_len:1][stake][slot:8][order:4][address]` -> +//! empty. Lexicographic key order is chronological order, so a page read is a +//! prefix scan windowed from either end. +//! +//! Only the first appearance of a pair is stored. The write batch cannot read +//! its own pending inserts, so the 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; + +/// 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) +} + +#[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))); + } + + #[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 970f36eef..4d074e370 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,67 @@ 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) +} + +/// `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, @@ -231,72 +292,43 @@ 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?; - let stream = domain.query().blocks_by_stake_stream( - &account_key.address.to_vec(), - start_slot, - end_slot, - SlotOrder::from(pagination.order), - ); - - let mut items = vec![]; - let mut skipped = 0; - let mut seen = BTreeSet::new(); - - let mut stream = Box::pin(stream); - - while let Some(res) = stream.next().await { - if items.len() >= pagination.count { - break; - } - - let (_slot, block) = res.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - let Some(block) = block else { - continue; - }; + // 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(), + pagination.skip(), + pagination.count, + matches!(pagination.order, Order::Desc), + ) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - 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 items.len() >= pagination.count { - break; - } - } + 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::, _>>()?; Ok(Json(items)) } @@ -1760,15 +1792,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 +1841,81 @@ 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_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. + 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); + } + #[tokio::test] async fn accounts_by_stake_addresses_bad_request() { let app = TestApp::new(); diff --git a/crates/snapshot/tests/export.rs b/crates/snapshot/tests/export.rs index fc772c2c1..799b94735 100644 --- a/crates/snapshot/tests/export.rs +++ b/crates/snapshot/tests/export.rs @@ -1592,6 +1592,17 @@ 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 iter_exact_records( &self, slots: std::ops::Range, @@ -1627,6 +1638,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..216aeb42a 100644 --- a/crates/testing/src/faults.rs +++ b/crates/testing/src/faults.rs @@ -355,6 +355,20 @@ 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 iter_archive_tags( &self, dimensions: &[TagDimension], diff --git a/crates/testing/src/measured.rs b/crates/testing/src/measured.rs index 0d05e4b54..7ab032841 100644 --- a/crates/testing/src/measured.rs +++ b/crates/testing/src/measured.rs @@ -248,6 +248,26 @@ 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)?; + + self.counters + .tag_candidates + .fetch_add(page.len() as u64, Ordering::Relaxed); + + Ok(page) + } + 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..f7f720f9b 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. -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..6af230ff9 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -1195,6 +1195,33 @@ 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 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..858157de4 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,96 @@ fn seeded_block() -> (Vec, u64, BlockSlot) { .expect("the seed writes blocks") .clone() } + +/// 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(); + + 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") + }; + + apply(&store, &[delta(1, vec![appearance(0, &stake_a, 0x01)])]); + assert_eq!(page(&stake_a, 0, 10, false), vec![addr(0x01)]); + + // 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); + assert_eq!(asc, vec![addr(0x01), addr(0x02)]); + + // desc is the exact reverse of asc + 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), 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 + 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( + &store, + &[delta( + 5, + vec![appearance(7, &stake_b, 0x05), appearance(3, &stake_b, 0x04)], + )], + ); + assert_eq!( + page(&stake_b, 0, 10, false), + 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), 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), 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(), }); }