db-sync: Add validation for rewards and withdrawals

For every address that has ever seen a withdrawal:

    sum rewards >= sum withdrawals

cardano-db-sync/cardano-db-sync.cabal

@@ -72,6 +72,9 @@ library
                         Cardano.DbSync.Era.Shelley.Query
                         Cardano.DbSync.Era.Shelley.Validate
 
+                        -- Temporary debugging validation
+                        Cardano.DbSync.Era.Shelley.ValidateWithdrawal
+
                         Cardano.DbSync.Era.Util
 
                         Cardano.DbSync.Metrics

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

@@ -13,6 +13,7 @@ import           Cardano.BM.Trace (Trace, logInfo, logWarning)
 
 import           Cardano.Db (DbLovelace, RewardSource)
 import qualified Cardano.Db as Db
+import           Cardano.DbSync.Era.Shelley.ValidateWithdrawal (validateRewardWithdrawals)
 
 import           Cardano.Ledger.BaseTypes (Network)
 import           Cardano.Ledger.Coin (Coin (..))
@@ -55,6 +56,7 @@ validateEpochRewards tracer nw currentEpoch rmap = do
                       [ "validateEpochRewards: total rewards that become spendable in epoch "
                       , textShow (unEpochNo currentEpoch), " is ", textShow actual, " ADA"
                       ]
+    validateRewardWithdrawals currentEpoch
   where
     expected :: Db.Ada
     expected = Db.word64ToAda . fromIntegral . sum $ map unCoin (Map.elems rmap)
@@ -114,6 +116,7 @@ diffRewardMap
     -> ReaderT SqlBackend m ()
 diffRewardMap epochNo dbMap ledgerMap = do
     liftIO $ do
+      putStrLn $ "Epoch No: " ++ show (unEpochNo epochNo)
       putStrLn $ "dbMap length: " ++ show (Map.size dbMap)
       putStrLn $ "ledgerMap length: " ++ show (Map.size ledgerMap)
       putStrLn $ "diffMap length: " ++ show (Map.size diffMap)

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

@@ -0,0 +1,121 @@
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE OverloadedStrings #-}
+module Cardano.DbSync.Era.Shelley.ValidateWithdrawal
+  ( validateRewardWithdrawals
+  ) where
+
+import           Control.Monad.IO.Class (MonadIO, liftIO)
+import           Control.Monad.Trans.Control (MonadBaseControl)
+import           Control.Monad.Trans.Reader (ReaderT)
+
+import           Cardano.Db (Ada (..))
+import qualified Cardano.Db as Db
+import           Cardano.Slotting.Slot (EpochNo (..))
+import           Cardano.Sync.Util
+
+import           Data.Either (partitionEithers)
+import           Data.Fixed (Micro)
+import qualified Data.List as List
+import           Data.Text (Text)
+import qualified Data.Text as Text
+
+import           Database.Esqueleto.Legacy (InnerJoin (..), Value (..), asc, distinct, from,
+                   groupBy, having, on, orderBy, select, sum_, unValue, val, where_, (<.), (==.),
+                   (^.))
+
+import           Database.Persist.Sql (SqlBackend)
+
+
+-- For any stake address which has seen a withdrawal, the sum of the withdrawals for that address
+-- should be less than or equal to the sum of the rewards for that address.
+
+validateRewardWithdrawals
+    :: (MonadBaseControl IO m, MonadIO m)
+    => EpochNo -> ReaderT SqlBackend m ()
+validateRewardWithdrawals (EpochNo epochNo) = do
+  res <- mapM validateAccounting =<< queryWithdrawalAddresses
+  _bad <- queryBadWithdrawals
+  liftIO $
+    case partitionEithers res of
+      ([], _) -> pure ()
+      (xs, _) -> do
+        putStr $ show (length xs) ++ " errors, eg\n" ++ unlines (map reportError xs)
+        panicAbort $ "validateRewardWithdrawals: " <> textShow epochNo
+
+-- -----------------------------------------------------------------------------
+
+data AddressInfo = AddressInfo
+  { aiStakeAddress :: !Text
+  , aiSumRewards :: !Ada
+  , aiSumWithdrawals :: !Ada
+  } deriving (Eq, Ord, Show)
+
+reportError :: AddressInfo -> String
+reportError ai =
+  mconcat
+    [ "  ", Text.unpack (aiStakeAddress ai), " rewards are ", show (aiSumRewards ai)
+    , " ADA and withdrawals are ", show (aiSumWithdrawals ai), " ADA\n"
+    ]
+
+-- For a given TxId, validate the input/output accounting.
+validateAccounting
+    :: (MonadBaseControl IO m, MonadIO m)
+    => Db.StakeAddressId -> ReaderT SqlBackend m  (Either AddressInfo ())
+validateAccounting addrId = do
+   ai <- queryAddressInfo addrId
+   pure $ if aiSumRewards ai < aiSumWithdrawals ai
+            then Left ai
+            else Right ()
+
+-- -------------------------------------------------------------------------------------------------
+
+-- Get all stake addresses with have seen a withdrawal, and return them in shuffled order.
+queryWithdrawalAddresses :: MonadIO m => ReaderT SqlBackend m [Db.StakeAddressId]
+queryWithdrawalAddresses = do
+  res <- select . distinct . from $ \ wd -> do
+            orderBy [asc (wd ^. Db.WithdrawalAddrId)]
+            pure (wd ^. Db.WithdrawalAddrId)
+  pure $ map unValue res
+
+queryAddressInfo :: MonadIO m => Db.StakeAddressId -> ReaderT SqlBackend m AddressInfo
+queryAddressInfo addrId = do
+    rwds <- select . from $ \ rwd -> do
+              where_ (rwd ^. Db.RewardAddrId ==. val addrId)
+              pure (sum_ $ rwd ^. Db.RewardAmount)
+    wdls <- select . from $ \ wdl -> do
+              where_ (wdl ^. Db.WithdrawalAddrId ==. val addrId)
+              pure (sum_ (wdl ^. Db.WithdrawalAmount))
+    view <- select . from $ \ saddr -> do
+              where_ (saddr ^. Db.StakeAddressId ==. val addrId)
+              pure (saddr ^. Db.StakeAddressView)
+    pure $ convert (Db.listToMaybe rwds) (Db.listToMaybe wdls) (Db.listToMaybe view)
+  where
+    convert :: Maybe (Value (Maybe Micro)) -> Maybe (Value (Maybe Micro)) -> Maybe (Value Text) -> AddressInfo
+    convert rAmount wAmount mview =
+        AddressInfo
+          { aiStakeAddress  = maybe "unknown" unValue mview
+          , aiSumRewards = Db.unValueSumAda rAmount
+          , aiSumWithdrawals = Db.unValueSumAda wAmount
+          }
+
+-- A stake address state is bad if sum rewards < sum withdrawals
+queryBadWithdrawals :: MonadIO m => ReaderT SqlBackend m [AddressInfo]
+queryBadWithdrawals = do
+    res <- select . from $ \ (rwd `InnerJoin` wdrl `InnerJoin` sa) -> do
+              on (rwd ^. Db.RewardAddrId ==. sa ^. Db.StakeAddressId)
+              on (rwd ^. Db.RewardAddrId ==. wdrl ^. Db.WithdrawalAddrId)
+              groupBy (sa ^. Db.StakeAddressId)
+              let sumReward = sum_ (rwd ^. Db.RewardAmount)
+                  sumWithdraw = sum_ (wdrl ^. Db.WithdrawalAmount)
+              having (sumReward <. sumWithdraw)
+              pure (sa ^. Db.StakeAddressView, sumReward, sumWithdraw)
+    pure $ List.sort (map convert res)
+  where
+    convert :: (Value Text, Value (Maybe Micro), Value (Maybe Micro)) -> AddressInfo
+    convert (Value saView, rwdTotal, wdrlTotal) =
+      AddressInfo
+        { aiStakeAddress  = saView
+        , aiSumRewards  = Db.unValueSumAda (Just rwdTotal)
+        , aiSumWithdrawals = Db.unValueSumAda (Just wdrlTotal)
+        }
+