Add Epoch Stake Thread

Author: kderme

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

@@ -45,6 +45,7 @@ library
                         Cardano.DbSync.Api
                         Cardano.DbSync.Api.Ledger
                         Cardano.DbSync.Api.Types
+                        Cardano.DbSync.Ledger.Async
                         Cardano.DbSync.Config
                         Cardano.DbSync.Config.Alonzo
                         Cardano.DbSync.Config.Byron
@@ -130,6 +131,7 @@ library
                         Cardano.DbSync.LocalStateQuery
                         Cardano.DbSync.StateQuery
                         Cardano.DbSync.Sync
+                        Cardano.DbSync.Threads
                         Cardano.DbSync.Tracing.ToObjectOrphans
                         Cardano.DbSync.Types
 

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

@@ -41,6 +41,7 @@ import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.OffChain (runFetchOffChainPoolThread, runFetchOffChainVoteThread)
 import Cardano.DbSync.Rollback (unsafeRollback)
 import Cardano.DbSync.Sync (runSyncNodeClient)
+import Cardano.DbSync.Threads
 import Cardano.DbSync.Tracing.ToObjectOrphans ()
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util.Constraint (queryIsJsonbInSchema)
@@ -206,6 +207,7 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
               id
               [ runDbThread syncEnv metricsSetters threadChannels
               , runSyncNodeClient metricsSetters syncEnv iomgr trce threadChannels (enpSocketPath syncNodeParams)
+              , runEpochStakeThread syncEnv
               , runFetchOffChainPoolThread syncEnv
               , runFetchOffChainVoteThread syncEnv
               , runLedgerStateWriteThread (getTrace syncEnv) (envLedgerEnv syncEnv)

cardano-db-sync/src/Cardano/DbSync/Cache/Types.hs

@@ -52,7 +52,7 @@ import Ouroboros.Consensus.Cardano.Block (StandardCrypto)
 
 type StakePoolCache = Map PoolKeyHash DB.PoolHashId
 
--- | We use a stable cache for entries that are expected to be reused frequentyl.
+-- | We use a stable cache for entries that are expected to be reused frequently.
 -- These are stake addresses that have rewards, delegations etc.
 -- They are never removed unless manually eg when it's deregistered
 -- The LRU cache is much smaller for the rest stake addresses.

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

@@ -142,7 +142,7 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
           isMember
           applyResult
 
-  -- Here we insert the block and it's txs, but in adition we also cache some values which we later
+  -- Here we insert the block and it's txs, but in addition we also cache some values which we later
   -- use when updating the Epoch, thus saving us having to recalulating them later.
   case cblk of
     BlockByron blk ->

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

@@ -13,24 +13,22 @@ module Cardano.DbSync.Era.Shelley.Generic.StakeDist (
   StakeSliceRes (..),
   StakeSlice (..),
   getSecurityParameter,
-  getStakeSlice,
+  snapShotToList,
+  getSnapShot,
   getPoolDistr,
 ) where
 
 import Cardano.DbSync.Types
 import Cardano.Ledger.Coin (Coin (..))
 import qualified Cardano.Ledger.Compactible as Ledger
-import Cardano.Ledger.Credential (Credential)
 import qualified Cardano.Ledger.EpochBoundary as Ledger
 import Cardano.Ledger.Era (EraCrypto)
-import Cardano.Ledger.Keys (KeyHash (..), KeyRole (..))
 import qualified Cardano.Ledger.Shelley.LedgerState as Shelley
 import Cardano.Ledger.Val ((<+>))
 import Cardano.Prelude
 import qualified Data.Map.Strict as Map
 import Data.VMap (VB, VMap (..), VP)
 import qualified Data.VMap as VMap
-import qualified Data.Vector.Generic as VG
 import Lens.Micro
 import Ouroboros.Consensus.Block
 import Ouroboros.Consensus.Cardano.Block (LedgerState (..), StandardCrypto)
@@ -53,9 +51,6 @@ data StakeSlice = StakeSlice
   }
   deriving (Eq)
 
-emptySlice :: EpochNo -> StakeSlice
-emptySlice epoch = StakeSlice epoch Map.empty
-
 getSecurityParameter ::
   ConsensusProtocol (BlockProtocol blk) =>
   ProtocolInfo blk ->
@@ -70,112 +65,50 @@ getSecurityParameter = maxRollbacks . configSecurityParam . pInfoConfig
 -- On mainnet, for a value minSliceSize = 2000, it will be used as the actual size of slices
 -- until the size of delegations grows up to 8.6M, in which case, the size of slices
 -- will be adjusted.
-getStakeSlice ::
-  ConsensusProtocol (BlockProtocol blk) =>
-  ProtocolInfo blk ->
-  Word64 ->
+getSnapShot ::
   ExtLedgerState CardanoBlock ->
-  Bool ->
-  StakeSliceRes
-getStakeSlice pInfo !epochBlockNo els isMigration =
+  Maybe (Ledger.SnapShot StandardCrypto, EpochNo)
+getSnapShot els =
   case ledgerState els of
-    LedgerStateByron _ -> NoSlices
-    LedgerStateShelley sls -> genericStakeSlice pInfo epochBlockNo sls isMigration
-    LedgerStateAllegra als -> genericStakeSlice pInfo epochBlockNo als isMigration
-    LedgerStateMary mls -> genericStakeSlice pInfo epochBlockNo mls isMigration
-    LedgerStateAlonzo als -> genericStakeSlice pInfo epochBlockNo als isMigration
-    LedgerStateBabbage bls -> genericStakeSlice pInfo epochBlockNo bls isMigration
-    LedgerStateConway cls -> genericStakeSlice pInfo epochBlockNo cls isMigration
-
-genericStakeSlice ::
-  forall era c blk p.
-  (c ~ StandardCrypto, EraCrypto era ~ c, ConsensusProtocol (BlockProtocol blk)) =>
-  ProtocolInfo blk ->
-  Word64 ->
+    LedgerStateByron _ -> Nothing
+    LedgerStateShelley sls -> Just $ genericSnapShot sls
+    LedgerStateAllegra als -> Just $ genericSnapShot als
+    LedgerStateMary mls -> Just $ genericSnapShot mls
+    LedgerStateAlonzo als -> Just $ genericSnapShot als
+    LedgerStateBabbage bls -> Just $ genericSnapShot bls
+    LedgerStateConway cls -> Just $ genericSnapShot cls
+
+genericSnapShot ::
+  forall era p.
+  (EraCrypto era ~ StandardCrypto) =>
   LedgerState (ShelleyBlock p era) ->
-  Bool ->
-  StakeSliceRes
-genericStakeSlice pInfo epochBlockNo lstate isMigration
-  | index > delegationsLen = NoSlices
-  | index == delegationsLen = Slice (emptySlice epoch) True
-  | index + size > delegationsLen = Slice (mkSlice (delegationsLen - index)) True
-  | otherwise = Slice (mkSlice size) False
+  (Ledger.SnapShot StandardCrypto, EpochNo)
+genericSnapShot lstate = (stakeSnapshot, epoch)
   where
-    epoch :: EpochNo
-    epoch = EpochNo $ 1 + unEpochNo (Shelley.nesEL (Consensus.shelleyLedgerState lstate))
-
-    minSliceSize :: Word64
-    minSliceSize = 2000
-
-    -- On mainnet this is 2160
-    k :: Word64
-    k = getSecurityParameter pInfo
-
     -- 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 c
+    stakeSnapshot :: Ledger.SnapShot StandardCrypto
     stakeSnapshot =
       Ledger.ssStakeMark . Shelley.esSnapshots . Shelley.nesEs $
         Consensus.shelleyLedgerState lstate
 
-    delegations :: VMap.KVVector VB VB (Credential 'Staking c, KeyHash 'StakePool c)
-    delegations = VMap.unVMap $ Ledger.ssDelegations stakeSnapshot
-
-    delegationsLen :: Word64
-    delegationsLen = fromIntegral $ VG.length delegations
+    epoch = EpochNo $ 1 + unEpochNo (Shelley.nesEL (Consensus.shelleyLedgerState lstate))
 
-    stakes :: VMap VB VP (Credential 'Staking c) (Ledger.CompactForm Coin)
-    stakes = Ledger.unStake $ Ledger.ssStake stakeSnapshot
+snapShotToList ::
+  Ledger.SnapShot StandardCrypto ->
+  [(StakeCred, (Coin, PoolKeyHash))]
+snapShotToList snapShot =
+  VMap.toList $
+    VMap.mapMaybe id $ -- This line removes entries without stake. Should we assume 0 and insert it?
+      VMap.mapWithKey (\a p -> (,p) <$> lookupStake a) (Ledger.ssDelegations snapShot)
+  where
+    stakes :: VMap VB VP StakeCred (Ledger.CompactForm Coin)
+    stakes = Ledger.unStake $ Ledger.ssStake snapShot
 
-    lookupStake :: Credential 'Staking c -> Maybe Coin
+    lookupStake :: StakeCred -> Maybe Coin
     lookupStake cred = Ledger.fromCompact <$> VMap.lookup cred stakes
 
-    -- This is deterministic for the whole epoch and is the constant size of slices
-    -- until the data are over. This means the last slice could be of smaller size and slices
-    -- after that will be empty.
-    epochSliceSize :: Word64
-    epochSliceSize =
-      max minSliceSize defaultEpochSliceSize
-      where
-        -- On mainnet this is 21600
-        expectedBlocks :: Word64
-        expectedBlocks = 10 * k
-
-        -- This size of slices is enough to cover the whole list, even if only
-        -- the 20% of the expected blocks appear in an epoch.
-        defaultEpochSliceSize :: Word64
-        defaultEpochSliceSize = 1 + div (delegationsLen * 5) expectedBlocks
-
-    -- The starting index of the data in the delegation vector.
-    index :: Word64
-    index
-      | isMigration = 0
-      | epochBlockNo < k = delegationsLen + 1 -- so it creates the empty Slice.
-      | otherwise = (epochBlockNo - k) * epochSliceSize
-
-    size :: Word64
-    size
-      | isMigration, epochBlockNo + 1 < k = 0
-      | isMigration = (epochBlockNo + 1 - k) * epochSliceSize
-      | otherwise = epochSliceSize
-
-    mkSlice :: Word64 -> StakeSlice
-    mkSlice actualSize =
-      StakeSlice
-        { sliceEpochNo = epoch
-        , sliceDistr = distribution
-        }
-      where
-        delegationsSliced :: VMap VB VB (Credential 'Staking c) (KeyHash 'StakePool c)
-        delegationsSliced = VMap $ VG.slice (fromIntegral index) (fromIntegral actualSize) delegations
-
-        distribution :: Map StakeCred (Coin, PoolKeyHash)
-        distribution =
-          VMap.toMap $
-            VMap.mapMaybe id $
-              VMap.mapWithKey (\a p -> (,p) <$> lookupStake a) delegationsSliced
-
 getPoolDistr ::
   ExtLedgerState CardanoBlock ->
   Maybe (Map PoolKeyHash (Coin, Word64), Map PoolKeyHash Natural)

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

@@ -152,8 +152,6 @@ insertBlockUniversal syncEnv shouldLog withinTwoMins withinHalfHour blk details
     whenStrictJust (apNewEpoch applyResult) $ \newEpoch -> do
       insertOnNewEpoch syncEnv blkId (Generic.blkSlotNo blk) epochNo newEpoch
 
-    insertStakeSlice syncEnv $ apStakeSlice applyResult
-
     when (ioGov iopts && (withinHalfHour || unBlockNo (Generic.blkBlockNo blk) `mod` 10000 == 0))
       . lift
       $ insertOffChainVoteResults tracer (envOffChainVoteResultQueue syncEnv)

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

@@ -15,6 +15,7 @@ module Cardano.DbSync.Era.Universal.Epoch (
   insertRewards,
   hasNewEpochEvent,
   hasEpochStartEvent,
+  insertEpochStake,
   insertRewardRests,
   insertProposalRefunds,
   insertPoolDepositRefunds,
@@ -200,7 +201,7 @@ insertStakeSlice ::
   ExceptT SyncNodeError (ReaderT SqlBackend m) ()
 insertStakeSlice _ Generic.NoSlices = pure ()
 insertStakeSlice syncEnv (Generic.Slice slice finalSlice) = do
-  insertEpochStake syncEnv network (Generic.sliceEpochNo slice) (Map.toList $ Generic.sliceDistr slice)
+  insertEpochStake syncEnv (Generic.sliceEpochNo slice) (Map.toList $ Generic.sliceDistr slice)
   when finalSlice $ do
     lift $ DB.updateSetComplete $ unEpochNo $ Generic.sliceEpochNo slice
     size <- lift $ DB.queryEpochStakeCount (unEpochNo $ Generic.sliceEpochNo slice)
@@ -211,17 +212,14 @@ insertStakeSlice syncEnv (Generic.Slice slice finalSlice) = do
     tracer :: Trace IO Text
     tracer = getTrace syncEnv
 
-    network :: Network
-    network = getNetwork syncEnv
-
+-- This is used by the epoch stake thread.
 insertEpochStake ::
   (MonadBaseControl IO m, MonadIO m) =>
   SyncEnv ->
-  Network ->
   EpochNo ->
   [(StakeCred, (Shelley.Coin, PoolKeyHash))] ->
   ExceptT SyncNodeError (ReaderT SqlBackend m) ()
-insertEpochStake syncEnv nw epochNo stakeChunk = do
+insertEpochStake syncEnv epochNo stakeChunk = do
   let cache = envCache syncEnv
   DB.ManualDbConstraints {..} <- liftIO $ readTVarIO $ envDbConstraints syncEnv
   dbStakes <- mapM (mkStake cache) stakeChunk
@@ -235,8 +233,9 @@ insertEpochStake syncEnv nw epochNo stakeChunk = do
       (StakeCred, (Shelley.Coin, PoolKeyHash)) ->
       ExceptT SyncNodeError (ReaderT SqlBackend m) DB.EpochStake
     mkStake cache (saddr, (coin, pool)) = do
-      saId <- lift $ queryOrInsertStakeAddress trce cache UpdateCacheStrong nw saddr
-      poolId <- lift $ queryPoolKeyOrInsert "insertEpochStake" trce cache UpdateCache (ioShelley iopts) pool
+      -- TODO check that not updating the cache here is not an issue.
+      saId <- lift $ queryOrInsertStakeAddress trce cache DoNotUpdateCache network saddr
+      poolId <- lift $ queryPoolKeyOrInsert "insertEpochStake" trce cache DoNotUpdateCache (ioShelley iopts) pool
       pure $
         DB.EpochStake
           { DB.epochStakeAddrId = saId
@@ -247,6 +246,7 @@ insertEpochStake syncEnv nw epochNo stakeChunk = do
 
     trce = getTrace syncEnv
     iopts = getInsertOptions syncEnv
+    network = getNetwork syncEnv
 
 insertRewards ::
   (MonadBaseControl IO m, MonadIO m) =>

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

@@ -10,10 +10,9 @@ module Cardano.DbSync.Error (
   annotateInvariantTx,
   bsBase16Encode,
   dbSyncNodeError,
-  dbSyncInvariant,
   renderSyncInvariant,
   runOrThrowIO,
-  fromEitherSTM,
+  throwLeftIO,
   logAndThrowIO,
   shouldAbortOnPanic,
   hasAbortOnPanicEnv,
@@ -154,9 +153,6 @@ annotateInvariantTx tx ei =
 dbSyncNodeError :: (Monad m) => Text -> ExceptT SyncNodeError m a
 dbSyncNodeError = left . SNErrDefault
 
-dbSyncInvariant :: (Monad m) => Text -> SyncInvariant -> ExceptT SyncNodeError m a
-dbSyncInvariant loc = left . SNErrInvariant loc
-
 renderSyncInvariant :: SyncInvariant -> Text
 renderSyncInvariant ei =
   case ei of
@@ -174,8 +170,8 @@ renderSyncInvariant ei =
         , textShow tx
         ]
 
-fromEitherSTM :: (Exception e) => Either e a -> STM a
-fromEitherSTM = either throwSTM return
+throwLeftIO :: Exception e => Either e a -> IO a
+throwLeftIO = either throwIO pure
 
 bsBase16Encode :: ByteString -> Text
 bsBase16Encode bs =

cardano-db-sync/src/Cardano/DbSync/Ledger/Async.hs

@@ -0,0 +1,44 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Cardano.DbSync.Ledger.Async where
+
+import Cardano.DbSync.Ledger.Types
+import Cardano.Ledger.BaseTypes (EpochNo)
+import Cardano.Ledger.Crypto (StandardCrypto)
+import qualified Cardano.Ledger.EpochBoundary as Ledger
+import Control.Concurrent.Class.MonadSTM.Strict
+import qualified Control.Concurrent.STM.TBQueue as TBQ
+
+newEpochStakeChannels :: IO EpochStakeChannels
+newEpochStakeChannels =
+  -- This may never be more than 1. But let's keep it a queue for extensibility shake.
+  -- This may allow us to parallelize the events workload even further
+  EpochStakeChannels
+    <$> TBQ.newTBQueueIO 1
+    <*> newTVarIO Nothing
+
+-- To be used by the main thread
+ensureEpochDone :: EpochStakeChannels -> EpochNo -> Ledger.SnapShot StandardCrypto -> IO ()
+ensureEpochDone sQueue epoch snapshot = atomically $ do
+  mLastEpochDone <- waitFinished sQueue
+  case mLastEpochDone of
+    Just lastEpochDone | lastEpochDone == epoch -> pure ()
+    _ -> do
+      -- If last is not already there, put it to list and wait again
+      writeStakeAction sQueue epoch snapshot True
+      retry
+
+-- To be used by the main thread
+waitFinished :: EpochStakeChannels -> STM IO (Maybe EpochNo)
+waitFinished sQueue = do
+  stakeThreadState <- readTVar (epochResult sQueue)
+  case stakeThreadState of
+    Just (lastEpoch, Done) -> pure $ Just lastEpoch -- Normal case
+    Just (_, Running) -> retry -- Wait to finish current work.
+    Nothing -> pure Nothing -- This will happen after a restart
+
+-- To be used by the main thread
+writeStakeAction :: EpochStakeChannels -> EpochNo -> Ledger.SnapShot StandardCrypto -> Bool -> STM IO ()
+writeStakeAction sQueue epoch snapShot checkFirst = do
+  TBQ.writeTBQueue (estakeQueue sQueue) $ EpochStakeDBAction epoch snapShot checkFirst
+  writeTVar (epochResult sQueue) $ Just (epoch, Running)