Prc/rewards fix

common/src/ledger_state.rs

@@ -33,6 +33,8 @@ pub struct SPOState {
     #[n(0)]
     pub pools: BTreeMap<KeyHash, PoolRegistration>,
     #[n(1)]
+    pub updates: BTreeMap<KeyHash, PoolRegistration>,
+    #[n(2)]
     pub retiring: BTreeMap<KeyHash, u64>,
 }
 

modules/accounts_state/Cargo.toml

@@ -24,6 +24,8 @@ bigdecimal = "0.4.8"
 rayon = "1.10.0"
 dashmap = "6.1.0"
 chrono = "0.4.41"
+csv = "1.3.1"
+itertools = "0.14.0"
 
 [lib]
 path = "src/accounts_state.rs"

modules/accounts_state/NOTES.md

@@ -80,7 +80,7 @@ ORDER BY pool_id_hex;
 
 ## Specific SPO test
 
-Epoch 212 (spendable in 213), SPO 30c6319d1f680..., rewards actually given out:
+Epoch 211 (spendable in 213), SPO 30c6319d1f680..., rewards actually given out:
 
 ```sql
 SELECT
@@ -94,24 +94,45 @@ AND encode(ph.hash_raw, 'hex') LIKE '30c6319d1f680%'
 GROUP BY ph.hash_raw;
 ```
 
+Note: pool_id for this SPO is 93
+
 |                       pool_id_hex                        | member_rewards | leader_rewards |
 |----------------------------------------------------------|----------------|----------------|
-| 30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54 |    33869550293 |     2164196243 |
+| 30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54 |   32024424770 |      2067130351 |
 
 Total 34091555121
 
 We have
 
 ```
-2025-08-21T13:59:50.578627Z  INFO acropolis_module_accounts_state::rewards: Pool 30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54 blocks=1 pool_stake=44180895641393 relative_pool_stake=0.001392062719472796345022132114111547444335115561171699064775592918376184270138741760710696148952284469 relative_blocks=0.0005022601707684580612757408337518834756403817177297840281265695630336514314414866901054746358613761929 pool_performance=1 optimum_rewards=34113076193 pool_rewards=34113076193
+2025-08-26T10:49:39.003335Z  INFO acropolis_module_accounts_state::rewards: Pool 30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54 blocks=0 pool_stake=44180895641393 relative_pool_stake=0.001392062719472796345022132114111547444335115561171699064775592918376184270
+138741760710696148952284469 relative_blocks=0 pool_performance=1 optimum_rewards=34091555158 pool_rewards=34091555158
 ```
 
-Optimum rewards: 34113076193
+Optimum rewards: 34091555158
+
+Difference: We are too high by 37 LL compared to DBSync - suspect rounding of individual payments
+We match the maxP from the Haskell node:
+
+```
+**** Calculating PoolRewardInfo: epoch=0, rewardInfo=PoolRewardInfo {poolRelativeStake = StakeShare (44180895641393 % 31737719158318701), poolPot = Coin 34091555158, poolPs = PoolParams {ppId = KeyHash {unKeyHash = "30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54"}, ppVrf = VRFVerKeyHash {unVRFVerKeyHash = "f2b08e8ec5fe945b41ece1c254e25843e35e574dd43535cbf244524019f704e9"}, ppPledge = Coin 50000000000, ppCost = Coin 340000000, ppMargin = 1 % 20, ppRewardAccount = RewardAccount {raNetwork = Mainnet, raCredential = KeyHashObj (KeyHash {unKeyHash = "8a10720c17ce32b75f489ed13fb706dac51c6006b7fee1a687f36620"})}, ppOwners = fromList [KeyHash {unKeyHash = "8a10720c17ce32b75f489ed13fb706dac51c6006b7fee1a687f36620"}], ppRelays = StrictSeq {fromStrict = fromList [SingleHostName (SJust (Port {portToWord16 = 3001})) (DnsName {dnsToText = "europe1-relay.jpn-sp.net"})]}, ppMetadata = SJust (PoolMetadata {pmUrl = Url {urlToText = "https://tokyostaker.com/metadata/jp3.json"}, pmHash = "\201\246\183K\128\&1 \EOT*\f\194\GS>B\168\136j\239\241\&4\189\230\175\SI4\163\160P\206\162\163]"})}, poolBlocks = 1, poolLeaderReward = LeaderOnlyReward {lRewardPool = KeyHash {unKeyHash = "30c6319d1f680470c8d2d48f8d44fd2848fa9b8cd6ac944d4dfc0c54"}, lRewardAmount = Coin 2067130351}}, activeStake=Coin 10177811974822904, totalStake=Coin 31737719158318701, pledgeRelative=50000000000 % 31737719158318701, sigmaA=44180895641393 % 10177811974822904, maxP=34091555158, appPerf=1 % 1, R=Coin 31834688329017****
+```
 
-Difference: We are too high by 21521072, or 0.06%
+## ADA pots data from DBSync
 
-Input into this in epoch 212 is:
+First 10 epochs in ada_pots:
 
 ```
-Calculating rewards: epoch=212 total_supply=31737719158318701 stake_rewards=31854784667376
+id  |  slot_no  | epoch_no |     treasury     |     reserves      |     rewards     |       utxo        | deposits_stake |     fees     | block_id | deposits_drep | deposits_proposal 
+-----+-----------+----------+------------------+-------------------+-----------------+-------------------+----------------+--------------+----------+---------------+-------------------
+   1 |   4924800 |      209 |    8332813711755 | 13286160713028443 | 593536826186446 | 31111517964861148 |   441012000000 |  10670212208 |  4512244 |             0 |                 0
+   2 |   5356800 |      210 |   16306644182013 | 13278197552770393 | 277915861250199 | 31427038405450971 |   533870000000 |   7666346424 |  4533814 |             0 |                 0
+   3 |   5788800 |      211 |   24275595982960 | 13270236767315870 | 164918966125973 | 31539966264042924 |   594636000000 |   7770532273 |  4555361 |             0 |                 0
+   4 |   6220800 |      212 |   32239292149804 | 13262280841681299 | 147882943225525 | 31556964153057144 |   626252000000 |   6517886228 |  4576676 |             0 |                 0
+   5 |   6652800 |      213 |   40198464232058 | 13247093198353459 | 133110645284460 | 31578940375911744 |   651738000000 |   5578218279 |  4597956 |             0 |                 0
+   6 |   7084800 |      214 |   48148335794725 | 13230232787944838 | 121337581585558 | 31599599756081623 |   674438000000 |   7100593256 |  4619398 |             0 |                 0
+   7 |   7516800 |      215 |   55876297807656 | 13212986170770203 | 117660526059600 | 31612774463528795 |   695040000000 |   7501833746 |  4640850 |             0 |                 0
+   8 |   7948807 |      216 |   63707722011028 | 13195031638588164 | 122159720478561 | 31618386634872973 |   706174000000 |   8110049274 |  4662422 |             0 |                 0
+   9 |   8380800 |      217 |   71629614335572 | 13176528835451373 | 127730158329564 | 31623386398075064 |   719058000000 |   5935808427 |  4683639 |             0 |                 0
+  10 |   8812800 |      218 |   79429791062499 | 13157936081322000 | 134680552513121 | 31627219255406326 |   729244000000 |   5075696054 |  4704367 |             0 |                 0
 ```

modules/accounts_state/README.md

@@ -0,0 +1,62 @@
+# AccountsState module
+
+This is the module which does the majority of the work in calculating monetary change
+(reserves, treasury) and rewards
+
+## Notes on verification
+
+The module has an inbuilt 'Verifier' which can compare against CSV files dumped from
+DBSync.
+
+### Pots verification
+
+Verifying the 'pots' values (reserves, treasury, deposits) is a good overall marker of
+successful calculation since everything (including rewards) feeds into it.
+
+To create a pots verification file, export the `ada_pots` table as CSV
+from Postgres on a DBSync database:
+
+```sql
+\COPY (
+  SELECT epoch_no AS epoch, reserves, treasury, deposits_stake AS deposits
+  FROM ada_pots
+  ORDER BY epoch_no
+) TO 'pots.mainnet.csv' WITH CSV HEADER
+```
+
+Then configure this as (e.g.)
+
+```toml
+[module.accounts-state]
+verify-pots-file = "../../modules/accounts_state/test-data/pots.mainnet.csv"
+```
+
+This is the default, since the pots file is small.  It will be updated periodically.
+
+### Rewards verification
+
+The verifier can also compare the rewards paid to members (delegators) and leader (pool)
+against a capture from the DBSync `rewards` table.  We name the files for the epoch *earned*,
+which is one less than when we calculate it.
+
+To create a rewards CSV file in Postgres on a DBSync database:
+
+```sql
+\COPY (
+  select encode(ph.hash_raw, 'hex') as spo, encode(a.hash_raw, 'hex') as address,
+         r.type, r.amount
+  from reward r
+  join pool_hash ph on r.pool_id = ph.id
+  join stake_address a on a.id = r.addr_id
+  where r.earned_epoch=211 and r.amount > 0
+) to 'rewards.mainnet.211.csv' with csv header
+```
+
+To configure verification, provide a path template which takes the epoch number:
+
+```toml
+[module.accounts-state]
+verify-rewards-files = "../../modules/accounts_state/test-data/rewards.mainnet.{}.csv"
+```
+
+The verifier will only verify epochs where this file exists.

modules/accounts_state/src/accounts_state.rs

@@ -25,9 +25,11 @@ use state::State;
 mod monetary;
 mod rewards;
 mod snapshot;
+mod verifier;
 use acropolis_common::queries::accounts::{
     AccountInfo, AccountsStateQuery, AccountsStateQueryResponse,
 };
+use verifier::Verifier;
 
 const DEFAULT_SPO_STATE_TOPIC: &str = "cardano.spo.state";
 const DEFAULT_EPOCH_ACTIVITY_TOPIC: &str = "cardano.epoch.activity";
@@ -64,6 +66,7 @@ impl AccountsState {
         mut stake_subscription: Box<dyn Subscription<Message>>,
         mut drep_state_subscription: Box<dyn Subscription<Message>>,
         mut parameters_subscription: Box<dyn Subscription<Message>>,
+        verifier: &Verifier,
     ) -> Result<()> {
         // Get the stake address deltas from the genesis bootstrap, which we know
         // don't contain any stake, plus an extra parameter state (!unexplained)
@@ -181,30 +184,30 @@ impl AccountsState {
                     _ => error!("Unexpected message type: {message:?}"),
                 }
 
-                // Handle epoch activity
-                let (_, message) = ea_message_f.await?;
+                // Update parameters, ready for monetary/rewards calc triggered by epoch_activity
+                let (_, message) = params_message_f.await?;
                 match message.as_ref() {
-                    Message::Cardano((block_info, CardanoMessage::EpochActivity(ea_msg))) => {
+                    Message::Cardano((block_info, CardanoMessage::ProtocolParams(params_msg))) => {
                         let span = info_span!(
-                            "account_state.handle_epoch_activity",
+                            "account_state.handle_parameters",
                             block = block_info.number
                         );
                         async {
                             Self::check_sync(&current_block, &block_info);
-                            let spo_rewards = state
-                                .handle_epoch_activity(ea_msg)
-                                .await
-                                .inspect_err(|e| error!("EpochActivity handling error: {e:#}"))
-                                .ok();
-                            // SPO rewards is for previous epoch
-                            if let Some(spo_rewards) = spo_rewards {
-                                if let Err(e) = spo_rewards_publisher
-                                    .publish_spo_rewards(block_info, spo_rewards)
-                                    .await
-                                {
-                                    error!("Error publishing SPO rewards: {e:#}")
+                            if let Some(ref block) = current_block {
+                                if block.number != block_info.number {
+                                    error!(
+                                        expected = block.number,
+                                        received = block_info.number,
+                                        "Certificate and parameters messages re-ordered!"
+                                    );
                                 }
                             }
+
+                            state
+                                .handle_parameters(params_msg)
+                                .inspect_err(|e| error!("Messaging handling error: {e}"))
+                                .ok();
                         }
                         .instrument(span)
                         .await;
@@ -213,31 +216,30 @@ impl AccountsState {
                     _ => error!("Unexpected message type: {message:?}"),
                 }
 
-                // Update parameters - *after* reward calculation in epoch-activity above
-                // ready for the *next* epoch boundary
-                let (_, message) = params_message_f.await?;
+                // Handle epoch activity
+                let (_, message) = ea_message_f.await?;
                 match message.as_ref() {
-                    Message::Cardano((block_info, CardanoMessage::ProtocolParams(params_msg))) => {
+                    Message::Cardano((block_info, CardanoMessage::EpochActivity(ea_msg))) => {
                         let span = info_span!(
-                            "account_state.handle_parameters",
+                            "account_state.handle_epoch_activity",
                             block = block_info.number
                         );
                         async {
                             Self::check_sync(&current_block, &block_info);
-                            if let Some(ref block) = current_block {
-                                if block.number != block_info.number {
-                                    error!(
-                                        expected = block.number,
-                                        received = block_info.number,
-                                        "Certificate and parameters messages re-ordered!"
-                                    );
+                            let spo_rewards = state
+                                .handle_epoch_activity(ea_msg, &verifier)
+                                .await
+                                .inspect_err(|e| error!("EpochActivity handling error: {e:#}"))
+                                .ok();
+                            // SPO rewards is for previous epoch
+                            if let Some(spo_rewards) = spo_rewards {
+                                if let Err(e) = spo_rewards_publisher
+                                    .publish_spo_rewards(block_info, spo_rewards)
+                                    .await
+                                {
+                                    error!("Error publishing SPO rewards: {e:#}")
                                 }
                             }
-
-                            state
-                                .handle_parameters(params_msg)
-                                .inspect_err(|e| error!("Messaging handling error: {e}"))
-                                .ok();
                         }
                         .instrument(span)
                         .await;
@@ -388,6 +390,19 @@ impl AccountsState {
             .unwrap_or(DEFAULT_ACCOUNTS_QUERY_TOPIC.1.to_string());
         info!("Creating query handler on '{}'", accounts_query_topic);
 
+        // Create verifier and read comparison data according to config
+        let mut verifier = Verifier::new();
+
+        if let Ok(verify_pots_file) = config.get_string("verify-pots-file") {
+            info!("Verifying pots against '{verify_pots_file}'");
+            verifier.read_pots(&verify_pots_file);
+        }
+
+        if let Ok(verify_rewards_files) = config.get_string("verify-rewards-files") {
+            info!("Verifying rewards against '{verify_rewards_files}'");
+            verifier.read_rewards(&verify_rewards_files);
+        }
+
         // Create history
         let history = Arc::new(Mutex::new(StateHistory::<State>::new(
             "AccountsState",
@@ -529,6 +544,7 @@ impl AccountsState {
                 stake_subscription,
                 drep_state_subscription,
                 parameters_subscription,
+                &verifier,
             )
             .await
             .unwrap_or_else(|e| error!("Failed: {e}"));