feat(cardano): log per-account epoch stake distribution

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<u8>,
+}
+
+entity_boilerplate!(AccountStakeLog, "account-stakes");
+
 #[derive(Debug, Clone, PartialEq, Decode, Encode, Default)]
 pub struct StakeLog {
     /// Number of blocks created by pool

crates/cardano/src/model/mod.rs

@@ -75,6 +75,7 @@ pub enum CardanoEntity {
     DatumState(Box<DatumState>),
     PendingRewardState(Box<PendingRewardState>),
     PendingMirState(Box<PendingMirState>),
+    AccountStakeLog(Box<AccountStakeLog>),
 }
 
 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<Self, ChainError> {
@@ -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);

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

crates/cardano/src/rupd/mod.rs

@@ -93,6 +93,14 @@ pub struct StakeSnapshot {
     pub registered_accounts: HashSet<StakeCredential>,
     pub pools: HashMap<PoolHash, EpochValue<PoolSnapshot>>,
     pub pool_stake: HashMap<PoolHash, u64>,
+    /// 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<PoolHash, u64>,
     /// 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<Item = (&PoolHash, &StakeCredential, &u64)> {
         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)]