Introduce ApplyResult and make ledger state opaque

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

@@ -208,7 +208,8 @@ executable cardano-db-sync
                         -Wunused-packages
                         -Wno-unsafe
                         -threaded
-                        -with-rtsopts=-N3
+                        -rtsopts
+                        "-with-rtsopts=-T -A16m -N3 --disable-delayed-os-memory-return"
 
   autogen-modules:      Paths_cardano_db_sync
                         MigrationValidations

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

@@ -121,6 +121,7 @@ mkSyncEnv
     -> IO SyncEnv
 mkSyncEnv trce backend syncOptions protoInfo nw nwMagic systemStart dir stableEpochSlot = do
   ledgerEnv <- mkLedgerEnv trce protoInfo dir nw stableEpochSlot systemStart (soptAbortOnInvalid syncOptions)
+                 (snapshotEveryFollowing syncOptions) (snapshotEveryLagging syncOptions)
   cache <- newEmptyCache 200000
   pure $ SyncEnv
           { envProtocol = SyncProtocolCardano

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

@@ -59,7 +59,7 @@ type StakePoolCache = Map StakePoolKeyHash DB.PoolHashId
 -- insertions, where the cache has not been initiated yet.
 data Cache
   = UninitiatedCache
-  | Cache CacheInternal
+  | Cache !CacheInternal
 
 data CacheNew
   = CacheNew

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

@@ -144,7 +144,7 @@ insertBlockList
     :: SyncEnv -> [CardanoBlock]
     -> ExceptT SyncNodeError IO ()
 insertBlockList env blks =
-  newExceptT $ insertDefaultBlock env blks
+  newExceptT $ insertListBlocks env blks
 
 -- | Split the DbAction list into a prefix containing blocks to apply and a postfix.
 spanDbApply :: [DbAction] -> ([CardanoBlock], [DbAction])

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

@@ -4,14 +4,14 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 {-# LANGUAGE OverloadedStrings #-}
 module Cardano.DbSync.Default
-  ( insertDefaultBlock
+  ( insertListBlocks
   , rollbackToPoint
   ) where
 
 
 import           Cardano.Prelude
 
-import           Cardano.BM.Trace (logDebug, logInfo)
+import           Cardano.BM.Trace (logInfo)
 
 import qualified Cardano.Db as DB
 
@@ -26,24 +26,20 @@ import           Cardano.DbSync.Era.Shelley.Insert (insertShelleyBlock)
 import           Cardano.DbSync.Era.Shelley.Insert.Epoch (insertPoolDepositRefunds, insertRewards)
 import           Cardano.DbSync.Era.Shelley.Validate (validateEpochRewards)
 import           Cardano.DbSync.Error
-import           Cardano.DbSync.LedgerState (LedgerEvent (..), LedgerStateSnapshot (..), applyBlock,
-                   getAlonzoPParams, getBabbagePParams, saveCleanupState)
+import           Cardano.DbSync.LedgerState (LedgerEvent (..), ApplyResult (..), applyBlockAndSnapshot)
 import           Cardano.DbSync.Rollback (rollbackToPoint)
 import           Cardano.DbSync.Types
 import           Cardano.DbSync.Util
 
-import qualified Cardano.Ledger.Alonzo.PParams as Alonzo
-import qualified Cardano.Ledger.Babbage.PParams as Babbage
+import qualified Cardano.Ledger.Alonzo.Scripts as Ledger
 import           Cardano.Ledger.BaseTypes (Network)
 import           Cardano.Ledger.Coin (Coin (..))
 import           Cardano.Ledger.Credential (StakeCredential)
 import           Cardano.Ledger.Crypto (StandardCrypto)
 import           Cardano.Ledger.Keys (KeyHash (..), KeyRole (..))
 
-import           Cardano.Slotting.Block (BlockNo (..))
 import           Cardano.Slotting.Slot (EpochNo (..))
 
-
 import           Control.Monad.Logger (LoggingT)
 import           Control.Monad.Trans.Control (MonadBaseControl)
 import           Control.Monad.Trans.Except.Extra (newExceptT)
@@ -56,79 +52,64 @@ import           Database.Persist.SqlBackend.Internal.StatementCache
 import           Database.Persist.SqlBackend.Internal
 
 import           Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
-import           Ouroboros.Network.Block (blockNo)
 
 
-insertDefaultBlock
+insertListBlocks
     :: SyncEnv -> [CardanoBlock]
     -> IO (Either SyncNodeError ())
-insertDefaultBlock env blocks =
+insertListBlocks env blocks = do
     DB.runDbIohkLogging backend tracer .
       runExceptT $ do
-        traverse_ insertDetails blocks
+        traverse_ (applyAndInsert env) blocks
   where
     tracer = getTrace env
     backend = envBackend env
 
-    insertDetails
-        :: CardanoBlock -> ExceptT SyncNodeError (ReaderT SqlBackend (LoggingT IO)) ()
-    insertDetails cblk = do
-      -- Calculate the new ledger state to pass to the DB insert functions but do not yet
-      -- update ledgerStateVar.
-      !lStateSnap <- liftIO $ applyBlock (envLedger env) cblk
-      let !details = lssSlotDetails lStateSnap
-      mkSnapshotMaybe env lStateSnap (blockNo cblk) (isSyncedWithinSeconds details 600)
-      handleLedgerEvents env (sdEpochNo details) (lssEvents lStateSnap)
-      let firstBlockOfEpoch = hasEpochStartEvent (lssEvents lStateSnap)
-      case cblk of
-        BlockByron blk ->
-          newExceptT $ insertByronBlock env firstBlockOfEpoch blk details
-        BlockShelley blk ->
-          newExceptT $ insertShelleyBlock env firstBlockOfEpoch (Generic.fromShelleyBlock blk) lStateSnap details
-        BlockAllegra blk ->
-          newExceptT $ insertShelleyBlock env firstBlockOfEpoch (Generic.fromAllegraBlock blk) lStateSnap details
-        BlockMary blk ->
-          newExceptT $ insertShelleyBlock env firstBlockOfEpoch (Generic.fromMaryBlock blk) lStateSnap details
-        BlockAlonzo blk -> do
-          let prices = Alonzo._prices $ getAlonzoPParams $ lssState lStateSnap
-          newExceptT $ insertShelleyBlock env firstBlockOfEpoch (Generic.fromAlonzoBlock prices blk) lStateSnap details
-        BlockBabbage blk -> do
-          let prices = Babbage._prices $ getBabbagePParams $ lssState lStateSnap
-          newExceptT $ insertShelleyBlock env firstBlockOfEpoch (Generic.fromBabbageBlock prices blk) lStateSnap details
-      when (soptExtended $ envOptions env) .
-        newExceptT $ epochInsert tracer (BlockDetails cblk details)
-
-mkSnapshotMaybe
-        :: (MonadBaseControl IO m, MonadIO m)
-        => SyncEnv -> LedgerStateSnapshot -> BlockNo -> DB.SyncState
-        -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
-mkSnapshotMaybe env snapshot blkNo syncState =
-    case maybeFromStrict (lssNewEpoch snapshot) of
-      Just newEpoch -> do
-        liftIO $ logDebug (leTrace $ envLedger env) "Preparing for a snapshot"
-        let newEpochNo = Generic.neEpoch newEpoch
-        liftIO $ logDebug (leTrace $ envLedger env) "Taking a ledger a snapshot"
-        -- finally take a ledger snapshot
-        -- TODO: Instead of newEpochNo - 1, is there any way to get the epochNo from 'lssOldState'?
-        liftIO $ saveCleanupState (envLedger env) (lssOldState snapshot) (Just $ newEpochNo - 1)
-      Nothing ->
-        when (timeToSnapshot syncState blkNo) $
-          liftIO $ saveCleanupState (envLedger env) (lssOldState snapshot) Nothing
-
+applyAndInsert
+    :: SyncEnv -> CardanoBlock -> ExceptT SyncNodeError (ReaderT SqlBackend (LoggingT IO)) ()
+applyAndInsert env cblk = do
+    !applyResult <- liftIO $ applyBlockAndSnapshot (envLedger env) cblk
+    let !details = apSlotDetails applyResult
+    insertLedgerEvents env (sdEpochNo details) (apEvents applyResult)
+    insertEpoch details
+    let firstBlockOfEpoch = hasEpochStartEvent (apEvents applyResult)
+    let isMember = \poolId -> Set.member poolId (apPoolsRegistered applyResult)
+    case cblk of
+      BlockByron blk ->
+        newExceptT $ insertByronBlock env firstBlockOfEpoch blk details
+      BlockShelley blk -> newExceptT $
+        insertShelleyBlock env firstBlockOfEpoch (Generic.fromShelleyBlock blk)
+          details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
+      BlockAllegra blk -> newExceptT $
+        insertShelleyBlock env firstBlockOfEpoch (Generic.fromAllegraBlock blk)
+          details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
+      BlockMary blk -> newExceptT $
+        insertShelleyBlock env firstBlockOfEpoch (Generic.fromMaryBlock blk)
+          details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
+      BlockAlonzo blk -> newExceptT $
+        insertShelleyBlock env firstBlockOfEpoch (Generic.fromAlonzoBlock (getPrices applyResult) blk)
+          details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
+      BlockBabbage blk -> newExceptT $
+        insertShelleyBlock env firstBlockOfEpoch (Generic.fromBabbageBlock (getPrices applyResult) blk)
+          details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
   where
-    timeToSnapshot :: DB.SyncState -> BlockNo -> Bool
-    timeToSnapshot syncSt bNo =
-      case (syncSt, unBlockNo bNo) of
-        (DB.SyncFollowing, bno) -> bno `mod` snapshotEveryFollowing (envOptions env) == 0
-        (DB.SyncLagging, bno) -> bno `mod` snapshotEveryLagging (envOptions env) == 0
+    tracer = getTrace env
+
+    insertEpoch details = when (soptExtended $ envOptions env) .
+      newExceptT $ epochInsert tracer (BlockDetails cblk details)
+
+    getPrices :: ApplyResult -> Ledger.Prices
+    getPrices applyResult = case apPrices applyResult of
+      Strict.Just pr -> pr
+      Strict.Nothing -> Ledger.Prices minBound minBound
 
 -- -------------------------------------------------------------------------------------------------
 
-handleLedgerEvents
+insertLedgerEvents
     :: (MonadBaseControl IO m, MonadIO m)
     => SyncEnv -> EpochNo -> [LedgerEvent]
     -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
-handleLedgerEvents env currentEpochNo@(EpochNo curEpoch) =
+insertLedgerEvents env currentEpochNo@(EpochNo curEpoch) =
     mapM_ handler
   where
     tracer = getTrace env
@@ -141,14 +122,18 @@ handleLedgerEvents env currentEpochNo@(EpochNo curEpoch) =
       if unEpochNo currentEpochNo >= m then EpochNo $ unEpochNo currentEpochNo - m
       else EpochNo 0
 
+    toSyncState :: SyncState -> DB.SyncState
+    toSyncState SyncLagging = DB.SyncLagging
+    toSyncState SyncFollowing = DB.SyncFollowing
+
     handler
         :: (MonadBaseControl IO m, MonadIO m)
         => LedgerEvent -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
     handler ev =
       case ev of
         LedgerNewEpoch en ss -> do
           lift $ do
-            insertEpochSyncTime en ss (leEpochSyncTime lenv)
+            insertEpochSyncTime en (toSyncState ss) (leEpochSyncTime lenv)
           sqlBackend <- lift ask
           persistantCacheSize <- liftIO $ statementCacheSize $ connStmtMap sqlBackend
           liftIO . logInfo tracer $ "Persistant SQL Statement Cache size is " <> textShow persistantCacheSize
@@ -176,8 +161,9 @@ handleLedgerEvents env currentEpochNo@(EpochNo curEpoch) =
             let rewards = Map.toList rwd
             insertRewards (subFromCurrentEpoch 1) currentEpochNo cache rewards
             liftIO . logInfo tracer $ "Inserted " <> show (length rewards) <> " Mir rewards"
-        LedgerPoolReap en drs ->
-          insertPoolDepositRefunds env (Generic.Rewards en $ convertPoolDepositReunds (leNetwork lenv) drs)
+        LedgerPoolReap en drs -> do
+          unless (Map.null drs) $ do
+            insertPoolDepositRefunds env (Generic.Rewards en $ convertPoolDepositReunds (leNetwork lenv) drs)
 
 convertPoolDepositReunds
     :: Network -> Map (StakeCredential StandardCrypto) (Map (KeyHash 'StakePool StandardCrypto) Coin)

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

@@ -24,7 +24,7 @@ import           Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef)
 import           Database.Esqueleto.Experimental (SqlBackend, desc, from, limit, orderBy, replace,
                    select, table, unValue, val, where_, (==.), (^.))
 
-import           Cardano.Db (EntityField (..), EpochId, SyncState (..))
+import           Cardano.Db (EntityField (..), EpochId)
 import qualified Cardano.Db as DB
 
 import           Cardano.DbSync.Api

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

@@ -30,7 +30,7 @@ import qualified Cardano.Chain.UTxO as Byron
 
 import qualified Cardano.Crypto as Crypto (serializeCborHash)
 
-import           Cardano.Db (DbLovelace (..), SyncState (..))
+import           Cardano.Db (DbLovelace (..))
 import           Cardano.DbSync.Era.Util (liftLookupFail)
 
 import           Cardano.DbSync.Types

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

@@ -70,7 +70,7 @@ data Block = Block
   , blkVrfKey :: !Text
   , blkOpCert :: !ByteString
   , blkOpCertCounter :: !Word64
-  , blkTxs :: ![Tx]
+  , blkTxs :: [Tx] -- intentionally left lazy to delay the tx transformation
   }
 
 
@@ -219,7 +219,6 @@ blockTxs
     => ShelleyBlock p era -> [(Word64, Shelley.Tx era)]
 blockTxs = zip [0 ..] . unTxSeq . Ledger.bbody . Consensus.shelleyBlockRaw
 
--- (ShelleyBasedEra era, VRF (Crypto era) ~ PraosVRF) =>
 blockVrfKeyView :: VerKeyVRF StandardCrypto -> Text
 blockVrfKeyView = Api.serialiseToBech32 . Api.VrfVerificationKey
 

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
@@ -72,7 +73,7 @@ getSecurityParameter =  maxRollbacks . configSecurityParam . pInfoConfig
 getStakeSlice :: ConsensusProtocol (BlockProtocol blk)
               => ProtocolInfo IO blk -> Ledger.Network
               -> EpochNo -> Word64 -> Word64 -> ExtLedgerState CardanoBlock -> StakeSliceRes
-getStakeSlice pInfo network epoch sliceIndex minSliceSize els =
+getStakeSlice pInfo network epoch !sliceIndex !minSliceSize els =
   case ledgerState els of
     LedgerStateByron _ -> NoSlices
     LedgerStateShelley sls -> genericStakeSlice pInfo network epoch sliceIndex minSliceSize sls

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

@@ -265,7 +265,7 @@ insertStaking tracer cache blkId genesis = do
               }
   let params = zip [0..] $ Map.elems (sgsPools $ sgStaking genesis)
   let network = sgNetworkId genesis
-  forM_ params $ uncurry (insertPoolRegister tracer uninitiatedCache (Left 2) network 0 blkId txId)
+  forM_ params $ uncurry (insertPoolRegister tracer uninitiatedCache (\_ -> False) network 0 blkId txId)
   let stakes = zip [0..] $ Map.toList (sgsStake $ sgStaking genesis)
   forM_ stakes $ \(n, (keyStaking, keyPool)) -> do
     insertStakeRegistration (EpochNo 0) txId (2 * n) (Generic.annotateStakingCred network (KeyHashObj keyStaking))