db-sync: Include the InstantaneousRewards to the Rewards table

InstantaneousRewards are payments from either the treasury or the reserves
to a stake address. Since these are not associated with a stake pool so that
the `PoolId` field of the `Reward` table has now become NULLable.

Closes: #695

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

@@ -159,14 +159,13 @@ insertRewards epoch icache rewardsChunk = do
         -> ExceptT SyncNodeError (ReaderT SqlBackend m) [DB.Reward]
     mkRewards (saddr, rset) = do
       saId <- hoistEither $ lookupStakeAddrIdPair "insertRewards StakePool" saddr icache
-      forM (Set.toList rset) $ \ rwd -> do
-        poolId <- hoistEither $ lookupPoolIdPair "insertRewards StakePool" (Generic.rewardPool rwd) icache
+      forM (Set.toList rset) $ \ rwd ->
         pure $ DB.Reward
                   { DB.rewardAddrId = saId
                   , DB.rewardType = DB.showRewardSource (Generic.rewardSource rwd)
                   , DB.rewardAmount = Generic.coinToDbLovelace (Generic.rewardAmount rwd)
                   , DB.rewardEpochNo = unEpochNo epoch
-                  , DB.rewardPoolId = poolId
+                  , DB.rewardPoolId = lookupPoolIdPairMaybe (Generic.rewardPool rwd) icache
                   }
 
 insertOrphanedRewards
@@ -184,13 +183,12 @@ insertOrphanedRewards epoch icache orphanedRewardsChunk = do
     mkOrphanedReward (saddr, rset) = do
       saId <- hoistEither $ lookupStakeAddrIdPair "insertOrphanedRewards StakeCred" saddr icache
       forM (Set.toList rset) $ \ rwd -> do
-        poolId <- hoistEither $ lookupPoolIdPair "insertOrphanedRewards StakePool" (Generic.rewardPool rwd) icache
         pure $ DB.OrphanedReward
                   { DB.orphanedRewardAddrId = saId
                   , DB.orphanedRewardType = DB.showRewardSource (Generic.rewardSource rwd)
                   , DB.orphanedRewardAmount = Generic.coinToDbLovelace (Generic.rewardAmount rwd)
                   , DB.orphanedRewardEpochNo = unEpochNo epoch
-                  , DB.orphanedRewardPoolId = poolId
+                  , DB.orphanedRewardPoolId = lookupPoolIdPairMaybe (Generic.rewardPool rwd) icache
                   }
 
 -- -------------------------------------------------------------------------------------------------
@@ -207,6 +205,12 @@ lookupStakeAddrIdPair msg scred lcache =
         mconcat [ "lookupStakeAddrIdPair: ", msg, renderByteArray (Generic.unStakeCred scred) ]
 
 
+lookupPoolIdPairMaybe
+    :: Maybe PoolKeyHash -> IndexCache
+    -> Maybe DB.PoolHashId
+lookupPoolIdPairMaybe mpkh lcache =
+  maybe Nothing (\pkh -> Map.lookup pkh $ icPoolCache lcache) mpkh
+
 lookupPoolIdPair
     :: Text -> PoolKeyHash -> IndexCache
     -> Either SyncNodeError DB.PoolHashId

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

@@ -279,8 +279,9 @@ share
     type                Text                sqltype=rewardtype
     amount              DbLovelace          sqltype=lovelace
     epochNo             Word64
-    poolId              PoolHashId          OnDeleteCascade
-    UniqueReward        epochNo addrId poolId
+    poolId              PoolHashId Maybe    OnDeleteCascade
+    -- A NULL in the poolId field will not be considered unique.
+    UniqueReward        epochNo addrId poolId !force
 
   -- Orphaned rewards happen when a stake address earns rewards, but the stake address is
   -- deregistered before the rewards are distributed.
@@ -290,8 +291,9 @@ share
     type                Text                sqltype=rewardtype
     amount              DbLovelace          sqltype=lovelace
     epochNo             Word64
-    poolId              PoolHashId          OnDeleteCascade
-    UniqueOrphaned      epochNo addrId poolId
+    poolId              PoolHashId Maybe    OnDeleteCascade
+    -- A NULL in the poolId field will not be considered unique.
+    UniqueOrphaned      epochNo addrId poolId !force
 
   Withdrawal
     addrId              StakeAddressId      OnDeleteCascade
@@ -661,7 +663,8 @@ schemaDocs =
       RewardType # "The source of the rewards; pool `member` vs pool `owner`."
       RewardAmount # "The reward amount (in Lovelace)."
       RewardEpochNo # "The epoch in which the reward was earned."
-      RewardPoolId # "The PoolHash table index for the pool the stake address was delegated to when the reward is earned."
+      RewardPoolId # "The PoolHash table index for the pool the stake address was delegated to when\
+            \ the reward is earned. Will be NULL for payments from the treasury or the reserves."
 
     OrphanedReward --^ do
       "A table for rewards earned by staking, but are orphaned. Rewards are orphaned when the stake address\
@@ -670,7 +673,9 @@ schemaDocs =
       OrphanedRewardType # "The source of the rewards; pool `member` vs pool `owner`."
       OrphanedRewardAmount # "The reward amount (in Lovelace)."
       OrphanedRewardEpochNo # "The epoch in which the reward was earned."
-      OrphanedRewardPoolId # "The PoolHash table index for the pool the stake address was delegated to when the reward is earned."
+      OrphanedRewardPoolId # "The PoolHash table index for the pool the stake address was delegated to when\
+            \ the reward is earned. Will be NULL for payments from the treasury or the reserves."
+
 
     Withdrawal --^ do
       "A table for withdrawals from a reward account."

cardano-sync/src/Cardano/Sync/Era/Shelley/Generic/Rewards.hs

@@ -25,6 +25,7 @@ import           Data.Bifunctor (bimap)
 import           Data.Coerce (coerce)
 import           Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
+import           Data.Maybe (mapMaybe)
 import           Data.Set (Set)
 import qualified Data.Set as Set
 
@@ -39,7 +40,7 @@ import qualified Shelley.Spec.Ledger.Rewards as Shelley
 
 data Reward = Reward
   { rewardSource :: !RewardSource
-  , rewardPool :: !(Ledger.KeyHash 'Ledger.StakePool StandardCrypto)
+  , rewardPool :: !(Maybe (Ledger.KeyHash 'Ledger.StakePool StandardCrypto))
   , rewardAmount :: !Coin
   } deriving (Eq, Ord, Show)
 
@@ -63,8 +64,9 @@ epochRewards nw epoch lstate =
 
 rewardsPoolHashKeys :: Rewards -> Set PoolKeyHash
 rewardsPoolHashKeys rwds =
-  Set.unions . map (Set.map rewardPool) $
-    Map.elems (rwdRewards rwds) ++ Map.elems (rwdOrphaned rwds)
+  Set.fromList . mapMaybe rewardPool
+    $ concatMap Set.toList (Map.elems $ rwdRewards rwds)
+    ++ concatMap Set.toList (Map.elems $ rwdOrphaned rwds)
 
 rewardsStakeCreds :: Rewards -> Set StakeCred
 rewardsStakeCreds rwds =
@@ -93,7 +95,9 @@ genericRewards network epoch lstate =
     completeRewardUpdate x =
       case x of
         Shelley.Pulsing {} -> Nothing -- Should never happen.
-        Shelley.Complete ru -> Just $ convertRewardMap (Shelley.rs ru)
+        Shelley.Complete ru -> Just $ Map.unionWith mappend
+                                        (convertRewardMap $ Shelley.rs ru)
+                                        (getInstantaneousRewards network lstate)
 
     validRewardAddress :: StakeCred -> Set Reward -> Bool
     validRewardAddress addr _value = Set.member addr rewardAccounts
@@ -115,10 +119,30 @@ genericRewards network epoch lstate =
         { rewardSource = rewardTypeToSource $ Shelley.rewardType sr
         , rewardAmount = Shelley.rewardAmount sr
         , -- Coerce is safe here because we are coercing away an un-needed phantom type parameter (era).
-          rewardPool = coerce $ Shelley.rewardPool sr
+          rewardPool = Just $ coerce (Shelley.rewardPool sr)
         }
 
 
 mapBimap :: Ord k2 => (k1 -> k2) -> (a1 -> a2) -> Map k1 a1 -> Map k2 a2
 mapBimap fk fa = Map.fromAscList . map (bimap fk fa) . Map.toAscList
 
+
+getInstantaneousRewards :: forall era. Ledger.Network -> LedgerState (ShelleyBlock era) -> Map StakeCred (Set Reward)
+getInstantaneousRewards network lstate =
+    Map.unionWith mappend
+        (mapBimap (toStakeCred network) (convert RwdReserves) $ Shelley.iRReserves instRwds)
+        (mapBimap (toStakeCred network) (convert RwdTreasury) $ Shelley.iRTreasury instRwds)
+  where
+    convert :: RewardSource -> Coin -> Set Reward
+    convert rs coin =
+      Set.singleton
+        Reward
+          { rewardSource = rs
+          , rewardAmount = coin
+          , rewardPool = Nothing
+          }
+
+    instRwds :: Shelley.InstantaneousRewards (Crypto era)
+    instRwds =
+      Shelley._irwd . Shelley._dstate . Shelley._delegationState
+        . Shelley.esLState . Shelley.nesEs $ Consensus.shelleyLedgerState lstate

cardano-sync/src/Cardano/Sync/LedgerState.hs

@@ -350,7 +350,7 @@ generateEvents env oldEventState details cls pnt = do
       case lesLastRewardsEpoch oldEventState of
         Nothing -> mkRewards
         Just oldRewardEpoch ->
-          if oldRewardEpoch < currentEpochNo
+          if sdEpochSlot details >= leStableEpochSlot env && oldRewardEpoch < currentEpochNo
             then mkRewards
             else Nothing
 

cardano-sync/src/Cardano/Sync/Types.hs

@@ -45,7 +45,7 @@ data BlockDetails = BlockDetails
 -- | Slot within an Epoch.
 newtype EpochSlot = EpochSlot
   { unEpochSlot :: Word64
-  } deriving (Eq, Show)
+  } deriving (Eq, Ord, Show)
 
 data FetchResult
     = ResultMetadata !PoolOfflineData

schema/migration-2-0012-20210722.sql

@@ -0,0 +1,20 @@
+-- Persistent generated migration.
+
+CREATE FUNCTION migrate() RETURNS void AS $$
+DECLARE
+  next_version int ;
+BEGIN
+  SELECT stage_two + 1 INTO next_version FROM schema_version ;
+  IF next_version = 12 THEN
+    EXECUTE 'ALTER TABLE "reward" ALTER COLUMN "pool_id" DROP NOT NULL' ;
+    EXECUTE 'ALTER TABLE "orphaned_reward" ALTER COLUMN "pool_id" DROP NOT NULL' ;
+    -- Hand written SQL statements can be added here.
+    UPDATE schema_version SET stage_two = next_version ;
+    RAISE NOTICE 'DB has been migrated to stage_two version %', next_version ;
+  END IF ;
+END ;
+$$ LANGUAGE plpgsql ;
+
+SELECT migrate() ;
+
+DROP FUNCTION migrate() ;