simulation: validate IBs once they are needed for ledger state.

- also do not cleanup certified EBs for Full leios, as they might be
referenced later.

Author: Saizan

simulation/src/LeiosProtocol/Short.hs

@@ -127,71 +127,60 @@ data LeiosConfig = forall p. IsPipeline p => LeiosConfig
 data SomeStage = forall p. IsPipeline p => SomeStage (SingPipeline p) (Stage p)
 
 convertConfig :: OnDisk.Config -> LeiosConfig
-convertConfig disk = checkAssertions
-  $ ( if disk.treatBlocksAsFull
-        then delaysAndSizesAsFull
-        else (\x -> x)
-    )
-  $ case voting of
-    SomeStage pipeline voteSendStage -> do
-      let sliceLength = fromIntegral disk.leiosStageLengthSlots
-      LeiosConfig
-        { praos
-        , pipeline
-        , voteSendStage
-        , sliceLength
-        , inputBlockFrequencyPerSlot = disk.ibGenerationProbability
-        , endorseBlockFrequencyPerStage = disk.ebGenerationProbability
-        , maxEndorseBlockAgeSlots = fromIntegral disk.ebMaxAgeSlots
-        , maxEndorseBlockAgeForRelaySlots = fromIntegral disk.ebMaxAgeForRelaySlots
-        , cleanupPolicies = disk.cleanupPolicies
-        , variant = disk.leiosVariant
-        , headerDiffusionTime = realToFrac $ durationMsToDiffTime disk.leiosHeaderDiffusionTimeMs
-        , pipelinesToReferenceFromEB =
-            if disk.leiosVariant == Full
-              then
-                ceiling ((3 * disk.praosChainQuality) / fromIntegral sliceLength) - 2
-              else 0
-        , activeVotingStageLength = fromIntegral disk.leiosStageActiveVotingSlots
-        , votingFrequencyPerStage = disk.voteGenerationProbability
-        , votesForCertificate = fromIntegral disk.voteThreshold
-        , sizes
-        , delays
-        , ibDiffusion =
-            RelayDiffusionConfig
-              { strategy = disk.ibDiffusionStrategy
-              , maxWindowSize = disk.ibDiffusionMaxWindowSize
-              , maxHeadersToRequest = disk.ibDiffusionMaxHeadersToRequest
-              , maxBodiesToRequest = disk.ibDiffusionMaxBodiesToRequest
-              }
-        , ebDiffusion =
-            RelayDiffusionConfig
-              { strategy = disk.ebDiffusionStrategy
-              , maxWindowSize = disk.ebDiffusionMaxWindowSize
-              , maxHeadersToRequest = disk.ebDiffusionMaxHeadersToRequest
-              , maxBodiesToRequest = disk.ebDiffusionMaxBodiesToRequest
-              }
-        , voteDiffusion =
-            RelayDiffusionConfig
-              { strategy = disk.voteDiffusionStrategy
-              , maxWindowSize = disk.voteDiffusionMaxWindowSize
-              , maxHeadersToRequest = disk.voteDiffusionMaxHeadersToRequest
-              , maxBodiesToRequest = disk.voteDiffusionMaxBodiesToRequest
-              }
-        , relayStrategy = disk.relayStrategy
-        }
+convertConfig disk =
+  ( if disk.treatBlocksAsFull
+      then delaysAndSizesAsFull
+      else (\x -> x)
+  )
+    $ case voting of
+      SomeStage pipeline voteSendStage -> do
+        let sliceLength = fromIntegral disk.leiosStageLengthSlots
+        LeiosConfig
+          { praos
+          , pipeline
+          , voteSendStage
+          , sliceLength
+          , inputBlockFrequencyPerSlot = disk.ibGenerationProbability
+          , endorseBlockFrequencyPerStage = disk.ebGenerationProbability
+          , maxEndorseBlockAgeSlots = fromIntegral disk.ebMaxAgeSlots
+          , maxEndorseBlockAgeForRelaySlots = fromIntegral disk.ebMaxAgeForRelaySlots
+          , cleanupPolicies = disk.cleanupPolicies
+          , variant = disk.leiosVariant
+          , headerDiffusionTime = realToFrac $ durationMsToDiffTime disk.leiosHeaderDiffusionTimeMs
+          , pipelinesToReferenceFromEB =
+              if disk.leiosVariant == Full
+                then
+                  ceiling ((3 * disk.praosChainQuality) / fromIntegral sliceLength) - 2
+                else 0
+          , activeVotingStageLength = fromIntegral disk.leiosStageActiveVotingSlots
+          , votingFrequencyPerStage = disk.voteGenerationProbability
+          , votesForCertificate = fromIntegral disk.voteThreshold
+          , sizes
+          , delays
+          , ibDiffusion =
+              RelayDiffusionConfig
+                { strategy = disk.ibDiffusionStrategy
+                , maxWindowSize = disk.ibDiffusionMaxWindowSize
+                , maxHeadersToRequest = disk.ibDiffusionMaxHeadersToRequest
+                , maxBodiesToRequest = disk.ibDiffusionMaxBodiesToRequest
+                }
+          , ebDiffusion =
+              RelayDiffusionConfig
+                { strategy = disk.ebDiffusionStrategy
+                , maxWindowSize = disk.ebDiffusionMaxWindowSize
+                , maxHeadersToRequest = disk.ebDiffusionMaxHeadersToRequest
+                , maxBodiesToRequest = disk.ebDiffusionMaxBodiesToRequest
+                }
+          , voteDiffusion =
+              RelayDiffusionConfig
+                { strategy = disk.voteDiffusionStrategy
+                , maxWindowSize = disk.voteDiffusionMaxWindowSize
+                , maxHeadersToRequest = disk.voteDiffusionMaxHeadersToRequest
+                , maxBodiesToRequest = disk.voteDiffusionMaxBodiesToRequest
+                }
+          , relayStrategy = disk.relayStrategy
+          }
  where
-  checkAssertions (cfg :: LeiosConfig) =
-    if cfg.variant /= Full || (oldestEBToReference < cfg.maxEndorseBlockAgeSlots)
-      then cfg
-      else error $ "Parameter `eb-max-age-slots` should be greater than " ++ show oldestEBToReference ++ " given the chosen praos-chain-quality and leios-stage-length-slots."
-   where
-    oldestEBToReference =
-      ( ceiling ((3 * disk.praosChainQuality) / fromIntegral cfg.sliceLength)
-          + 1 {-Endorse-}
-          + 1 {- Vote (Send) -}
-      )
-        * cfg.sliceLength
   forEach n xs = n * fromIntegral (length @[] xs)
   forEachKey n m = n * fromIntegral (Map.size m)
   durationMsToDiffTime (DurationMs d) = secondsToDiffTime $ d / 1000
@@ -508,10 +497,6 @@ proposeRange cfg@LeiosConfig{pipeline = (_ :: SingPipeline p)} p =
 pipelineRange :: LeiosConfig -> PipelineNo -> (SlotNo, SlotNo)
 pipelineRange cfg p = (fst $ proposeRange cfg p, lastVoteRecv cfg p)
 
-lastUnadoptedEB :: LeiosConfig -> PipelineNo -> SlotNo
-lastUnadoptedEB leios@LeiosConfig{pipeline = (_ :: SingPipeline p), maxEndorseBlockAgeSlots} pipelineNo =
-  lastVoteRecv leios pipelineNo + toEnum maxEndorseBlockAgeSlots
-
 endorseBlockPipeline :: LeiosConfig -> EndorseBlock -> PipelineNo
 endorseBlockPipeline cfg@LeiosConfig{pipeline = _ :: SingPipeline p} eb = pipelineOf @p cfg Endorse eb.slot
 
@@ -712,22 +697,25 @@ endorseBlocksToReference cfg@LeiosConfig{variant = Full} pl EndorseBlocksSnapsho
  where
   result =
     [ (p, [eb | (eb, _, _) <- es])
-    | plRange <- maybeToList $ pipelinesToReferenceFromEB cfg.pipelinesToReferenceFromEB pl
+    | plRange <- maybeToList $ pipelinesToReferenceFromEB cfg pl
     , (p, es) <- certifiedEndorseBlocks plRange
     , or [checkDeliveryTime p t | (_, _, t) <- es]
     ]
 
-pipelinesToReferenceFromEB :: Int -> PipelineNo -> Maybe (PipelineNo, PipelineNo)
-pipelinesToReferenceFromEB n pl = do
-  predPl <- safePred pl
-  case fromEnum predPl - maxStagesAfterEndorse of
-    newestIx
-      | newestIx < 0 -> Nothing
-      | otherwise ->
-          Just
-            ( toEnum $ max 0 $ newestIx - (n - 1)
-            , toEnum newestIx
-            )
+pipelinesToReferenceFromEB :: LeiosConfig -> PipelineNo -> Maybe (PipelineNo, PipelineNo)
+pipelinesToReferenceFromEB cfg pl
+  | Full <- cfg.variant = do
+      let n = cfg.pipelinesToReferenceFromEB
+      predPl <- safePred pl
+      case fromEnum predPl - maxStagesAfterEndorse of
+        newestIx
+          | newestIx < 0 -> Nothing
+          | otherwise ->
+              Just
+                ( toEnum $ max 0 $ newestIx - (n - 1)
+                , toEnum newestIx
+                )
+  | otherwise = Nothing
  where
   maxStagesAfterEndorse = 2
   safePred x = do

simulation/src/LeiosProtocol/Short/Node.hs

@@ -139,6 +139,7 @@ data LeiosNodeState m = LeiosNodeState
   -- ^ waiting for ledger state of RB block to be validated.
   , ledgerStateVar :: !(TVar m (Map (HeaderHash RankingBlock) LedgerState))
   , ibsNeededForEBVar :: !(TVar m (Map EndorseBlockId (Set InputBlockId)))
+  , ibsValidationActionsVar :: !(TVar m (Map InputBlockId (STM m ())))
   , votesForEBVar :: !(TVar m (Map EndorseBlockId CertificateProgress))
   }
 
@@ -401,6 +402,7 @@ newLeiosNodeState cfg = do
   waitingForLedgerStateVar <- newTVarIO Map.empty
   taskQueue <- atomically $ newTaskMultiQueue cfg.processingQueueBound
   votesForEBVar <- newTVarIO Map.empty
+  ibsValidationActionsVar <- newTVarIO Map.empty
   return $ LeiosNodeState{..}
 
 leiosNode ::
@@ -495,7 +497,10 @@ leiosNode tracer cfg followers peers = do
 
   let blockGenerationThreads = [generator tracer cfg leiosState]
 
-  let computeLedgerStateThreads = [computeLedgerStateThread tracer cfg leiosState]
+  let computeLedgerStateThreads =
+        [ computeLedgerStateThread tracer cfg leiosState
+        -- , validateIBsOfCertifiedEBs tracer cfg leiosState
+        ]
 
   let pruningThreads =
         concat
@@ -507,6 +512,9 @@ leiosNode tracer cfg followers peers = do
             ]
           , [ pruneExpiredUnadoptedEBs tracer cfg leiosState
             | CleanupExpiredUnadoptedEb `isEnabledIn` cfg.leios.cleanupPolicies
+            , -- With Full a fresh EB might end up referencing all the way to Genesis.
+            -- TODO: could expire EBs not referenced by young enough EBs.
+            cfg.leios.variant /= Full
             ]
           , [ pruneExpiredIBs tracer cfg leiosState
             | CleanupExpiredIb `isEnabledIn` cfg.leios.cleanupPolicies
@@ -543,7 +551,7 @@ pruneExpiredIBs _tracer LeiosNodeConfig{leios, slotConfig} st = go (toEnum 0)
     --   traceWith tracer $! LeiosNodeEvent Pruned (EventIB ib)
     go (succ p)
 
--- rEB slots after the end of vote-receiving,
+-- rEB slots after the end of Endorse,
 -- prune EBs that became certified but were not adopted by an RB.
 pruneExpiredUnadoptedEBs ::
   forall m.
@@ -673,44 +681,107 @@ pruneExpiredVotes _tracer LeiosNodeConfig{leios = leios@LeiosConfig{pipeline = _
     --   traceWith tracer $! LeiosNodeEvent Pruned (EventVote $ snd vt)
     go (succ p)
 
+referencedEBs :: MonadSTM m => LeiosConfig -> LeiosNodeState m -> Set EndorseBlockId -> STM m [EndorseBlockId]
+referencedEBs cfg st ebIds0
+  | null ebIds0 = return []
+  | Short <- cfg.variant = pure $ Set.toList ebIds0
+  | otherwise = do
+      ebBuffer <- readTVar st.relayEBState.relayBufferVar
+      let
+        ebsReferenced :: Set EndorseBlockId -> Set EndorseBlockId -> [EndorseBlockId]
+        ebsReferenced !fetched ebIds
+          | null ebIds = []
+          | otherwise = do
+              let ebs =
+                    [ snd $ fromMaybe (error $ "EB missing:" ++ show ebId) $ RB.lookup ebBuffer ebId
+                    | ebId <- Set.toList ebIds
+                    ]
+              let fetched' = Set.union fetched ebIds
+              let refs =
+                    Set.fromList
+                      [ refId
+                      | eb <- ebs
+                      , refId <- eb.endorseBlocksEarlierPipeline
+                      , Set.notMember refId fetched'
+                      ]
+              map (.id) ebs ++ ebsReferenced fetched' refs
+      return $ ebsReferenced Set.empty ebIds0
+
 computeLedgerStateThread ::
   forall m.
   (MonadMVar m, MonadFork m, MonadAsync m, MonadSTM m, MonadTime m, MonadDelay m) =>
   Tracer m LeiosNodeEvent ->
   LeiosNodeConfig ->
   LeiosNodeState m ->
   m ()
-computeLedgerStateThread tracer _cfg st = forever $ do
+computeLedgerStateThread tracer cfg st = forever $ do
   readyLedgerState <- atomically $ do
     -- TODO: this will get more costly as the base chain grows,
     -- however it grows much more slowly than anything else.
     chain <- PraosNode.preferredChain st.praosState
     let rbsOnChain = Chain.toNewestFirst $ chain
-    let blocks = Map.fromList [(blockHash block, block) | block <- rbsOnChain]
-    when (Map.null blocks) retry
+    when (null rbsOnChain) retry
+    -- TODO: should we prune the ledger state to only cover RBs on the chain?
     ledgerState <- readTVar st.ledgerStateVar
-    let ledgerMissing = Map.elems $ blocks Map.\\ ledgerState
-    when (null ledgerMissing) retry
-    let ledgerEligible = flip filter ledgerMissing $ \blk ->
-          case blockPrevHash blk of
-            GenesisHash -> True
-            BlockHash prev -> prev `Map.member` ledgerState
-    when (null ledgerEligible) retry
-
-    ibsNeededForEB <- readTVar st.ibsNeededForEBVar
-    let readyLedgerState =
-          [ (blockHash rb, LedgerState)
-          | rb <- ledgerMissing
-          , flip all rb.blockBody.endorseBlocks $ \(ebId, _) ->
-              Map.lookup ebId ibsNeededForEB == Just Set.empty
-          ]
-    when (null readyLedgerState) retry
-    modifyTVar' st.ledgerStateVar (`Map.union` Map.fromList readyLedgerState)
-    return readyLedgerState
+    let oldestMissingLedgerState = go Nothing rbsOnChain
+         where
+          go acc [] = acc
+          go acc (x : xs)
+            | Map.member (blockHash x) ledgerState = acc
+            | otherwise = go (Just x) xs
+    ledgerEligible <- case oldestMissingLedgerState of
+      Nothing -> retry
+      Just block -> pure block
+
+    todo <- do
+      let doLedgerState = Left (blockHash ledgerEligible, LedgerState)
+      case (map fst $ ledgerEligible.blockBody.endorseBlocks) of
+        [] -> return $ doLedgerState
+        ebIds -> do
+          ibsNeededForEB <- readTVar st.ibsNeededForEBVar
+          ibsNeeded <- do
+            ebs <- referencedEBs cfg.leios st (Set.fromList ebIds)
+            return $ Set.unions <$> mapM (flip Map.lookup ibsNeededForEB) ebs
+          case ibsNeeded of
+            -- Some EB was missing ibsNeeded info
+            Nothing -> undefined
+            Just ibs
+              | Set.null ibs -> pure $ doLedgerState
+              | otherwise -> pure $ Right ibs
+
+    case todo of
+      Left readyLedgerState -> do
+        modifyTVar' st.ledgerStateVar (uncurry Map.insert readyLedgerState)
+        return [readyLedgerState]
+      Right ibsEligibleToValidate -> do
+        ibValActions <- readTVar st.ibsValidationActionsVar
+        let ibsReadyToValidate = Map.elems $ Map.restrictKeys ibValActions ibsEligibleToValidate
+        if null ibsReadyToValidate
+          then retry
+          else do
+            modifyTVar' st.ibsValidationActionsVar $ flip Map.withoutKeys ibsEligibleToValidate
+            sequence_ ibsReadyToValidate
+            return []
   for_ readyLedgerState $ \(rb, _) -> do
     traceWith tracer $! LeiosNodeEventLedgerState rb
   return ()
 
+-- TODO: Use or remove.
+--       Might be sensible to validate IBs as soon as we have a certified EB including them: the network managed to validate the IB, so a suitable ledger state is available.
+validateIBsOfCertifiedEBs :: MonadSTM m => Tracer m LeiosNodeEvent -> LeiosNodeConfig -> LeiosNodeState m -> m ()
+validateIBsOfCertifiedEBs _trace _cfg st = forever . atomically $ do
+  ibsNeeded <- readTVar st.ibsNeededForEBVar
+  ebs <- readTVar st.votesForEBVar
+  let certEBs = Set.fromList [eb | (eb, Certified{}) <- Map.toList ebs]
+  let ibsEligible = Set.unions $ Map.elems $ Map.restrictKeys ibsNeeded certEBs
+  when (null ibsEligible) retry
+  ibsValActions <- readTVar st.ibsValidationActionsVar
+  let ibsToValidate = Map.toList $ Map.restrictKeys ibsValActions ibsEligible
+  when (null ibsToValidate) $ retry
+  forM_ ibsToValidate $ \(ibId, m) -> do
+    modifyTVar' st.ibsValidationActionsVar $ Map.delete ibId
+    m
+
 adoptIB :: MonadSTM m => LeiosNodeState m -> InputBlock -> UTCTime -> STM m ()
 adoptIB leiosState ib deliveryTime = do
   let !ibSlot = ib.header.slot
@@ -787,12 +858,28 @@ dispatchValidation tracer cfg leiosState req =
           return []
     ValidateIBS ibs deliveryTime completion -> do
       -- NOTE: IBs with an RB reference have to wait for ledger state of that RB.
-      let waitingLedgerState =
-            [ (rbHash, [queue [valIB ib deliveryTime completion]])
-            | ib <- ibs
-            , BlockHash rbHash <- [(fst ib).rankingBlock]
-            ]
+      --       However, if they get referenced by the chain they should be validated anyway.
+      --       We use a map to store the validation logic, so we can force it happening in the latter case.
+      let lookupValAction ibId = do
+            ibValActions <- readTVar leiosState.ibsValidationActionsVar
+            case Map.lookup ibId ibValActions of
+              Just m -> do
+                modifyTVar' leiosState.ibsValidationActionsVar $
+                  Map.delete ibId
+                m
+              Nothing -> pure ()
+      let storeAction rbHash ib@(h, _) = do
+            modifyTVar' leiosState.ibsValidationActionsVar $
+              Map.insert h.id (queue [valIB ib deliveryTime completion])
+            return (rbHash, [lookupValAction $ (fst ib).id])
+      waitingLedgerState <-
+        sequence $
+          [ storeAction rbHash ib
+          | ib <- ibs
+          , BlockHash rbHash <- [(fst ib).rankingBlock]
+          ]
 
+      -- TODO: cancel the ones forced by computeLedgerState
       waitFor
         leiosState.waitingForLedgerStateVar
         waitingLedgerState