Populate the rewards table

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

@@ -97,6 +97,7 @@ library
                       , monad-logger
                       , network-mux
                       , optparse-applicative
+                      , pretty-show
                       , network
                       , ouroboros-consensus
                       , ouroboros-consensus-byron

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

@@ -43,8 +43,7 @@ import qualified Data.ByteString.Char8 as BS
 import qualified Data.Text as Text
 import qualified Data.Text.Encoding as Text
 
-import           Database.Persist.Sql (IsolationLevel (Serializable), SqlBackend,
-                    transactionSaveWithIsolation)
+import           Database.Persist.Sql (SqlBackend)
 
 import qualified Cardano.Db as DB
 import qualified Cardano.DbSync.Era.Byron.Util as Byron
@@ -70,7 +69,7 @@ insertByronBlock tracer blk details = do
   -- Serializiing things during syncing can drastically slow down full sync
   -- times (ie 10x or more).
   when (getSyncStatus details == SyncFollowing) $
-    transactionSaveWithIsolation Serializable
+    DB.transactionCommit
   pure res
 
 

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

@@ -49,8 +49,7 @@ import qualified Data.Text as Text
 import qualified Data.Text.Encoding as Text
 import qualified Data.Text.Encoding.Error as Text
 
-import           Database.Persist.Sql (IsolationLevel (Serializable), SqlBackend,
-                    transactionSaveWithIsolation)
+import           Database.Persist.Sql (SqlBackend)
 
 import           Ouroboros.Consensus.HardFork.Combinator.Basics (LedgerState (..))
 import           Ouroboros.Consensus.Shelley.Protocol (StandardShelley)
@@ -59,17 +58,20 @@ import qualified Shelley.Spec.Ledger.Address as Shelley
 import           Shelley.Spec.Ledger.BaseTypes (StrictMaybe, strictMaybeToMaybe)
 import qualified Shelley.Spec.Ledger.BaseTypes as Shelley
 import qualified Shelley.Spec.Ledger.Coin as Shelley
+import qualified Shelley.Spec.Ledger.Credential as Shelley
 import qualified Shelley.Spec.Ledger.Keys as Shelley
+import qualified Shelley.Spec.Ledger.LedgerState as Shelley
 import qualified Shelley.Spec.Ledger.MetaData as Shelley
 import qualified Shelley.Spec.Ledger.PParams as Shelley
 import qualified Shelley.Spec.Ledger.Tx as Shelley
 import qualified Shelley.Spec.Ledger.TxBody as Shelley
 
 
 insertShelleyBlock
-    :: Trace IO Text -> DbSyncEnv -> ShelleyBlock -> LedgerState CardanoBlock -> SlotDetails
+    :: Trace IO Text -> DbSyncEnv -> ShelleyBlock -> LedgerState ShelleyBlock
+    -> Maybe (Shelley.RewardUpdate StandardShelley) -> SlotDetails
     -> ReaderT SqlBackend (LoggingT IO) (Either DbSyncNodeError ())
-insertShelleyBlock tracer env blk _ledger details = do
+insertShelleyBlock tracer env blk _ledgerState mRewards details = do
   runExceptT $ do
     pbid <- liftLookupFail "insertShelleyBlock" $ DB.queryBlockId (Shelley.blockPrevHash blk)
     mPhid <- lift $ queryPoolHashId (Shelley.blockVrfKeyToPoolHash blk)
@@ -116,10 +118,17 @@ insertShelleyBlock tracer env blk _ledger details = do
         , ", hash ", renderByteArray (Shelley.blockHash blk)
         ]
 
+    case mRewards of
+      Nothing -> pure ()
+      Just rewards -> do
+        -- The slot
+        let rEpoch = unEpochNo (sdEpochNo details) - 2
+        insertRewards tracer env blkId rEpoch rewards
+
     when (getSyncStatus details == SyncFollowing) $
       -- Serializiing things during syncing can drastically slow down full sync
       -- times (ie 10x or more).
-      lift $ transactionSaveWithIsolation Serializable
+      lift DB.transactionCommit
   where
     logger :: Bool -> Trace IO a -> a -> IO ()
     logger followingClosely
@@ -248,7 +257,7 @@ insertPoolRegister tracer txId idx params = do
         , " > maxLovelace. See https://github.com/input-output-hk/cardano-ledger-specs/issues/1551"
         ]
 
-  poolHashId <- lift . DB.insertPoolHash $ DB.PoolHash (Shelley.unKeyHashBS $ Shelley._poolPubKey params)
+  poolHashId <- lift . DB.insertPoolHash $ DB.PoolHash (Shelley.unKeyHash $ Shelley._poolPubKey params)
   poolUpdateId <- lift . DB.insertPoolUpdate $
                     DB.PoolUpdate
                       { DB.poolUpdateHashId = poolHashId
@@ -314,7 +323,7 @@ insertPoolOwner
 insertPoolOwner poolHashId txId skh =
   void . lift . DB.insertPoolOwner $
     DB.PoolOwner
-      { DB.poolOwnerHash = Shelley.unKeyHashBS skh
+      { DB.poolOwnerHash = Shelley.unKeyHash skh
       , DB.poolOwnerPoolHashId = poolHashId
       , DB.poolOwnerRegisteredTxId = txId
       }
@@ -470,7 +479,7 @@ insertParamUpdate _tracer txId (Shelley.Update (Shelley.ProposedPPUpdates umap)
       void . lift . DB.insertParamUpdate $
         DB.ParamUpdate
           { DB.paramUpdateEpochNo = epoch
-          , DB.paramUpdateKey = Shelley.unKeyHashBS key
+          , DB.paramUpdateKey = Shelley.unKeyHash key
           , DB.paramUpdateMinFeeA = fromIntegral <$> strictMaybeToMaybe (Shelley._minfeeA pmap)
           , DB.paramUpdateMinFeeB = fromIntegral <$> strictMaybeToMaybe (Shelley._minfeeB pmap)
           , DB.paramUpdateMaxBlockSize = fromIntegral <$> strictMaybeToMaybe (Shelley._maxBBSize pmap)
@@ -531,3 +540,26 @@ safeDecodeUtf8 bs
 
 containsUnicodeNul :: Text -> Bool
 containsUnicodeNul = Text.isInfixOf "\\u000"
+
+insertRewards
+    :: (MonadBaseControl IO m, MonadIO m)
+    => Trace IO Text -> DbSyncEnv -> DB.BlockId -> Word64 -> Shelley.RewardUpdate StandardShelley
+    -> ExceptT DbSyncNodeError (ReaderT SqlBackend m) ()
+insertRewards _tracer env blkId epoch rewards =
+    mapM_ insertOneReward $ Map.toList (Shelley.rs rewards)
+  where
+    insertOneReward
+        :: (MonadBaseControl IO m, MonadIO m)
+        => (Shelley.Credential 'Shelley.Staking StandardShelley, Shelley.Coin)
+        -> ExceptT DbSyncNodeError (ReaderT SqlBackend m) ()
+    insertOneReward (saddr, coin) = do
+      saId <- firstExceptT (NELookup "insertReward")
+                . newExceptT
+                $ queryStakeAddress (Shelley.stakingCredHash env saddr)
+      void . lift . DB.insertReward $
+        DB.Reward
+          { DB.rewardAddrId = saId
+          , DB.rewardAmount = fromIntegral (Shelley.unCoin coin)
+          , DB.rewardEpochNo = epoch
+          , DB.rewardBlockId = blkId
+          }

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

@@ -14,7 +14,8 @@ module Cardano.DbSync.Era.Shelley.Query
 import qualified Cardano.Crypto.Hash as Crypto
 import           Cardano.Db
 import           Cardano.DbSync.Types
-import           Cardano.DbSync.Era.Shelley.Util (unKeyHashBS)
+import           Cardano.DbSync.Era.Shelley.Util (unKeyHash)
+import           Cardano.DbSync.Util
 
 import           Control.Monad.IO.Class (MonadIO)
 import           Control.Monad.Trans.Reader (ReaderT)
@@ -42,15 +43,15 @@ queryStakeAddress addr = do
   res <- select . from $ \ saddr -> do
             where_ (saddr ^. StakeAddressHashRaw ==. val addr)
             pure (saddr ^. StakeAddressId)
-  pure $ maybeToEither (DbLookupMessage "StakeAddress") unValue (listToMaybe res)
+  pure $ maybeToEither (DbLookupMessage $ "StakeAddress " <> renderByteArray addr) unValue (listToMaybe res)
 
 queryStakePoolKeyHash :: MonadIO m => ShelleyStakePoolKeyHash -> ReaderT SqlBackend m (Either LookupFail PoolHashId)
 queryStakePoolKeyHash kh = do
   res <- select . from $ \ (poolUpdate `InnerJoin` poolHash `InnerJoin` tx `InnerJoin` blk) -> do
             on (blk ^. BlockId ==. tx ^. TxBlock)
             on (tx ^. TxId ==. poolUpdate ^. PoolUpdateRegisteredTxId)
             on (poolUpdate ^. PoolUpdateHashId ==. poolHash ^. PoolHashId)
-            where_ (poolHash ^. PoolHashHash ==. val (unKeyHashBS kh))
+            where_ (poolHash ^. PoolHashHash ==. val (unKeyHash kh))
             orderBy [desc (blk ^. BlockSlotNo)]
             pure (poolHash ^. PoolHashId)
   pure $ maybeToEither (DbLookupMessage "StakePoolKeyHash") unValue (listToMaybe res)

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
@@ -38,7 +39,7 @@ module Cardano.DbSync.Era.Shelley.Util
   , txWithdrawalSum
   , unHeaderHash
   , unitIntervalToDouble
-  , unKeyHashBS
+  , unKeyHash
   , unTxHash
   ) where
 
@@ -55,9 +56,6 @@ import qualified Cardano.Db as Db
 import           Cardano.DbSync.Config
 import           Cardano.DbSync.Types
 
-import qualified Cardano.Ledger.Crypto as Shelley
-import qualified Cardano.Ledger.Era as ShelleyEra
-
 import           Cardano.Slotting.Slot (SlotNo (..))
 
 import qualified Data.Binary.Put as Binary
@@ -70,7 +68,7 @@ import qualified Data.Text as Text
 import qualified Data.Text.Encoding as Text
 
 import qualified Ouroboros.Consensus.Shelley.Ledger.Block as Consensus
-import           Ouroboros.Consensus.Shelley.Protocol (StandardShelley, StandardShelley)
+import           Ouroboros.Consensus.Shelley.Protocol (StandardShelley)
 import           Ouroboros.Network.Block (BlockNo (..))
 
 import qualified Shelley.Spec.Ledger.Address as Shelley
@@ -157,7 +155,6 @@ mkSlotLeader blk mPoolId =
                 Just _ -> "Pool-" <> short
   in Db.SlotLeader slHash mPoolId slName
 
-
 nonceToBytes :: Shelley.Nonce -> ByteString
 nonceToBytes nonce =
   case nonce of
@@ -215,7 +212,7 @@ txOutputList tx =
 
 txOutputSum :: ShelleyTx -> Word64
 txOutputSum tx =
-    foldl' (+) 0 $ map outValue (Shelley._outputs $ Shelley._body tx)
+    sum $ map outValue (Shelley._outputs $ Shelley._body tx)
   where
     outValue :: ShelleyTxOut -> Word64
     outValue (Shelley.TxOut _ coin) = fromIntegral $ unCoin coin
@@ -234,15 +231,8 @@ unHeaderHash = Crypto.hashToBytes . Shelley.unHashHeader . Consensus.unShelleyHa
 unitIntervalToDouble :: Shelley.UnitInterval -> Double
 unitIntervalToDouble = fromRational . Shelley.unitIntervalToRational
 
-unKeyHash :: Shelley.KeyHash disc era
-                   -> Crypto.Hash
-                        (Shelley.ADDRHASH (ShelleyEra.Crypto era))
-                        (DSIGN.VerKeyDSIGN (Shelley.DSIGN (ShelleyEra.Crypto era)))
-
-unKeyHash (Shelley.KeyHash x) = x
-
-unKeyHashBS :: Shelley.KeyHash d crypto -> ByteString
-unKeyHashBS = Crypto.hashToBytes . unKeyHash
+unKeyHash :: Shelley.KeyHash d era -> ByteString
+unKeyHash (Shelley.KeyHash kh) = Crypto.hashToBytes kh
 
 unTxHash :: ShelleyTxId -> ByteString
 unTxHash (Shelley.TxId txid) = Crypto.hashToBytes txid

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

@@ -1,4 +1,5 @@
 {-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -39,12 +40,18 @@ import qualified Data.List as List
 import           Ouroboros.Consensus.Block (CodecConfig, blockSlot, blockPrevHash)
 import           Ouroboros.Consensus.Byron.Ledger (initByronLedgerState)
 import           Ouroboros.Consensus.Cardano.CanHardFork ()
+import           Ouroboros.Consensus.Cardano.Block (LedgerState (LedgerStateByron, LedgerStateShelley))
 import           Ouroboros.Consensus.Config (TopLevelConfig (..))
-import           Ouroboros.Consensus.Ledger.Abstract (LedgerConfig, ledgerTipHash, ledgerTipSlot, tickThenReapply)
+import           Ouroboros.Consensus.Ledger.Abstract (LedgerConfig, ledgerTipHash, ledgerTipSlot, tickThenApply, tickThenReapply)
 import           Ouroboros.Consensus.HardFork.Combinator.Basics (LedgerState (..))
 import           Ouroboros.Consensus.HardFork.Combinator.State.Infra (initHardForkState)
+import qualified Ouroboros.Consensus.Shelley.Ledger.Ledger as Consensus
+import           Ouroboros.Consensus.Shelley.Protocol (StandardShelley)
 import           Ouroboros.Consensus.Storage.Serialisation (DecodeDisk (..), EncodeDisk (..))
 
+import qualified Shelley.Spec.Ledger.BaseTypes as Shelley
+import qualified Shelley.Spec.Ledger.LedgerState as Shelley
+
 import           System.Directory (listDirectory, removeFile)
 import           System.FilePath ((</>), dropExtension, takeExtension)
 
@@ -79,30 +86,42 @@ initLedgerStateVar genesisConfig = do
 -- The function 'tickThenReapply' does zero validation, so add minimal validation ('blockPrevHash'
 -- matches the tip hash of the 'LedgerState'). This was originally for debugging but the check is
 -- cheap enough to keep.
-applyBlock :: LedgerStateVar -> CardanoBlock -> IO CardanoLedgerState
+applyBlock :: LedgerStateVar -> CardanoBlock -> IO (CardanoLedgerState, Maybe (Shelley.RewardUpdate StandardShelley))
 applyBlock (LedgerStateVar stateVar) blk =
-  -- 'LedgerStateVar' is just being used as a mutable variable. There should not ever
-  -- be any contention on this variable, so putting everything inside 'atomically'
-  -- is fine.
-  atomically $ do
-    oldState <- readTVar stateVar
-    if ledgerTipHash (clsState oldState) == blockPrevHash blk
-      then do
-        let !newState = oldState { clsState = tickThenReapply (clsConfig oldState) blk (clsState oldState) }
-        writeTVar stateVar newState
-        pure newState
-      else panic $ mconcat
-             [ "applyBlock: Hash mismatch when applying block with slot no ", textShow (blockSlot blk), "\n"
-             , "applyBlock: ", textShow (ledgerTipHash $ clsState oldState), " / = ", textShow (blockPrevHash blk)
-             ]
+    -- 'LedgerStateVar' is just being used as a mutable variable. There should not ever
+    -- be any contention on this variable, so putting everything inside 'atomically'
+    -- is fine.
+    atomically $ do
+      oldState <- readTVar stateVar
+      if ledgerTipHash (clsState oldState) == blockPrevHash blk
+        then do
+          let !newState = oldState { clsState = applyBlk (clsConfig oldState) blk (clsState oldState) }
+          writeTVar stateVar newState
+          let mRewards = case (ledgerRewardUpdate (clsState newState), ledgerRewardUpdate (clsState oldState)) of
+                          (Nothing, Just r) -> Just r
+                          _otherwise -> Nothing
+          pure $ (newState, mRewards)
+        else panic $ mconcat
+               [ "applyBlock: Hash mismatch when applying block with slot no ", textShow (blockSlot blk), "\n"
+               , "applyBlock: ", textShow (ledgerTipHash $ clsState oldState), " /= ", textShow (blockPrevHash blk)
+               ]
+  where
+    applyBlk :: LedgerConfig CardanoBlock -> CardanoBlock -> LedgerState CardanoBlock -> LedgerState CardanoBlock
+    applyBlk cfg block lsb =
+      -- Set to False to get better error messages from Consensus (but slower block application).
+      if True
+        then tickThenReapply cfg block lsb
+        else case runExcept $ tickThenApply cfg block lsb of
+                Left err -> panic $ textShow err
+                Right result -> result
 
 saveLedgerState :: LedgerStateDir -> LedgerStateVar -> CardanoLedgerState -> SyncState -> IO ()
 saveLedgerState lsd@(LedgerStateDir stateDir) (LedgerStateVar stateVar) ledger synced = do
   atomically $ writeTVar stateVar ledger
   case synced of
     SyncFollowing -> saveState                      -- If following, save every state.
     SyncLagging
-      | unSlotNo slot == 0 -> pure ()               -- Genesis and firs EBB are weird so do not store them.
+      | unSlotNo slot == 0 -> pure ()               -- Genesis and the first EBB are weird so do not store them.
       | unSlotNo slot `mod` 10000 == 0 -> saveState -- Only save state ocassionally.
       | otherwise -> pure ()
   where
@@ -213,3 +232,10 @@ readLedgerState (LedgerStateVar stateVar) =
 safeRemoveFile :: FilePath -> IO ()
 safeRemoveFile fp = handle (\(_ :: IOException) -> pure ()) $ removeFile fp
 
+ledgerRewardUpdate :: LedgerState CardanoBlock -> Maybe (Shelley.RewardUpdate StandardShelley)
+ledgerRewardUpdate cls =
+  case cls of
+    LedgerStateByron _ -> Nothing
+    LedgerStateShelley sls -> Shelley.strictMaybeToMaybe . Shelley.nesRu
+                                $ Consensus.shelleyLedgerState sls
+    _otherwise -> panic "ledgerRewardUpdate: Bad pattern match (should be complete)"

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

@@ -25,8 +25,8 @@ import           Control.Monad.Trans.Reader (ReaderT)
 
 import           Database.Persist.Sql (SqlBackend)
 
-import           Ouroboros.Consensus.Cardano.Block (HardForkBlock (BlockByron, BlockShelley))
-
+import           Ouroboros.Consensus.Cardano.Block (HardForkBlock (BlockByron, BlockShelley),
+                    LedgerState (LedgerStateByron, LedgerStateShelley))
 
 
 -- | The default DbSyncNodePlugin.
@@ -48,12 +48,14 @@ insertDefaultBlock
 insertDefaultBlock tracer env ledgerStateVar (BlockDetails cblk details) = do
   -- Calculate the new ledger state to pass to the DB insert functions but do not yet
   -- update ledgerStateVar.
-  newLedgerState <- liftIO $ applyBlock ledgerStateVar cblk
-  res <- case cblk of
-            BlockByron blk ->
+  (newLedgerState, mRewards) <- liftIO $ applyBlock ledgerStateVar cblk
+  res <- case (cblk, clsState newLedgerState) of
+            (BlockByron blk, LedgerStateByron _st) ->
               Byron.insertByronBlock tracer blk details
-            BlockShelley blk ->
-              Shelley.insertShelleyBlock tracer env blk (clsState newLedgerState) details
+            (BlockShelley blk, LedgerStateShelley lstate) ->
+              Shelley.insertShelleyBlock tracer env blk lstate mRewards details
+            -- Should never happen.
+            _otherwise -> panic "insertDefaultBlock: Era mismatch on block and ledger state"
   -- Now we update it in ledgerStateVar and (possibly) store it to disk.
   liftIO $ saveLedgerState (LedgerStateDir "ledger-state") ledgerStateVar
                 newLedgerState (isSyncedWithinSeconds details 60)

cardano-db-sync/src/Cardano/DbSync/Plugin/Epoch.hs

@@ -30,8 +30,7 @@ import           Database.Esqueleto (Value (..), (^.), (==.),
                     desc, from, limit, orderBy, select, val, where_)
 
 import           Database.Persist.Class (replace)
-import           Database.Persist.Sql (IsolationLevel (Serializable), SqlBackend,
-                    transactionSaveWithIsolation)
+import           Database.Persist.Sql (SqlBackend)
 
 import           Cardano.Db (EpochId, EntityField (..), listToMaybe)
 import qualified Cardano.Db as DB
@@ -123,7 +122,7 @@ latestCachedEpochVar = unsafePerformIO $ newIORef Nothing -- Gets updated later.
 
 updateEpochNum :: (MonadBaseControl IO m, MonadIO m) => Word64 -> Trace IO Text -> ReaderT SqlBackend m (Either DbSyncNodeError ())
 updateEpochNum epochNum trce = do
-    transactionSaveWithIsolation Serializable
+    DB.transactionCommit
     mid <- queryEpochId epochNum
     res <- maybe insertEpoch updateEpoch mid
     liftIO $ atomicWriteIORef latestCachedEpochVar (Just epochNum)