From 11a289f5fd1b71e0422692e6a7bfd3a8e7496105 Mon Sep 17 00:00:00 2001 From: Marek Mahut Date: Thu, 30 Jul 2026 11:12:08 +0200 Subject: [PATCH] feat(cardano): log per-account epoch stake distribution --- crates/cardano/src/model/logs.rs | 33 + crates/cardano/src/model/mod.rs | 5 + crates/cardano/src/rupd/loading.rs | 9 + crates/cardano/src/rupd/mod.rs | 12 + crates/cardano/src/rupd/work_unit.rs | 901 ++++++++++++++++++++++----- crates/core/src/work_unit.rs | 25 +- src/bin/dolos/data/dump_logs.rs | 57 +- 7 files changed, 877 insertions(+), 165 deletions(-) diff --git a/crates/cardano/src/model/logs.rs b/crates/cardano/src/model/logs.rs index c057e39e1..dbef45776 100644 --- a/crates/cardano/src/model/logs.rs +++ b/crates/cardano/src/model/logs.rs @@ -36,6 +36,39 @@ pub struct PoolDepositRefundLog { entity_boilerplate!(PoolDepositRefundLog, "pool-deposit-refunds"); +/// Per-account snapshot of the stake that was active during an epoch. +/// +/// Written by RUPD under the same temporal key as the per-pool [`StakeLog`] +/// (the epoch whose active stake the snapshot describes), keyed by the +/// account's credential, with the pool in the value. +/// +/// # Access patterns +/// +/// The key layout is chosen for the two cheap cases and accepts the third: +/// +/// - epoch-wide distribution — one prefix scan over the epoch's temporal key +/// - one account across epochs — one point read per epoch +/// - one pool within an epoch — **scans the epoch and filters on `pool_id`** +/// +/// The pool-scoped case is a deliberate, deferred tradeoff, not an oversight. +/// Keying by pool instead would turn the per-account lookup into a full scan, +/// and a composite `pool ++ credential` key does not fit the fixed-size entity +/// half of a `LogKey` without truncating one of them. A pool-keyed secondary +/// namespace would fix it at the cost of duplicating every row (~1.3M per +/// epoch on mainnet); revisit only if pool-scoped traffic justifies that. +#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode, Default)] +pub struct AccountStakeLog { + /// Active stake in Lovelaces + #[n(0)] + pub amount: u64, + + /// Pool the account delegated to at the snapshot epoch + #[n(1)] + pub pool_id: Vec, +} + +entity_boilerplate!(AccountStakeLog, "account-stakes"); + #[derive(Debug, Clone, PartialEq, Decode, Encode, Default)] pub struct StakeLog { /// Number of blocks created by pool diff --git a/crates/cardano/src/model/mod.rs b/crates/cardano/src/model/mod.rs index 57f22b0d5..b6566e803 100644 --- a/crates/cardano/src/model/mod.rs +++ b/crates/cardano/src/model/mod.rs @@ -75,6 +75,7 @@ pub enum CardanoEntity { DatumState(Box), PendingRewardState(Box), PendingMirState(Box), + AccountStakeLog(Box), } macro_rules! variant_boilerplate { @@ -110,6 +111,7 @@ variant_boilerplate!(StakeLog); variant_boilerplate!(DatumState); variant_boilerplate!(PendingRewardState); variant_boilerplate!(PendingMirState); +variant_boilerplate!(AccountStakeLog); impl dolos_core::Entity for CardanoEntity { fn decode_entity(ns: Namespace, value: &EntityValue) -> Result { @@ -130,6 +132,7 @@ impl dolos_core::Entity for CardanoEntity { DatumState::NS => DatumState::decode_entity(ns, value).map(Into::into), PendingRewardState::NS => PendingRewardState::decode_entity(ns, value).map(Into::into), PendingMirState::NS => PendingMirState::decode_entity(ns, value).map(Into::into), + AccountStakeLog::NS => AccountStakeLog::decode_entity(ns, value).map(Into::into), _ => Err(ChainError::InvalidNamespace(ns)), } } @@ -150,6 +153,7 @@ impl dolos_core::Entity for CardanoEntity { Self::DatumState(x) => DatumState::encode_entity(x), Self::PendingRewardState(x) => PendingRewardState::encode_entity(x), Self::PendingMirState(x) => PendingMirState::encode_entity(x), + Self::AccountStakeLog(x) => AccountStakeLog::encode_entity(x), } } } @@ -167,6 +171,7 @@ pub fn build_schema() -> StateSchema { schema.insert(MemberRewardLog::NS, NamespaceType::KeyValue); schema.insert(PoolDepositRefundLog::NS, NamespaceType::KeyValue); schema.insert(StakeLog::NS, NamespaceType::KeyValue); + schema.insert(AccountStakeLog::NS, NamespaceType::KeyValue); schema.insert(DatumState::NS, NamespaceType::KeyValue); schema.insert(PendingRewardState::NS, NamespaceType::KeyValue); schema.insert(PendingMirState::NS, NamespaceType::KeyValue); diff --git a/crates/cardano/src/rupd/loading.rs b/crates/cardano/src/rupd/loading.rs index b5cc87a3c..654f5a920 100644 --- a/crates/cardano/src/rupd/loading.rs +++ b/crates/cardano/src/rupd/loading.rs @@ -131,6 +131,15 @@ impl StakeSnapshot { .and_modify(|x| *x += stake) .or_insert(stake); + // Same reasoning for the delegator count, which `finalize` reports in + // the per-pool `StakeLog`: counting it here (the globals pass sees + // every account) keeps it correct after a mid-RUPD restart, which a + // per-shard tally would not be. + self.pool_delegator_counts + .entry(pool_id) + .and_modify(|x| *x += 1) + .or_insert(1); + self.active_stake_sum += stake; // The per-account map is shard-scoped: only credentials in this diff --git a/crates/cardano/src/rupd/mod.rs b/crates/cardano/src/rupd/mod.rs index d9b47d9df..d9224c35a 100644 --- a/crates/cardano/src/rupd/mod.rs +++ b/crates/cardano/src/rupd/mod.rs @@ -93,6 +93,14 @@ pub struct StakeSnapshot { pub registered_accounts: HashSet, pub pools: HashMap>, pub pool_stake: HashMap, + /// Per-pool delegator count over the *whole* snapshot, filled by + /// `load_globals` alongside `pool_stake`. + /// + /// Counted globally rather than summed from each shard's + /// `accounts_by_pool`, so it survives a mid-RUPD restart: `initialize` + /// rebuilds it, whereas a per-shard accumulator would only cover the + /// shards that actually ran after the resume cursor. + pub pool_delegator_counts: HashMap, /// Per-pool live pledge: sum of stake delegated to the pool by its /// declared owners, computed in `load_globals` over every account. /// Owner credentials can land in any shard, so a per-shard @@ -118,6 +126,10 @@ impl StakeSnapshot { pub fn iter_accounts(&self) -> impl Iterator { self.accounts_by_pool.iter_all() } + + pub fn get_pool_delegator_count(&self, pool: &PoolHash) -> u64 { + *self.pool_delegator_counts.get(pool).unwrap_or(&0) + } } #[derive(Debug)] diff --git a/crates/cardano/src/rupd/work_unit.rs b/crates/cardano/src/rupd/work_unit.rs index 28624a704..d52428d80 100644 --- a/crates/cardano/src/rupd/work_unit.rs +++ b/crates/cardano/src/rupd/work_unit.rs @@ -12,14 +12,21 @@ //! `define_rewards` over every pool but only emits rewards for in-range //! credentials, persists the in-range `PendingRewardState` entities, and //! emits a `RupdProgress` delta to advance `EpochState.rupd_progress`. -//! `finalize()` writes `EpochState.incentives` once and emits per-pool -//! `StakeLog` entries to the archive from the rolled-up shard -//! contributions. +//! Each shard also writes its own slice of the per-account +//! `AccountStakeLog` entries — from `commit_state`, committed to the +//! archive before the state transaction opens, while the shard's +//! snapshot is still in memory. +//! `finalize()` writes `EpochState.incentives` once and emits the per-pool +//! `StakeLog` entries, deriving their figures from the rebuilt snapshot +//! globals and the persisted `PendingRewardState` entities rather than from +//! any tally carried across shards. //! //! `PendingRewardState` writes are overwrite-by-key (idempotent), so a //! crashed shard can resume safely. The `RupdProgress` delta carries the //! same idempotency / ordering / total-mismatch guards as -//! `EWrapProgress` / `EStartProgress`. +//! `EWrapProgress` / `EStartProgress`. Everything a shard persists +//! therefore lands before its progress cursor advances and is idempotent +//! on replay — the invariant that makes skipping committed shards safe. use std::collections::HashMap; use std::sync::Arc; @@ -34,18 +41,82 @@ use crate::{ rewards::{Reward, RewardMap}, rupd::credential_to_key, shard::{shard_key_ranges, ACCOUNT_SHARDS}, - CardanoLogic, ChainPoint, EpochState, FixedNamespace, PendingRewardState, PoolHash, StakeLog, + AccountStakeLog, CardanoLogic, ChainPoint, EpochState, FixedNamespace, PendingRewardState, + PoolHash, StakeLog, }; -use super::RupdWork; +use super::{RupdWork, StakeSnapshot}; + +/// Sum the rewards this RUPD emitted, per pool, from the `PendingRewardState` +/// entities the shards persisted. +/// +/// Returns `(total_rewards, operator_share)` per pool, matching +/// [`crate::rewards::RewardMap::aggregate_pool_rewards`]: every reward counts +/// toward the pool total, and leader rewards additionally toward the operator +/// share. +/// +/// Derived from state rather than accumulated in memory across shards so a +/// mid-RUPD restart reports the full epoch: `initialize` starts a fresh work +/// unit and skips already-committed shards, so an in-memory tally would only +/// cover the shards that ran after the resume cursor. The entities live until +/// `Ewrap` consumes them at the epoch boundary, which is after `finalize`. +fn aggregate_pending_pool_rewards( + state: &S, +) -> Result, DomainError> { + let mut out: HashMap = HashMap::new(); + + for record in state.iter_entities_typed::(PendingRewardState::NS, None)? { + let (_, pending) = record?; + + // Iterate the two reward lists directly (`into_log_entries` would + // allocate an intermediate Vec per record — this loop visits every + // rewarded account on the network). + for (pool, value) in &pending.as_delegator { + let (total_rewards, _) = out.entry(*pool).or_insert((0, 0)); + *total_rewards = total_rewards.saturating_add(*value); + } + + for (pool, value) in &pending.as_leader { + let (total_rewards, operator_share) = out.entry(*pool).or_insert((0, 0)); + *total_rewards = total_rewards.saturating_add(*value); + *operator_share = operator_share.saturating_add(*value); + } + } + + Ok(out) +} + +/// Emit one [`AccountStakeLog`] per delegator held in `snapshot`. +/// +/// Keyed by `(temporal_key, credential)`, where `temporal_key` is the +/// epoch-start slot of the epoch whose active stake the snapshot describes — +/// the same key the per-pool [`StakeLog`] uses, so both views agree on the +/// epoch label by construction. +/// +/// The caller passes a shard-scoped snapshot: `accounts_by_pool` only holds +/// the credentials in the current shard's key range, so peak memory stays at +/// one shard's worth of delegators. Writes are overwrite-by-key, so re-running +/// a shard after a crash is idempotent. Returns the number of entries written. +fn write_account_stake_logs( + writer: &W, + temporal_key: &TemporalKey, + snapshot: &StakeSnapshot, +) -> Result { + let mut written = 0; + + for (pool, credential, stake) in snapshot.iter_accounts() { + let log = AccountStakeLog { + amount: *stake, + pool_id: pool.as_slice().to_vec(), + }; + + let log_key = LogKey::from((temporal_key.clone(), credential_to_key(credential))); + writer.write_log_typed(&log_key, &log)?; -/// Per-pool shard contribution rolled up across shards so `finalize` -/// can emit the per-pool `StakeLog` entries from the full epoch's data. -#[derive(Debug, Default, Clone, Copy)] -struct PoolLogShare { - total_rewards: u64, - operator_share: u64, - delegators_count: u64, + written += 1; + } + + Ok(written) } /// Sharded work unit for computing rewards at the stability window. @@ -77,12 +148,6 @@ pub struct RupdWorkUnit { /// Computed rewards for the currently-loaded shard. Replaced on /// each `compute()`. rewards: Option>, - - /// Per-pool reward / delegator-count totals accumulated across all - /// shards as they commit. Memory is O(pools) ≈ a few thousand - /// entries, independent of delegator count. `finalize()` reads - /// these to emit `StakeLog` entries with the full-epoch values. - pool_log_shares: HashMap, } impl RupdWorkUnit { @@ -94,7 +159,6 @@ impl RupdWorkUnit { start_shard: 0, work: None, rewards: None, - pool_log_shares: HashMap::new(), } } @@ -121,6 +185,202 @@ impl RupdWorkUnit { } self.rewards = None; } + + /// Persist everything one shard owns: its slice of the per-account stake + /// distribution, its pending rewards, and the progress cursor that + /// records the shard as done. + /// + /// The archive rows are committed before the state transaction is even + /// opened — see the comment on the archive block for why the cursor must + /// go last (and the state transaction is kept as short as possible while + /// the potentially large archive batch commits). + /// + /// Split out of the `WorkUnit::commit_state` phase so the ordering can be + /// exercised against fault-injecting stores: the trait method's + /// `Domain` bound is unusable from this crate's own + /// tests, because `dolos-testing` is a dev-dependency that links its own + /// instance of `dolos-cardano` and the two `CardanoLogic` types never + /// unify. Taking the two stores directly keeps the bounds on `dolos-core` + /// traits, which are shared. Returns the number of `AccountStakeLog` + /// entries written. + fn commit_shard( + &self, + state: &S, + archive: &A, + shard_index: u32, + ) -> Result { + let work = self + .work + .as_ref() + .ok_or_else(|| DomainError::Internal("rupd work not loaded".into()))?; + + let rewards = self + .rewards + .as_ref() + .ok_or_else(|| DomainError::Internal("rewards not computed".into()))?; + + // ---- Archive: per-account AccountStakeLog entries ---- + // + // Committed before the state transaction below is even opened. The + // ordering that matters is archive-commit before the state *commit* + // that advances `rupd_progress`: a resumed RUPD skips every already + // committed shard (`initialize` sets `start_shard` from + // `progress.committed`), and nothing re-derives these rows — the + // shard's `accounts_by_pool` only exists while the shard is loaded, + // so a crash between an advanced cursor and the archive commit would + // drop this shard's slice of the distribution for good. Committing + // archive-first inverts the failure into a harmless one — the shard + // re-runs and rewrites the same rows (overwrite-by-key). Running the + // whole batch before `start_writer` also keeps the state write + // transaction from sitting open while a large archive batch commits. + // + // No relevant epochs means no snapshot was loaded (pre-Shelley or the + // first few epochs), so there is no stake distribution to log. + let account_stake_logs = match work.relevant_epochs() { + Some((_, epoch)) => { + let start_of_epoch = ChainPoint::Slot(work.chain.epoch_start(epoch)); + let temporal_key = TemporalKey::from(&start_of_epoch); + + let archive_writer = archive.start_writer()?; + let written = + write_account_stake_logs(&archive_writer, &temporal_key, &work.snapshot)?; + archive_writer.commit()?; + + written + } + None => 0, + }; + + debug!( + shard = shard_index, + pending_count = rewards.len(), + "persisting pending rewards to state" + ); + + let writer = state.start_writer()?; + + // Persist this shard's pending rewards as PendingRewardState + // entities. Writes are overwrite-by-key, so a crashed shard + // re-run is idempotent. + for (credential, reward) in rewards.iter_pending() { + let key = credential_to_key(credential); + + let (as_leader, as_delegator) = match reward { + Reward::MultiPool(r) => ( + r.leader_rewards().collect(), + r.delegator_rewards().collect(), + ), + Reward::PreAllegra(r) => { + let (pool, value) = r.pool_and_value(); + if r.is_leader() { + (vec![(pool, value)], vec![]) + } else { + (vec![], vec![(pool, value)]) + } + } + }; + + let pending = PendingRewardState { + credential: credential.clone(), + is_spendable: reward.is_spendable(), + as_leader, + as_delegator, + }; + + writer.write_entity_typed(&key, &pending)?; + } + + // Apply the progress delta — advances EpochState.rupd_progress + // and captures total_shards on the first commit so a config + // change mid-RUPD can't break the in-flight pipeline. Read the + // current EpochState, apply the delta, and write back. The + // delta's idempotency / ordering / total-mismatch guards make + // this safe to repeat on crash recovery. + let epoch_key = dolos_core::EntityKey::from(crate::model::CURRENT_EPOCH_KEY); + let mut epoch_entity: Option = + state.read_entity_typed::(EpochState::NS, &epoch_key)?; + let mut progress_delta = crate::RupdProgress::new(shard_index, self.total_shards); + progress_delta.apply(&mut epoch_entity); + if let Some(epoch_state) = epoch_entity { + writer.write_entity_typed(&epoch_key, &epoch_state)?; + } + + writer.commit()?; + + Ok(account_stake_logs) + } + + /// Emit the per-pool `StakeLog` entries for the epoch this RUPD covers. + /// + /// Every field comes from data that `initialize()` rebuilds or that the + /// shards persisted, never from an in-memory tally spanning shards: + /// `total_stake` / `delegators_count` from the globals pass of the stake + /// snapshot, and the reward figures from the stored + /// `PendingRewardState` entities. A RUPD resumed mid-pipeline therefore + /// reports the whole epoch rather than only the shards that ran after the + /// resume cursor. + /// + /// Store-generic for the same reason as [`Self::commit_shard`]. + fn write_stake_logs( + &self, + state: &S, + archive: &A, + ) -> Result<(), DomainError> { + let work = self + .work + .as_ref() + .ok_or_else(|| DomainError::Internal("rupd work not loaded".into()))?; + + let Some((_, epoch)) = work.relevant_epochs() else { + return Ok(()); + }; + + let start_of_epoch = ChainPoint::Slot(work.chain.epoch_start(epoch)); + let temporal_key = TemporalKey::from(&start_of_epoch); + + let pool_rewards = aggregate_pending_pool_rewards(state)?; + + let snapshot = &work.snapshot; + let archive_writer = archive.start_writer()?; + + for (pool_hash, pool_state) in snapshot.pools.iter() { + let pool_id = EntityKey::from(pool_hash.as_slice()); + let pool_stake = snapshot.get_pool_stake(pool_hash); + let relative_size = if snapshot.active_stake_sum > 0 { + (pool_stake as f64) / snapshot.active_stake_sum as f64 + } else { + 0.0 + }; + let params = pool_state.go().map(|x| &x.params); + let declared_pledge = params.map(|x| x.pledge).unwrap_or(0); + let fixed_cost = params.map(|x| x.cost).unwrap_or(0); + let margin_cost = params.map(|x| x.margin.clone()); + let blocks_minted = pool_state.mark().map(|x| x.blocks_minted).unwrap_or(0) as u64; + + let (total_rewards, operator_share) = + pool_rewards.get(pool_hash).copied().unwrap_or((0, 0)); + + let log = StakeLog { + blocks_minted, + total_stake: pool_stake, + relative_size, + live_pledge: 0, + declared_pledge, + delegators_count: snapshot.get_pool_delegator_count(pool_hash), + total_rewards, + operator_share, + fixed_cost, + margin_cost, + }; + + let log_key = LogKey::from((temporal_key.clone(), pool_id)); + archive_writer.write_log_typed(&log_key, &log)?; + } + + archive_writer.commit()?; + + Ok(()) + } } impl WorkUnit for RupdWorkUnit @@ -220,106 +480,26 @@ where } fn commit_state(&mut self, domain: &D, shard_index: u32) -> Result<(), DomainError> { - let work = self - .work - .as_ref() - .ok_or_else(|| DomainError::Internal("rupd work not loaded".into()))?; - - let rewards = self - .rewards - .as_ref() - .ok_or_else(|| DomainError::Internal("rewards not computed".into()))?; + let account_stake_logs = + self.commit_shard(domain.state(), domain.archive(), shard_index)?; debug!( shard = shard_index, - pending_count = rewards.len(), - "persisting pending rewards to state" + account_stake_logs, "rupd shard state committed" ); - - let writer = domain.state().start_writer()?; - - // Persist this shard's pending rewards as PendingRewardState - // entities. Writes are overwrite-by-key, so a crashed shard - // re-run is idempotent. - for (credential, reward) in rewards.iter_pending() { - let key = credential_to_key(credential); - - let (as_leader, as_delegator) = match reward { - Reward::MultiPool(r) => ( - r.leader_rewards().collect(), - r.delegator_rewards().collect(), - ), - Reward::PreAllegra(r) => { - let (pool, value) = r.pool_and_value(); - if r.is_leader() { - (vec![(pool, value)], vec![]) - } else { - (vec![], vec![(pool, value)]) - } - } - }; - - let state = PendingRewardState { - credential: credential.clone(), - is_spendable: reward.is_spendable(), - as_leader, - as_delegator, - }; - - writer.write_entity_typed(&key, &state)?; - } - - // Apply the progress delta — advances EpochState.rupd_progress - // and captures total_shards on the first commit so a config - // change mid-RUPD can't break the in-flight pipeline. Read the - // current EpochState, apply the delta, and write back. The - // delta's idempotency / ordering / total-mismatch guards make - // this safe to repeat on crash recovery. - let epoch_key = dolos_core::EntityKey::from(crate::model::CURRENT_EPOCH_KEY); - let mut epoch_entity: Option = domain - .state() - .read_entity_typed::(EpochState::NS, &epoch_key)?; - let mut progress_delta = crate::RupdProgress::new(shard_index, self.total_shards); - progress_delta.apply(&mut epoch_entity); - if let Some(epoch_state) = epoch_entity { - writer.write_entity_typed(&epoch_key, &epoch_state)?; - } - - writer.commit()?; - - // Roll up this shard's per-pool reward + delegator-count - // contributions for the finalize-phase StakeLog write. Memory - // is O(pools), independent of delegator count. Build the - // contributions while `self.work` / `self.rewards` are - // immutably borrowed, then mutate `self.pool_log_shares` after - // the borrows end. - let pool_rewards = rewards.aggregate_pool_rewards(); - let mut shard_delegator_counts: HashMap = HashMap::new(); - for pool_hash in work.snapshot.pools.keys() { - let count = work.snapshot.accounts_by_pool.count_delegators(pool_hash); - if count > 0 { - shard_delegator_counts.insert(*pool_hash, count); - } - } - let _ = work; - let _ = rewards; - for (pool_hash, (total_rewards, operator_share)) in pool_rewards { - let entry = self.pool_log_shares.entry(pool_hash).or_default(); - entry.total_rewards = entry.total_rewards.saturating_add(total_rewards); - entry.operator_share = entry.operator_share.saturating_add(operator_share); - } - for (pool_hash, count) in shard_delegator_counts { - let entry = self.pool_log_shares.entry(pool_hash).or_default(); - entry.delegators_count = entry.delegators_count.saturating_add(count); - } - - debug!(shard = shard_index, "rupd shard state committed"); Ok(()) } fn commit_archive(&mut self, _domain: &D, _shard_index: u32) -> Result<(), DomainError> { - // Per-pool StakeLog entries are written in finalize() once the - // shard accumulator covers the full epoch's data. + // Per-pool StakeLog entries are written in finalize(), once every + // shard's pending rewards have landed in state. + // + // Per-account AccountStakeLog entries are written by commit_state, not + // here: this phase runs *after* the state commit that advances + // `rupd_progress`, and a resumed RUPD would skip the shard before + // these rows were ever written. This follows the durability-ordering + // rule documented on `WorkUnit::commit_state`, same as ewrap and + // estart; see the archive block in `commit_shard`. Ok(()) } @@ -353,59 +533,454 @@ where writer.commit()?; // ---- Archive: per-pool StakeLog entries ---- - // - // `pool_log_shares` was filled by each shard's `commit_state`. It - // already aggregates across shards, so we write one log per pool - // straight from the accumulator. - if let Some((_, epoch)) = work.relevant_epochs() { - let start_of_epoch = work.chain.epoch_start(epoch); - let start_of_epoch = ChainPoint::Slot(start_of_epoch); - let temporal_key = TemporalKey::from(&start_of_epoch); - - let snapshot = &work.snapshot; - let archive_writer = domain.archive().start_writer()?; - - for (pool_hash, pool_state) in snapshot.pools.iter() { - let pool_id = EntityKey::from(pool_hash.as_slice()); - let pool_stake = snapshot.get_pool_stake(pool_hash); - let relative_size = if snapshot.active_stake_sum > 0 { - (pool_stake as f64) / snapshot.active_stake_sum as f64 - } else { - 0.0 - }; - let params = pool_state.go().map(|x| &x.params); - let declared_pledge = params.map(|x| x.pledge).unwrap_or(0); - let fixed_cost = params.map(|x| x.cost).unwrap_or(0); - let margin_cost = params.map(|x| x.margin.clone()); - let blocks_minted = pool_state.mark().map(|x| x.blocks_minted).unwrap_or(0) as u64; - - let share = self - .pool_log_shares - .get(pool_hash) - .copied() - .unwrap_or_default(); - - let log = StakeLog { - blocks_minted, - total_stake: pool_stake, - relative_size, - live_pledge: 0, - declared_pledge, - delegators_count: share.delegators_count, - total_rewards: share.total_rewards, - operator_share: share.operator_share, - fixed_cost, - margin_cost, - }; - - let log_key = LogKey::from((temporal_key.clone(), pool_id)); - archive_writer.write_log_typed(&log_key, &log)?; - } - - archive_writer.commit()?; - } + self.write_stake_logs(domain.state(), domain.archive())?; debug!("rupd finalize committed"); Ok(()) } } + +#[cfg(test)] +mod tests { + use dolos_core::{ArchiveStore as _, Domain as _, EntityKey, LogKey, TemporalKey}; + use dolos_testing::{ + faults::{FaultyToyDomain, TestFault}, + toy_domain::ToyDomain, + }; + use pallas::{codec::minicbor, crypto::hash::Hash, ledger::primitives::StakeCredential}; + + use super::*; + + const EPOCH_SLOT: u64 = 4_000; + const OTHER_EPOCH_SLOT: u64 = 5_000; + + fn pool(byte: u8) -> PoolHash { + Hash::from([byte; 28]) + } + + fn key_cred(byte: u8) -> StakeCredential { + StakeCredential::AddrKeyhash(Hash::from([byte; 28])) + } + + fn script_cred(byte: u8) -> StakeCredential { + StakeCredential::ScriptHash(Hash::from([byte; 28])) + } + + /// Build a snapshot as `merge_shard` would leave it: per-account entries + /// for the shard's slice plus the pool-level totals from `load_globals`. + fn snapshot_with(entries: &[(PoolHash, StakeCredential, u64)]) -> StakeSnapshot { + let mut snapshot = StakeSnapshot::empty(); + + for (pool, credential, stake) in entries { + snapshot + .accounts_by_pool + .insert(*pool, credential.clone(), *stake); + *snapshot.pool_stake.entry(*pool).or_default() += *stake; + *snapshot.pool_delegator_counts.entry(*pool).or_default() += 1; + snapshot.active_stake_sum += *stake; + } + + snapshot + } + + fn write(domain: &ToyDomain, snapshot: &StakeSnapshot, slot: u64) -> usize { + let writer = domain.archive().start_writer().unwrap(); + let written = + write_account_stake_logs(&writer, &TemporalKey::from(slot), snapshot).unwrap(); + writer.commit().unwrap(); + + written + } + + /// Read back every log written under a single temporal key, exercising the + /// prefix scan an epoch-scoped query relies on. + fn read_epoch(domain: &D, slot: u64) -> Vec<(StakeCredential, AccountStakeLog)> { + let range = + LogKey::from(TemporalKey::from(slot))..LogKey::from(TemporalKey::from(slot + 1)); + + domain + .archive() + .iter_logs_typed::(AccountStakeLog::NS, Some(range)) + .unwrap() + .map(|entry| { + let (key, log) = entry.unwrap(); + let credential: StakeCredential = + minicbor::decode(EntityKey::from(key).as_ref()).unwrap(); + (credential, log) + }) + .collect() + } + + #[test] + fn writes_one_entry_per_delegator_keyed_by_credential() { + let domain = ToyDomain::new(None, None); + + let entries = [ + (pool(1), key_cred(0xa1), 100), + (pool(1), script_cred(0xa2), 250), + (pool(2), key_cred(0xa3), 75), + ]; + + let snapshot = snapshot_with(&entries); + assert_eq!(write(&domain, &snapshot, EPOCH_SLOT), entries.len()); + + let found = read_epoch(&domain, EPOCH_SLOT); + assert_eq!(found.len(), entries.len()); + + for (pool_hash, credential, stake) in entries.iter() { + let log = found + .iter() + .find(|(cred, _)| cred == credential) + .map(|(_, log)| log) + .unwrap_or_else(|| panic!("missing log for {credential:?}")); + + assert_eq!(log.amount, *stake); + assert_eq!(log.pool_id, pool_hash.as_slice()); + } + } + + #[test] + fn per_pool_sums_match_the_snapshot_pool_stake() { + let domain = ToyDomain::new(None, None); + + let snapshot = snapshot_with(&[ + (pool(1), key_cred(0xb1), 10), + (pool(1), key_cred(0xb2), 30), + (pool(2), key_cred(0xb3), 55), + ]); + + write(&domain, &snapshot, EPOCH_SLOT); + + let found = read_epoch(&domain, EPOCH_SLOT); + + for (pool_hash, expected) in snapshot.pool_stake.iter() { + let total: u64 = found + .iter() + .filter(|(_, log)| log.pool_id == pool_hash.as_slice()) + .map(|(_, log)| log.amount) + .sum(); + + assert_eq!(total, *expected, "pool {pool_hash} stake mismatch"); + } + } + + #[test] + fn zero_stake_delegators_are_kept() { + let domain = ToyDomain::new(None, None); + + let snapshot = snapshot_with(&[(pool(1), key_cred(0xc1), 0)]); + + assert_eq!(write(&domain, &snapshot, EPOCH_SLOT), 1); + + let found = read_epoch(&domain, EPOCH_SLOT); + assert_eq!(found.len(), 1); + assert_eq!(found[0].1.amount, 0); + } + + #[test] + fn entries_land_only_under_the_snapshot_epoch() { + let domain = ToyDomain::new(None, None); + + let snapshot = snapshot_with(&[(pool(1), key_cred(0xd1), 42)]); + write(&domain, &snapshot, EPOCH_SLOT); + + assert_eq!(read_epoch(&domain, EPOCH_SLOT).len(), 1); + assert!(read_epoch(&domain, OTHER_EPOCH_SLOT).is_empty()); + } + + #[test] + fn rerunning_a_shard_is_idempotent() { + let domain = ToyDomain::new(None, None); + + let snapshot = + snapshot_with(&[(pool(1), key_cred(0xe1), 10), (pool(1), key_cred(0xe2), 20)]); + + write(&domain, &snapshot, EPOCH_SLOT); + let first = read_epoch(&domain, EPOCH_SLOT); + + write(&domain, &snapshot, EPOCH_SLOT); + let second = read_epoch(&domain, EPOCH_SLOT); + + assert_eq!(first.len(), 2); + assert_eq!(first, second); + } + + #[test] + fn empty_snapshot_writes_nothing() { + let domain = ToyDomain::new(None, None); + + assert_eq!(write(&domain, &StakeSnapshot::empty(), EPOCH_SLOT), 0); + assert!(read_epoch(&domain, EPOCH_SLOT).is_empty()); + } + + // --- crash window between the archive rows and the progress cursor --- + + const CURRENT_EPOCH: u64 = 5; + const EPOCH_LENGTH: u64 = 100; + + /// Single Conway era, `epoch_length = 100`, so epoch boundaries land on + /// slots 0, 100, 200, ... and `first_shelley_epoch()` is 0 — enough for + /// `relevant_epochs()` to resolve at `CURRENT_EPOCH`. + fn test_chain_summary() -> crate::ChainSummary { + let mut summary = crate::ChainSummary::default(); + summary.append_era( + 7, + crate::model::EraSummary { + start: crate::model::EraBoundary { + epoch: 0, + slot: 0, + timestamp: 0, + }, + end: None, + epoch_length: EPOCH_LENGTH, + slot_length: 1, + protocol: 7, + }, + ); + + summary + } + + /// A RUPD work unit loaded as the executor would leave it after + /// `load` + `compute` for shard 0, carrying `snapshot` as that shard's + /// slice and no rewards to emit. + fn loaded_work_unit( + genesis: std::sync::Arc, + snapshot: StakeSnapshot, + ) -> (RupdWorkUnit, crate::ChainSummary) { + let chain = test_chain_summary(); + + let work = RupdWork { + current_epoch: CURRENT_EPOCH, + snapshot, + pots: Default::default(), + incentives: Default::default(), + blocks_made_total: 0, + max_supply: 0, + chain: test_chain_summary(), + pparams: None, + shard_ranges: None, + }; + + let rewards = RewardMap::::from_pending(Default::default(), Default::default()); + + let mut unit = RupdWorkUnit::new(chain.epoch_start(CURRENT_EPOCH) + 1, genesis); + unit.total_shards = ACCOUNT_SHARDS; + unit.work = Some(work); + unit.rewards = Some(rewards); + + (unit, chain) + } + + fn committed_shards(domain: &D) -> Option { + crate::load_epoch::(domain.state()) + .expect("epoch state") + .rupd_progress + .map(|progress| progress.committed) + } + + #[test] + fn committing_a_shard_persists_logs_and_advances_progress() { + let domain = FaultyToyDomain::new(ToyDomain::new(None, None), TestFault::None); + + let snapshot = + snapshot_with(&[(pool(1), key_cred(0xf1), 10), (pool(1), key_cred(0xf2), 20)]); + let (unit, chain) = loaded_work_unit(domain.genesis(), snapshot); + + assert_eq!(committed_shards(&domain), None); + + unit.commit_shard(domain.state(), domain.archive(), 0) + .expect("commit_shard"); + + // Logs land under the performance epoch — the same key `finalize` + // uses for the per-pool `StakeLog`. + let epoch_slot = chain.epoch_start(CURRENT_EPOCH - 1); + assert_eq!(read_epoch(&domain, epoch_slot).len(), 2); + assert_eq!(committed_shards(&domain), Some(1)); + } + + /// The window Codex flagged: if the archive rows landed *after* the state + /// commit, a failure here would advance `rupd_progress` past a shard whose + /// distribution was never written, and `initialize` would skip it on + /// restart — losing that slice for good. Committing the archive first makes + /// the failure leave the cursor untouched, so the shard simply re-runs. + #[test] + fn a_failed_archive_write_leaves_the_progress_cursor_untouched() { + let domain = FaultyToyDomain::new(ToyDomain::new(None, None), TestFault::ArchiveStoreError); + + let snapshot = snapshot_with(&[(pool(1), key_cred(0xf3), 10)]); + let (unit, _) = loaded_work_unit(domain.genesis(), snapshot); + + let result = unit.commit_shard(domain.state(), domain.archive(), 0); + + assert!(result.is_err(), "archive failure must abort the shard"); + assert_eq!( + committed_shards(&domain), + None, + "progress advanced past a shard whose logs were never written" + ); + } + + // --- per-pool StakeLog figures survive a mid-RUPD restart --- + + fn pool_params(pledge: u64, cost: u64) -> crate::PoolParams { + crate::PoolParams { + vrf_keyhash: Hash::from([0; 32]), + pledge, + cost, + margin: crate::pallas_extras::default_rational_number(), + reward_account: vec![], + pool_owners: vec![], + relays: vec![], + pool_metadata: None, + } + } + + /// A pool snapshot aligned to `CURRENT_EPOCH`: `go` carries the params + /// `StakeLog` reports, `mark` the blocks minted. + fn pool_snapshot( + pledge: u64, + cost: u64, + blocks_minted: u32, + ) -> crate::EpochValue { + let snapshot = |blocks_minted| crate::PoolSnapshot { + is_retired: false, + blocks_minted, + params: pool_params(pledge, cost), + is_new: false, + }; + + crate::EpochValue::from_parts( + CURRENT_EPOCH, + Some(snapshot(blocks_minted)), + None, + Some(snapshot(blocks_minted)), + Some(snapshot(blocks_minted)), + Some(snapshot(blocks_minted)), + ) + } + + fn seed_pending_reward( + domain: &D, + credential: StakeCredential, + as_leader: Vec<(PoolHash, u64)>, + as_delegator: Vec<(PoolHash, u64)>, + ) { + let pending = PendingRewardState { + credential: credential.clone(), + is_spendable: true, + as_leader, + as_delegator, + }; + + let writer = domain.state().start_writer().unwrap(); + writer + .write_entity_typed(&credential_to_key(&credential), &pending) + .unwrap(); + writer.commit().unwrap(); + } + + fn read_stake_log(domain: &D, slot: u64, pool: PoolHash) -> Option { + let log_key = LogKey::from((TemporalKey::from(slot), EntityKey::from(pool.as_slice()))); + + domain + .archive() + .read_log_typed::(StakeLog::NS, &log_key) + .unwrap() + } + + /// The bug three reviewers flagged: `finalize` used to read a per-pool + /// tally accumulated by each shard's `commit_state`, but a restart builds a + /// fresh work unit and `initialize` skips already-committed shards — so a + /// resumed RUPD wrote `StakeLog` with under-counted rewards and delegators, + /// or zeros if every shard had committed before the crash. + /// + /// This drives exactly that state: a work unit that ran no shards at all + /// (`start_shard == total_shards`), with the shards' `PendingRewardState` + /// already in state. + #[test] + fn stake_logs_are_complete_when_every_shard_committed_before_the_restart() { + let domain = ToyDomain::new(None, None); + + let pool = pool(7); + + // Exactly what `initialize()` leaves behind before any `load()`: + // pool-level globals, and an empty per-account map because no shard + // ran in this process. + let mut snapshot = StakeSnapshot::empty(); + snapshot.pools.insert(pool, pool_snapshot(100, 20, 3)); + snapshot.pool_stake.insert(pool, 1_000); + snapshot.pool_delegator_counts.insert(pool, 2); + snapshot.active_stake_sum = 1_000; + assert_eq!(snapshot.iter_accounts().count(), 0); + + // Rewards as the shards left them in state: one leader, one delegator. + seed_pending_reward(&domain, key_cred(0x01), vec![(pool, 30)], vec![]); + seed_pending_reward(&domain, key_cred(0x02), vec![], vec![(pool, 70)]); + + let (mut unit, chain) = loaded_work_unit(domain.genesis(), snapshot); + // Simulate the resume: every shard already committed, so none runs. + unit.start_shard = unit.total_shards; + + unit.write_stake_logs(domain.state(), domain.archive()) + .expect("write_stake_logs"); + + let log = read_stake_log(&domain, chain.epoch_start(CURRENT_EPOCH - 1), pool) + .expect("missing stake log"); + + assert_eq!(log.total_stake, 1_000); + assert_eq!(log.delegators_count, 2); + assert_eq!(log.total_rewards, 100); + assert_eq!(log.operator_share, 30); + assert_eq!(log.blocks_minted, 3); + assert_eq!(log.declared_pledge, 100); + assert_eq!(log.fixed_cost, 20); + } + + #[test] + fn pending_rewards_aggregate_per_pool_splitting_out_the_operator_share() { + let domain = ToyDomain::new(None, None); + + let first = pool(1); + let second = pool(2); + + seed_pending_reward( + &domain, + key_cred(0x11), + vec![(first, 5)], + vec![(second, 11)], + ); + seed_pending_reward(&domain, key_cred(0x12), vec![], vec![(first, 7)]); + + let aggregated = aggregate_pending_pool_rewards(domain.state()).expect("aggregate"); + + assert_eq!(aggregated.get(&first).copied(), Some((12, 5))); + assert_eq!(aggregated.get(&second).copied(), Some((11, 0))); + } + + #[test] + fn pending_rewards_aggregate_to_nothing_when_no_rewards_were_emitted() { + let domain = ToyDomain::new(None, None); + + let aggregated = aggregate_pending_pool_rewards(domain.state()).expect("aggregate"); + + assert!(aggregated.is_empty()); + } + + /// Pre-Shelley / early epochs load no snapshot, so there is nothing to + /// write — but the progress cursor must still advance, or the RUPD would + /// never complete. + #[test] + fn progress_advances_when_there_is_no_snapshot_to_log() { + let domain = FaultyToyDomain::new(ToyDomain::new(None, None), TestFault::None); + + let (mut unit, chain) = loaded_work_unit(domain.genesis(), StakeSnapshot::empty()); + unit.work.as_mut().unwrap().current_epoch = 1; + + unit.commit_shard(domain.state(), domain.archive(), 0) + .expect("commit_shard"); + + assert!(read_epoch(&domain, chain.epoch_start(0)).is_empty()); + assert_eq!(committed_shards(&domain), Some(1)); + } +} diff --git a/crates/core/src/work_unit.rs b/crates/core/src/work_unit.rs index 42974f732..2bfd4c5fe 100644 --- a/crates/core/src/work_unit.rs +++ b/crates/core/src/work_unit.rs @@ -40,6 +40,11 @@ pub struct MempoolUpdate { /// e. **Commit Archive** - Apply changes to the archive store. /// f. **Commit Indexes** - Apply changes to index stores. /// +/// Each commit phase owns its own transaction, so the order above is a +/// durability order, not an atomic one. Work units that persist a resume +/// cursor in `commit_state` must respect the ordering rule documented on +/// [`WorkUnit::commit_state`]. +/// /// 4. **Finalize** - Shard-agnostic teardown that runs once after the last /// shard's commits succeed. /// @@ -142,6 +147,21 @@ pub trait WorkUnit: Send { /// It should be called after `commit_wal()` to ensure crash recovery /// is possible. /// + /// # Durability ordering + /// + /// The commit phases name stores, not transactions: each one opens and + /// commits its own, so a crash between two phases leaves the earlier + /// store written and the later one not. A work unit that persists a + /// resume cursor here — see [`WorkUnit::start_shard`] — therefore has to + /// make everything that cursor implies durable *before* this phase's + /// transaction commits. Otherwise the restart skips a shard whose + /// remaining outputs never landed, and nothing re-derives them. + /// + /// So the shard-resumable Cardano work units (`ewrap`, `estart`, `rupd`) + /// write their per-shard archive logs from `commit_state`, ahead of the + /// state commit, and leave `commit_archive` empty. Work units without a + /// resume cursor (`roll`) are free to use the phases as named. + /// /// # Errors /// /// Returns an error if state persistence fails. @@ -150,7 +170,10 @@ pub trait WorkUnit: Send { /// Apply computed changes to the archive store. /// /// This phase persists historical data and logs to the archive. - /// It is called after `commit_state()`. + /// It is called after `commit_state()` — which means data that must be + /// durable before a resume cursor written in `commit_state` advances + /// cannot be written here. See the durability-ordering note on + /// [`WorkUnit::commit_state`]. /// /// # Errors /// diff --git a/src/bin/dolos/data/dump_logs.rs b/src/bin/dolos/data/dump_logs.rs index e0aea5a71..b354be81e 100644 --- a/src/bin/dolos/data/dump_logs.rs +++ b/src/bin/dolos/data/dump_logs.rs @@ -4,7 +4,7 @@ use comfy_table::Table; use dolos_cardano::{ eras::load_chain_summary_from_state, eras::log_epoch_range_to_key_range, - model::{LeaderRewardLog, MemberRewardLog, PParamKind}, + model::{AccountStakeLog, LeaderRewardLog, MemberRewardLog, PParamKind}, ChainSummary, EpochState, StakeLog, }; use dolos_core::config::RootConfig; @@ -154,6 +154,51 @@ impl TableRow for MemberRewardLog { } } +impl TableRow for AccountStakeLog { + fn header(format: OutputFormat) -> Vec<&'static str> { + match format { + OutputFormat::Default => vec!["slot", "stake", "pool", "amount"], + // Matches the `stake-{epoch}.csv` layout the epoch_pots harness + // compares against db-sync ground truth. + OutputFormat::Dbsync => vec!["stake", "pool", "lovelace"], + } + } + + fn row(&self, key: &LogKey, ctx: &RowContext) -> Vec { + let temporal = TemporalKey::from(key.clone()); + let entity = EntityKey::from(key.clone()); + let slot = u64::from_be_bytes(temporal.as_ref().try_into().unwrap()); + + // An undecodable key means the row isn't a credential-keyed log at all + // (wrong namespace, corruption). Report it as `` rather than + // falling back to a zero credential, which would render as a + // well-formed stake address and silently mislabel the account — and + // collapse every failed row onto the same address in a dbsync diff. + let stake = decode_stake_credential(&entity) + .ok() + .and_then(|credential| { + pallas_extras::stake_credential_to_address(ctx.network, &credential) + .to_bech32() + .ok() + }) + .unwrap_or_else(|| "".to_string()); + // bech32 happily encodes any byte length, so a corrupted pool id + // would render as a plausible-looking pool address — require the + // exact hash size before encoding. + let pool = if self.pool_id.len() == 28 { + bech32::encode::(POOL_HRP, &self.pool_id) + .unwrap_or_else(|_| "".to_string()) + } else { + "".to_string() + }; + + match ctx.format { + OutputFormat::Default => vec![slot.to_string(), stake, pool, self.amount.to_string()], + OutputFormat::Dbsync => vec![stake, pool, self.amount.to_string()], + } + } +} + impl TableRow for EpochState { fn header(format: OutputFormat) -> Vec<&'static str> { match format { @@ -590,6 +635,16 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { end_slot, range.clone(), )?, + "account-stakes" => dump_logs::( + &archive, + "account-stakes", + args.skip, + args.take, + &ctx, + start_slot, + end_slot, + range.clone(), + )?, "epochs" => dump_logs::( &archive, "epochs", args.skip, args.take, &ctx, start_slot, end_slot, range, )?,