db: Populate the Stake table

For each epoch, store '(stake-address, amount, pool-id)' where pool id
is the pool to which the stake address delegated.

Closes: #319

Author: erikd

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

@@ -60,6 +60,7 @@ 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.EpochBoundary 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
@@ -118,12 +119,11 @@ insertShelleyBlock tracer env blk lStateSnap details = do
         , ", hash ", renderByteArray (Shelley.blockHash blk)
         ]
 
-    whenJust (lssRewardUpdate lStateSnap) $ \ rewards ->
+    whenJust (lssEpochUpdate lStateSnap) $ \ esum -> do
       -- Subtract 2 from the epoch to calculate when the epoch in which the reward was earned.
-      insertRewards tracer env blkId (sdEpochNo details - 2) rewards
-
-    whenJust (lssParamUpdate lStateSnap) $ \ params ->
-      insertEpochParam tracer blkId (sdEpochNo details) params
+      insertRewards tracer env blkId (sdEpochNo details - 2) (esRewardUpdate esum)
+      insertEpochParam tracer blkId (sdEpochNo details) (esParamUpdate esum)
+      insertEpochStake tracer env blkId (sdEpochNo details) (esStakeUpdate esum)
 
     when (getSyncStatus details == SyncFollowing) $
       -- Serializiing things during syncing can drastically slow down full sync
@@ -496,7 +496,7 @@ insertParamProposal _tracer txId (Shelley.Update (Shelley.ProposedPPUpdates umap
           , DB.paramProposalMonetaryExpandRate = Shelley.unitIntervalToDouble <$> strictMaybeToMaybe (Shelley._rho pmap)
           , DB.paramProposalTreasuryGrowthRate = Shelley.unitIntervalToDouble <$> strictMaybeToMaybe (Shelley._tau pmap)
           , DB.paramProposalDecentralisation = Shelley.unitIntervalToDouble <$> strictMaybeToMaybe (Shelley._d pmap)
-          , DB.paramProposalEntropy = join (Shelley.nonceToBytes <$> strictMaybeToMaybe (Shelley._extraEntropy pmap))
+          , DB.paramProposalEntropy = Shelley.nonceToBytes =<< strictMaybeToMaybe (Shelley._extraEntropy pmap)
           , DB.paramProposalProtocolVersion = strictMaybeToMaybe (Shelley._protocolVersion pmap)
           , DB.paramProposalMinUtxoValue = fromIntegral . Shelley.unCoin <$> strictMaybeToMaybe (Shelley._minUTxOValue pmap)
           , DB.paramProposalMinPoolCost = fromIntegral . Shelley.unCoin <$> strictMaybeToMaybe (Shelley._minPoolCost pmap)
@@ -596,3 +596,28 @@ insertEpochParam _tracer blkId (EpochNo epoch) params =
       , DB.epochParamMinPoolCost = fromIntegral $ Shelley.unCoin (Shelley._minPoolCost params)
       , DB.epochParamBlockId = blkId
       }
+
+insertEpochStake
+    :: (MonadBaseControl IO m, MonadIO m)
+    => Trace IO Text -> DbSyncEnv -> DB.BlockId -> EpochNo -> Shelley.Stake StandardShelley
+    -> ExceptT DbSyncNodeError (ReaderT SqlBackend m) ()
+insertEpochStake _tracer env blkId (EpochNo epoch) smap =
+    mapM_ insert $ Map.toList (Shelley.unStake smap)
+  where
+    insert
+        :: (MonadBaseControl IO m, MonadIO m)
+        => (Shelley.Credential 'Shelley.Staking StandardShelley, Shelley.Coin)
+        -> ExceptT DbSyncNodeError (ReaderT SqlBackend m) ()
+    insert (saddr, coin) = do
+      (saId, poolId) <- firstExceptT (NELookup "insertEpochStake")
+                          . newExceptT
+                          $ queryStakeAddressAndPool epoch
+                                (Shelley.stakingCredHash env saddr)
+      void . lift . DB.insertEpochStake $
+        DB.EpochStake
+          { DB.epochStakeAddrId = saId
+          , DB.epochStakePoolId = poolId
+          , DB.epochStakeAmount = fromIntegral $ Shelley.unCoin coin
+          , DB.epochStakeEpochNo = epoch + 1 -- The epoch where this delegation becomes valid.
+          , DB.epochStakeBlockId = blkId
+          }

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

@@ -6,6 +6,7 @@
 
 module Cardano.DbSync.LedgerState
   ( CardanoLedgerState (..)
+  , EpochUpdate (..)
   , LedgerStateSnapshot (..)
   , LedgerStateVar (..)
   , applyBlock
@@ -29,7 +30,7 @@ import           Cardano.Prelude
 import           Cardano.Slotting.EpochInfo (EpochInfo, epochInfoEpoch)
 import           Cardano.Slotting.Slot (EpochNo (..), SlotNo (..), fromWithOrigin)
 
-import           Control.Concurrent.STM.TVar (TVar, newTVarIO, writeTVar, readTVar)
+import           Control.Concurrent.STM.TVar (TVar, newTVarIO, writeTVar, readTVar, readTVarIO)
 import           Control.Exception (IOException, handle)
 import qualified Control.Exception as Exception
 import           Control.Monad.Extra (firstJustM)
@@ -54,6 +55,7 @@ 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.EpochBoundary as Shelley
 import qualified Shelley.Spec.Ledger.LedgerState as Shelley
 import qualified Shelley.Spec.Ledger.PParams as Shelley
 
@@ -66,6 +68,12 @@ data CardanoLedgerState = CardanoLedgerState
   , clsCodec :: !(CodecConfig CardanoBlock)
   }
 
+data EpochUpdate = EpochUpdate
+  { esParamUpdate :: !(Shelley.PParams StandardShelley)
+  , esRewardUpdate :: !(Shelley.RewardUpdate StandardShelley)
+  , esStakeUpdate :: !(Shelley.Stake StandardShelley)
+  }
+
 newtype LedgerStateVar = LedgerStateVar
   { unLedgerStateVar :: TVar CardanoLedgerState
   }
@@ -77,10 +85,10 @@ data LedgerStateFile = LedgerStateFile -- Internal use only.
 
 data LedgerStateSnapshot = LedgerStateSnapshot
   { lssState :: !CardanoLedgerState
-  , lssRewardUpdate :: !(Maybe (Shelley.RewardUpdate StandardShelley))
-  , lssParamUpdate :: !(Maybe (Shelley.PParams StandardShelley))
+  , lssEpochUpdate :: !(Maybe EpochUpdate) -- Only Just for a single block at the epoch boundary
   }
 
+
 initLedgerStateVar :: GenesisConfig -> IO LedgerStateVar
 initLedgerStateVar genesisConfig = do
   LedgerStateVar <$>
@@ -108,19 +116,12 @@ applyBlock (LedgerStateVar stateVar) 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
-              mParams =
-                  if ledgerEpochNo newState == ledgerEpochNo oldState + 1
-                    then ledgerEpochProtocolParams (clsState newState)
-                    else Nothing
-
           pure $ LedgerStateSnapshot
                     { lssState = newState
-                    , lssRewardUpdate = mRewards
-                    , lssParamUpdate = mParams
+                    , lssEpochUpdate =
+                        if ledgerEpochNo newState == ledgerEpochNo oldState + 1
+                          then Just $ ledgerEpochUpdate (clsState newState) (ledgerRewardUpdate $ clsState oldState)
+                          else Nothing
                     }
         else panic $ mconcat
                [ "applyBlock: Hash mismatch when applying block with slot no ", textShow (blockSlot blk), "\n"
@@ -179,7 +180,7 @@ cleanupLedgerStateFiles stateDir slotNo = do
     -- Remove invalid (ie SlotNo >= current) ledger state files (occurs on rollback).
     mapM_ safeRemoveFile invalid
     -- Remove all but 8 most recent state files.
-    mapM_ safeRemoveFile $ map lsfFilePath (List.drop 8 valid)
+    mapM_ (safeRemoveFile . lsfFilePath) (List.drop 8 valid)
   where
     -- Left files are deleted, Right files are kept.
     keepFile :: LedgerStateFile ->  Either FilePath LedgerStateFile
@@ -246,8 +247,7 @@ listLedgerStateSlotNos :: LedgerStateDir -> IO [SlotNo]
 listLedgerStateSlotNos = fmap3 (SlotNo . lsfSlotNo) listLedgerStateFilesOrdered
 
 readLedgerState :: LedgerStateVar -> IO CardanoLedgerState
-readLedgerState (LedgerStateVar stateVar) =
-  atomically $ readTVar stateVar
+readLedgerState (LedgerStateVar stateVar) = readTVarIO stateVar
 
 -- | Remove given file path and ignore any IOEXceptions.
 safeRemoveFile :: FilePath -> IO ()
@@ -262,15 +262,30 @@ ledgerEpochNo cls =
     epochInfo :: EpochInfo Identity
     epochInfo = epochInfoLedger (clsConfig cls) $ hardForkLedgerStatePerEra (clsState cls)
 
-ledgerEpochProtocolParams :: LedgerState CardanoBlock -> Maybe (Shelley.PParams StandardShelley)
-ledgerEpochProtocolParams lsc =
-  case lsc of
-    LedgerStateByron _ -> Nothing
-    LedgerStateShelley sls -> Just $ Shelley.esPp (Shelley.nesEs $ Consensus.shelleyLedgerState sls)
-
+-- Create an EpochUpdate from the current epoch state and the rewards from the last epoch.
+ledgerEpochUpdate :: LedgerState CardanoBlock -> Maybe (Shelley.RewardUpdate StandardShelley) -> EpochUpdate
+ledgerEpochUpdate lcs mRewards =
+  case lcs of
+    LedgerStateByron _ -> panic "ledgerEpochUpdate: LedgerStateByron but should be Shelley"
+    LedgerStateShelley sls ->
+      EpochUpdate
+        { esParamUpdate = Shelley.esPp $ Shelley.nesEs (Consensus.shelleyLedgerState sls)
+        , esRewardUpdate = fromMaybe Shelley.emptyRewardUpdate mRewards
+
+        -- Use '_pstakeSet' here instead of '_pstateMark' because the stake addresses for the
+        -- later may not have been added to the database yet. That means that whne these values
+        -- are added to the database, the epoch number where they become active is the current
+        -- epoch plus one.
+        , esStakeUpdate = Shelley._stake . Shelley._pstakeSet . Shelley.esSnapshots
+                            $ Shelley.nesEs (Consensus.shelleyLedgerState sls)
+        }
+
+-- This will return a 'Just' from the time the rewards are updated until the end of the
+-- epoch. It is 'Nothing' for the first block of a new epoch (which is slightly inconvenient).
 ledgerRewardUpdate :: LedgerState CardanoBlock -> Maybe (Shelley.RewardUpdate StandardShelley)
 ledgerRewardUpdate lsc =
   case lsc of
-    LedgerStateByron _ -> Nothing
+    LedgerStateByron _ -> Nothing -- This actually happens on the Byron/Shelley boundary.
     LedgerStateShelley sls -> Shelley.strictMaybeToMaybe . Shelley.nesRu
                                 $ Consensus.shelleyLedgerState sls
+

cardano-db/src/Cardano/Db/Insert.hs

@@ -6,6 +6,7 @@ module Cardano.Db.Insert
   , insertDelegation
   , insertEpoch
   , insertEpochParam
+  , insertEpochStake
   , insertMeta
   , insertParamProposal
   , insertPoolHash
@@ -58,6 +59,9 @@ insertEpoch = insertByReturnKey "Epoch"
 insertEpochParam :: (MonadBaseControl IO m, MonadIO m) => EpochParam -> ReaderT SqlBackend m EpochParamId
 insertEpochParam = insertByReturnKey "EpochParam"
 
+insertEpochStake :: (MonadBaseControl IO m, MonadIO m) => EpochStake -> ReaderT SqlBackend m EpochStakeId
+insertEpochStake = insertByReturnKey "EpochStake"
+
 insertMeta :: (MonadBaseControl IO m, MonadIO m) => Meta -> ReaderT SqlBackend m MetaId
 insertMeta = insertByReturnKey "Meta"
 

cardano-db/src/Cardano/Db/Schema.hs

@@ -263,11 +263,13 @@ share
     blockId             BlockId
     UniqueReward        addrId blockId
 
-  Stake
+  EpochStake
     addrId              StakeAddressId
-    txId                TxId
-    stake               Word64              sqltype=lovelace
-    UniqueStake         addrId stake
+    poolId              PoolHashId
+    amount              Word64              sqltype=lovelace
+    epochNo             Word64
+    blockId             BlockId             -- To make rollbacks work correctly.
+    UniqueStake         addrId epochNo
 
   Treasury
     addrId              StakeAddressId

schema/migration-2-0003-20201006.sql renamed to schema/migration-2-0003-20201007.sql

@@ -76,10 +76,11 @@ BEGIN
     EXECUTE 'ALTER TABLE "reward" ADD CONSTRAINT "reward_addr_id_fkey" FOREIGN KEY("addr_id") REFERENCES "stake_address"("id")' ;
     EXECUTE 'ALTER TABLE "reward" ADD CONSTRAINT "reward_pool_id_fkey" FOREIGN KEY("pool_id") REFERENCES "pool_hash"("id")' ;
     EXECUTE 'ALTER TABLE "reward" ADD CONSTRAINT "reward_block_id_fkey" FOREIGN KEY("block_id") REFERENCES "block"("id")' ;
-    EXECUTE 'CREATe TABLE "stake"("id" SERIAL8  PRIMARY KEY UNIQUE,"addr_id" INT8 NOT NULL,"tx_id" INT8 NOT NULL,"stake" lovelace NOT NULL)' ;
-    EXECUTE 'ALTER TABLE "stake" ADD CONSTRAINT "unique_stake" UNIQUE("addr_id","stake")' ;
-    EXECUTE 'ALTER TABLE "stake" ADD CONSTRAINT "stake_addr_id_fkey" FOREIGN KEY("addr_id") REFERENCES "stake_address"("id")' ;
-    EXECUTE 'ALTER TABLE "stake" ADD CONSTRAINT "stake_tx_id_fkey" FOREIGN KEY("tx_id") REFERENCES "tx"("id")' ;
+    EXECUTE 'CREATe TABLE "epoch_stake"("id" SERIAL8  PRIMARY KEY UNIQUE,"addr_id" INT8 NOT NULL,"pool_id" INT8 NOT NULL,"amount" lovelace NOT NULL,"epoch_no" INT8 NOT NULL,"block_id" INT8 NOT NULL)' ;
+    EXECUTE 'ALTER TABLE "epoch_stake" ADD CONSTRAINT "unique_stake" UNIQUE("addr_id","epoch_no")' ;
+    EXECUTE 'ALTER TABLE "epoch_stake" ADD CONSTRAINT "epoch_stake_addr_id_fkey" FOREIGN KEY("addr_id") REFERENCES "stake_address"("id")' ;
+    EXECUTE 'ALTER TABLE "epoch_stake" ADD CONSTRAINT "epoch_stake_pool_id_fkey" FOREIGN KEY("pool_id") REFERENCES "pool_hash"("id")' ;
+    EXECUTE 'ALTER TABLE "epoch_stake" ADD CONSTRAINT "epoch_stake_block_id_fkey" FOREIGN KEY("block_id") REFERENCES "block"("id")' ;
     EXECUTE 'CREATe TABLE "treasury"("id" SERIAL8  PRIMARY KEY UNIQUE,"addr_id" INT8 NOT NULL,"cert_index" INT4 NOT NULL,"amount" lovelace NOT NULL,"tx_id" INT8 NOT NULL)' ;
     EXECUTE 'ALTER TABLE "treasury" ADD CONSTRAINT "unique_treasury" UNIQUE("addr_id","tx_id")' ;
     EXECUTE 'ALTER TABLE "treasury" ADD CONSTRAINT "treasury_addr_id_fkey" FOREIGN KEY("addr_id") REFERENCES "stake_address"("id")' ;