Fix epoch stake

cardano-db-sync/src/Cardano/DbSync/Default.hs

@@ -157,7 +157,7 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
   let !details = apSlotDetails applyResult
   let !withinTwoMin = isWithinTwoMin details
   let !withinHalfHour = isWithinHalfHour details
-  insertNewEpochLedgerEvents syncEnv (sdEpochNo details) (apEvents applyResult)
+  insertNewEpochLedgerEvents syncEnv applyResult (sdEpochNo details) (apEvents applyResult)
 
   let isNewEpochEvent = hasNewEpochEvent (apEvents applyResult)
   let isStartEventOrRollback = hasEpochStartEvent (apEvents applyResult) || firstAfterRollback

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Generic/StakeDist.hs

@@ -12,6 +12,8 @@ module Cardano.DbSync.Era.Shelley.Generic.StakeDist (
   StakeSlice (..),
   getSecurityParameter,
   getStakeSlice,
+  countEpochStake,
+  fullEpochStake,
   getPoolDistr,
 ) where
 
@@ -175,6 +177,110 @@ genericStakeSlice pInfo epochBlockNo lstate isMigration
             VMap.mapMaybe id $
               VMap.mapWithKey (\a p -> (,p) <$> lookupStake a) delegationsSliced
 
+countEpochStake ::
+  ExtLedgerState CardanoBlock mk ->
+  Maybe (Word64, EpochNo)
+countEpochStake els =
+  case ledgerState els of
+    LedgerStateByron _ -> Nothing
+    LedgerStateShelley sls -> genericCountEpochStake sls
+    LedgerStateAllegra als -> genericCountEpochStake als
+    LedgerStateMary mls -> genericCountEpochStake mls
+    LedgerStateAlonzo als -> genericCountEpochStake als
+    LedgerStateBabbage bls -> genericCountEpochStake bls
+    LedgerStateConway cls -> genericCountEpochStake cls
+    LedgerStateDijkstra dls -> genericCountEpochStake dls
+
+genericCountEpochStake ::
+  LedgerState (ShelleyBlock p era) mk ->
+  Maybe (Word64, EpochNo)
+genericCountEpochStake lstate =
+  Just (delegationsLen, epoch)
+  where
+    epoch :: EpochNo
+    epoch = EpochNo $ 1 + unEpochNo (Shelley.nesEL (Consensus.shelleyLedgerState lstate))
+
+    stakeSnapshot :: Ledger.SnapShot
+    stakeSnapshot =
+      Ledger.ssStakeMark . Shelley.esSnapshots . Shelley.nesEs $
+        Consensus.shelleyLedgerState lstate
+
+    delegations :: VMap VB VB (Credential 'Staking) (KeyHash 'StakePool)
+    delegations = Ledger.ssDelegations stakeSnapshot
+
+    delegationsLen :: Word64
+    delegationsLen = fromIntegral $ VMap.size $ VMap.filter (\k _ -> hasStake k) delegations
+
+    stakes :: VMap VB VP (Credential 'Staking) (Ledger.CompactForm Coin)
+    stakes = Ledger.unStake $ Ledger.ssStake stakeSnapshot
+
+    hasStake :: Credential 'Staking -> Bool
+    hasStake cred = isJust (VMap.lookup cred stakes)
+
+fullEpochStake ::
+  ExtLedgerState CardanoBlock mk ->
+  StakeSliceRes
+fullEpochStake els =
+  case ledgerState els of
+    LedgerStateByron _ -> NoSlices
+    LedgerStateShelley sls -> genericFullStakeSlice sls
+    LedgerStateAllegra als -> genericFullStakeSlice als
+    LedgerStateMary mls -> genericFullStakeSlice mls
+    LedgerStateAlonzo als -> genericFullStakeSlice als
+    LedgerStateBabbage bls -> genericFullStakeSlice bls
+    LedgerStateConway cls -> genericFullStakeSlice cls
+    LedgerStateDijkstra dls -> genericFullStakeSlice dls
+
+genericFullStakeSlice ::
+  forall era p mk.
+  LedgerState (ShelleyBlock p era) mk ->
+  StakeSliceRes
+genericFullStakeSlice lstate =
+  Slice stakeSlice True
+  where
+    epoch :: EpochNo
+    epoch = EpochNo $ 1 + unEpochNo (Shelley.nesEL (Consensus.shelleyLedgerState lstate))
+
+    -- We use 'ssStakeMark' here. That means that when these values
+    -- are added to the database, the epoch number where they become active is the current
+    -- epoch plus one.
+    stakeSnapshot :: Ledger.SnapShot
+    stakeSnapshot =
+      Ledger.ssStakeMark . Shelley.esSnapshots . Shelley.nesEs $
+        Consensus.shelleyLedgerState lstate
+
+    delegations :: VMap.KVVector VB VB (Credential 'Staking, KeyHash 'StakePool)
+    delegations = VMap.unVMap $ Ledger.ssDelegations stakeSnapshot
+
+    delegationsLen :: Word64
+    delegationsLen = fromIntegral $ VG.length delegations
+
+    stakes :: VMap VB VP (Credential 'Staking) (Ledger.CompactForm Coin)
+    stakes = Ledger.unStake $ Ledger.ssStake stakeSnapshot
+
+    lookupStake :: Credential 'Staking -> Maybe Coin
+    lookupStake cred = Ledger.fromCompact <$> VMap.lookup cred stakes
+
+    -- The starting index of the data in the delegation vector.
+    index :: Word64
+    index = 0
+
+    stakeSlice :: StakeSlice
+    stakeSlice =
+      StakeSlice
+        { sliceEpochNo = epoch
+        , sliceDistr = distribution
+        }
+      where
+        delegationsSliced :: VMap VB VB (Credential 'Staking) (KeyHash 'StakePool)
+        delegationsSliced = VMap $ VG.slice (fromIntegral index) (fromIntegral delegationsLen) delegations
+
+        distribution :: Map StakeCred (Coin, PoolKeyHash)
+        distribution =
+          VMap.toMap $
+            VMap.mapMaybe id $
+              VMap.mapWithKey (\a p -> (,p) <$> lookupStake a) delegationsSliced
+
 getPoolDistr ::
   ExtLedgerState CardanoBlock mk ->
   Maybe (Map PoolKeyHash (Coin, Word64), Map PoolKeyHash Natural)

cardano-db-sync/src/Cardano/DbSync/Era/Universal/Insert/LedgerEvent.hs

@@ -25,11 +25,12 @@ import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
 import Cardano.DbSync.Era.Universal.Adjust (adjustEpochRewards)
 import Cardano.DbSync.Era.Universal.Epoch (insertPoolDepositRefunds, insertProposalRefunds, insertRewardRests, insertRewards)
 import Cardano.DbSync.Era.Universal.Insert.GovAction
-import Cardano.DbSync.Era.Universal.Validate (validateEpochRewards)
+import Cardano.DbSync.Era.Universal.Validate (validateEpochRewards, validateEpochStake)
 import Cardano.DbSync.Ledger.Event
 import Cardano.DbSync.Types
 
 import Cardano.DbSync.Error (SyncNodeError)
+import Cardano.DbSync.Ledger.Types
 import Cardano.DbSync.Metrics (setDbEpochSyncDuration, setDbEpochSyncNumber)
 import Control.Concurrent.Class.MonadSTM.Strict (readTVarIO, writeTVar)
 import Control.Monad.Extra (whenJust)
@@ -44,10 +45,11 @@ import Text.Printf (printf)
 --------------------------------------------------------------------------------------------
 insertNewEpochLedgerEvents ::
   SyncEnv ->
+  ApplyResult ->
   EpochNo ->
   [LedgerEvent] ->
   ExceptT SyncNodeError DB.DbM ()
-insertNewEpochLedgerEvents syncEnv currentEpochNo@(EpochNo curEpoch) =
+insertNewEpochLedgerEvents syncEnv applyRes currentEpochNo@(EpochNo curEpoch) =
   mapM_ handler
   where
     metricSetters = envMetricSetters syncEnv
@@ -72,6 +74,7 @@ insertNewEpochLedgerEvents syncEnv currentEpochNo@(EpochNo curEpoch) =
     handler ev =
       case ev of
         LedgerNewEpoch en ss -> do
+          validateEpochStake syncEnv applyRes True
           databaseCacheSize <- lift DB.queryStatementCacheSize
           liftIO . logInfo tracer $ "Database Statement Cache size is " <> textShow databaseCacheSize
           currentTime <- liftIO getCurrentTime

cardano-db-sync/src/Cardano/DbSync/Era/Universal/Validate.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Era.Universal.Validate (
+  validateEpochStake,
   validateEpochRewards,
 ) where
 
@@ -20,10 +21,53 @@ import qualified Data.Set as Set
 import GHC.Err (error)
 
 import qualified Cardano.Db as DB
+import Cardano.DbSync.Api
+import Cardano.DbSync.Api.Types
 import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
+import Cardano.DbSync.Era.Universal.Epoch
 import Cardano.DbSync.Error (SyncNodeError)
 import Cardano.DbSync.Ledger.Event
+import Cardano.DbSync.Ledger.Types
 import Cardano.DbSync.Types
+import Cardano.DbSync.Util.Constraint
+import qualified Data.Strict.Maybe as Strict
+
+validateEpochStake ::
+  SyncEnv ->
+  ApplyResult ->
+  Bool ->
+  ExceptT SyncNodeError DB.DbM ()
+validateEpochStake syncEnv applyRes firstCall = case apOldLedger applyRes of
+  Strict.Just lstate | Just (expectedCount, epoch) <- Generic.countEpochStake (clsState lstate) -> do
+    actualCount <- lift $ DB.queryNormalEpochStakeCount (unEpochNo epoch)
+    if actualCount /= expectedCount
+      then do
+        liftIO
+          . logWarning tracer
+          $ mconcat
+            [ "validateEpochStake: epoch stake in epoch "
+            , textShow (unEpochNo epoch)
+            , " expected total of "
+            , textShow expectedCount
+            , " but got "
+            , textShow actualCount
+            ]
+        let slice = Generic.fullEpochStake (clsState lstate)
+        addStakeConstraintsIfNotExist syncEnv tracer
+        insertStakeSlice syncEnv slice
+        when firstCall $ validateEpochStake syncEnv applyRes False
+      else
+        liftIO $
+          logInfo tracer $
+            mconcat
+              [ "Validate Epoch Stake: total entries in epoch "
+              , textShow (unEpochNo epoch)
+              , " are "
+              , textShow actualCount
+              ]
+  _ -> pure ()
+  where
+    tracer = getTrace syncEnv
 
 validateEpochRewards ::
   Trace IO Text ->

cardano-db/src/Cardano/Db/Statement/StakeDelegation.hs

@@ -216,6 +216,29 @@ queryNormalEpochRewardCount epochNum =
   runSession mkDbCallStack $
     HsqlSes.statement epochNum queryNormalEpochRewardCountStmt
 
+-- | QUERY ---------------------------------------------------------------------
+queryNormalEpochStakeCountStmt :: HsqlStmt.Statement Word64 Word64
+queryNormalEpochStakeCountStmt =
+  HsqlStmt.Statement sql encoder decoder True
+  where
+    sql =
+      TextEnc.encodeUtf8 $
+        Text.concat
+          [ "SELECT COUNT(*)::bigint"
+          , " FROM epoch_stake"
+          , " WHERE epoch_no = $1"
+          ]
+
+    encoder = HsqlE.param (HsqlE.nonNullable $ fromIntegral >$< HsqlE.int8)
+    decoder =
+      HsqlD.singleRow $
+        fromIntegral <$> HsqlD.column (HsqlD.nonNullable HsqlD.int8)
+
+queryNormalEpochStakeCount :: Word64 -> DbM Word64
+queryNormalEpochStakeCount epochNum =
+  runSession mkDbCallStack $
+    HsqlSes.statement epochNum queryNormalEpochStakeCountStmt
+
 --------------------------------------------------------------------------------
 queryRewardCount :: DbM Word64
 queryRewardCount =