wip

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

@@ -43,6 +43,7 @@ library
                         Cardano.DbSync.Era.Cardano.Insert
                         Cardano.DbSync.Era.Cardano.Util
 
+                        Cardano.DbSync.Era.Shelley.Adjust
                         Cardano.DbSync.Era.Shelley.Genesis
                         Cardano.DbSync.Era.Shelley.Generic
                         Cardano.DbSync.Era.Shelley.Generic.Block

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

@@ -0,0 +1,79 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE NoImplicitPrelude #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeFamilies #-}
+
+module Cardano.DbSync.Era.Shelley.Adjust
+  ( adjustEpochRewards
+  ) where
+
+import           Cardano.Prelude hiding (from, on)
+
+import           Cardano.BM.Trace (Trace, logInfo)
+
+import qualified Cardano.Db as Db
+
+import           Cardano.Sync.Util
+
+import           Cardano.Slotting.Slot (EpochNo (..))
+
+import           Control.Monad.Trans.Control (MonadBaseControl)
+
+import           Database.Esqueleto.Legacy (InnerJoin (..), SqlExpr, Value (..), ValueList, delete,
+                   from, in_, not_, on, select, subSelectList, unValue, val, valList, where_, (==.),
+                   (^.))
+
+import           Database.Persist.Sql (SqlBackend)
+
+
+-- This is a hack/workaround for an issue around the `Reward` table. The problem is as
+-- follows:
+--
+-- * Rewards for epoch `N` are made available by ledger state towards the end of epoch `N + 1`
+--   at which time they are inserted into the database in chunks of say 1000 so interleaved
+--   between the insertion of regular block data.
+-- * If the stake address for the reward is de-registered after they are extracted from the
+--   ledger state but before the end of epoch `N + 1` (and not re-registered) then they should
+--   have been orphaned *instead* of being added to the `Reward` table.
+--
+--   To fix this, we call this function at the start of the epoch `N + 2`, find all the stake
+--   addresses that were de-registered (and not re-registered) in epoch `N - 1` and delete any
+--   `Reward` table entries destined for that stake address.
+
+adjustEpochRewards
+    :: (MonadBaseControl IO m, MonadIO m)
+    => Trace IO Text -> EpochNo
+    -> ReaderT SqlBackend m ()
+adjustEpochRewards tracer epochNo = do
+  liftIO . logInfo tracer $ "adjustEpochRewards: epoch " <> textShow (unEpochNo epochNo)
+  addrs <- queryOrphanedAddrs (epochNo - 1)
+  liftIO . logInfo tracer $ "adjustEpochiRewards: " <> textShow addrs
+  deleteOrphanedRewards addrs
+
+-- ------------------------------------------------------------------------------------------------
+
+deleteOrphanedRewards :: MonadIO m => [Db.StakeAddressId] -> ReaderT SqlBackend m ()
+deleteOrphanedRewards xs =
+  delete . from $ \ rwd ->
+    where_ (rwd ^. Db.RewardAddrId `in_` valList xs)
+
+-- Find all stake addresses that have been de-registered in the specified epoch and not
+-- re-registered in the same epoch.
+queryOrphanedAddrs :: MonadIO m => EpochNo -> ReaderT SqlBackend m [Db.StakeAddressId]
+queryOrphanedAddrs (EpochNo epochNo) = do
+    res <- select . from $ \ (sa `InnerJoin` dereg) -> do
+               on (sa ^. Db.StakeAddressId ==. dereg ^. Db.StakeDeregistrationAddrId)
+               where_ (dereg ^. Db.StakeDeregistrationEpochNo ==. val epochNo)
+               where_ (not_ $ sa ^. Db.StakeAddressId `in_` reregistered)
+               pure (sa ^. Db.StakeAddressId)
+    pure $ map unValue res
+  where
+    reregistered :: SqlExpr (ValueList Db.StakeAddressId)
+    reregistered =
+      subSelectList . from $ \ reg -> do
+        where_ (reg ^. Db.StakeRegistrationEpochNo ==. val epochNo)
+        pure (reg ^. Db.StakeRegistrationAddrId)
+

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

@@ -308,7 +308,7 @@ insertDelegCert
     -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
 insertDelegCert tracer network txId idx epochNo slotNo dCert =
   case dCert of
-    Shelley.RegKey cred -> insertStakeRegistration tracer txId idx $ Generic.annotateStakingCred network cred
+    Shelley.RegKey cred -> insertStakeRegistration tracer epochNo txId idx $ Generic.annotateStakingCred network cred
     Shelley.DeRegKey cred -> insertStakeDeregistration tracer network epochNo txId idx cred
     Shelley.Delegate (Shelley.Delegation cred poolkh) -> insertDelegation tracer network txId idx epochNo slotNo cred poolkh
 
@@ -463,28 +463,29 @@ insertPoolOwner network poolHashId txId skh = do
 
 insertStakeRegistration
     :: (MonadBaseControl IO m, MonadIO m)
-    => Trace IO Text -> DB.TxId -> Word16 -> Shelley.RewardAcnt StandardCrypto
+    => Trace IO Text -> EpochNo -> DB.TxId -> Word16 -> Shelley.RewardAcnt StandardCrypto
     -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
-insertStakeRegistration _tracer txId idx rewardAccount = do
+insertStakeRegistration _tracer epochNo txId idx rewardAccount = do
   saId <- lift $ insertStakeAddress txId rewardAccount
   void . lift . DB.insertStakeRegistration $
     DB.StakeRegistration
       { DB.stakeRegistrationAddrId = saId
       , DB.stakeRegistrationCertIndex = idx
+      , DB.stakeRegistrationEpochNo = unEpochNo epochNo
       , DB.stakeRegistrationTxId = txId
       }
 
 insertStakeDeregistration
     :: (MonadBaseControl IO m, MonadIO m)
     => Trace IO Text -> Ledger.Network -> EpochNo -> DB.TxId -> Word16 -> Ledger.StakeCredential StandardCrypto
     -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
-insertStakeDeregistration _tracer network (EpochNo epochNo) txId idx cred = do
+insertStakeDeregistration _tracer network epochNo txId idx cred = do
   scId <- liftLookupFail "insertStakeDeregistration" $ queryStakeAddress (Generic.stakingCredHash network cred)
   void . lift . DB.insertStakeDeregistration $
     DB.StakeDeregistration
       { DB.stakeDeregistrationAddrId = scId
       , DB.stakeDeregistrationCertIndex = idx
-      , DB.stakeDeregistrationEpochNo = epochNo
+      , DB.stakeDeregistrationEpochNo = unEpochNo epochNo
       , DB.stakeDeregistrationTxId = txId
       }
 

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

@@ -14,17 +14,18 @@ import           Cardano.BM.Trace (Trace, logDebug, logInfo)
 
 import qualified Cardano.Db as DB
 
+import           Cardano.DbSync.Era
+
 import           Cardano.DbSync.Era.Byron.Insert (insertByronBlock)
 import           Cardano.DbSync.Era.Cardano.Insert (insertEpochSyncTime)
+import           Cardano.DbSync.Era.Shelley.Adjust (adjustEpochRewards)
 import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
 import           Cardano.DbSync.Era.Shelley.Insert (insertShelleyBlock)
 import           Cardano.DbSync.Era.Shelley.Insert.Epoch
 import           Cardano.DbSync.Rollback (rollbackToPoint)
 
 import           Cardano.Slotting.Slot (EpochNo (..))
 
-import           Cardano.DbSync.Era
-
 import           Cardano.Sync.Api
 import           Cardano.Sync.Error
 import           Cardano.Sync.LedgerState
@@ -137,6 +138,7 @@ handleLedgerEvents tracer lenv point =
       case ev of
         LedgerNewEpoch en ss -> do
           lift $ insertEpochSyncTime en ss (leEpochSyncTime lenv)
+          lift $ adjustEpochRewards tracer en
           liftIO . logInfo tracer $ "Starting epoch " <> textShow (unEpochNo en)
         LedgerStartAtEpoch en ->
           -- This is different from the previous case in that the db-sync started