Leios: add LeiosTxCache, but don't rely on it at all yet

ouroboros-consensus-cardano/src/byron/Ouroboros/Consensus/Byron/Node.hs

@@ -145,13 +145,14 @@ byronBlockForging creds =
           tickedPBftState
     , forgeBlock = \ForgeBlockArgs{..} ->
         return $
-          forgeByronBlock
-            fbConfig
-            fbCurrentBlockNo
-            fbCurrentSlotNo
-            fbCurrentTickedLedgerState
-            fbRbTxs
-            fbIsLeader
+          flip (,) Nothing $
+            forgeByronBlock
+              fbConfig
+              fbCurrentBlockNo
+              fbCurrentSlotNo
+              fbCurrentTickedLedgerState
+              fbRbTxs
+              fbIsLeader
     , finalize = pure ()
     }
  where

ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Forge.hs

@@ -24,18 +24,16 @@ import qualified Cardano.Ledger.Shelley.API as SL (Block (..), extractTx)
 import Cardano.Prelude (nonEmpty)
 import qualified Cardano.Protocol.TPraos.BHeader as SL
 import Control.Exception
-import Control.Monad (void, when)
+import Control.Monad (when)
 import Control.Tracer (traceWith)
 import Data.ByteString.Short (fromShort)
 import Data.Maybe (isJust)
 import Data.Maybe.Strict (StrictMaybe (..), maybeToStrictMaybe)
 import qualified Data.Sequence.Strict as Seq
 import qualified Data.Typeable as Typeable
-import LeiosDemoDb (LeiosDbConnection (..))
 import LeiosDemoTypes
   ( EbAnnouncement (..)
   , ForgedLeiosEb (..)
-  , LeiosPoint (..)
   , RbHash (..)
   , TraceLeiosKernel (..)
   , forgeLeiosEb
@@ -73,7 +71,7 @@ forgeShelleyBlock ::
   HotKey (ProtoCrypto proto) m ->
   CanBeLeader proto ->
   ForgeBlockArgs m (ShelleyBlock proto era) ->
-  m (ShelleyBlock proto era)
+  m (ShelleyBlock proto era, Maybe ForgedLeiosEb)
 forgeShelleyBlock hotKey cbl ForgeBlockArgs{..} = do
   -- Forge an RB and attempt to announce an EB and/or certify a previously announced one:
   --
@@ -85,7 +83,7 @@ forgeShelleyBlock hotKey cbl ForgeBlockArgs{..} = do
   --    been rebased onto the post-certificate ledger state.
   mayEbAnn <-
     case Typeable.eqT @era @DijkstraEra of
-      Just Refl -> mkAndStoreEb
+      Just Refl -> mkEb
       Nothing -> pure Nothing
   let rbBody = mkBody fbMayLeiosCert
       actualRbBodySize = SL.blockBodySize protocolVersion rbBody
@@ -122,9 +120,10 @@ forgeShelleyBlock hotKey cbl ForgeBlockArgs{..} = do
         traceWith fbLeiosTracer $
           TraceLeiosCertifiedAndAnnounced{atSlot = fbCurrentSlotNo, rbHash = MkRbHash announcingRbHashBytes}
     Nothing -> pure ()
-  return $
-    assert (verifyBlockIntegrity (configSlotsPerKESPeriod $ configConsensus fbConfig) blk) $
-      blk
+  return
+    ( assert (verifyBlockIntegrity (configSlotsPerKESPeriod $ configConsensus fbConfig) blk) blk
+    , fst <$> mayEbAnn
+    )
  where
   protocolVersion = shelleyProtocolVersion $ configBlock fbConfig
 
@@ -154,16 +153,12 @@ forgeShelleyBlock hotKey cbl ForgeBlockArgs{..} = do
       . getTipHash
       $ fbCurrentTickedLedgerState
 
-  -- Produce an EB from fbEbTxs, store it into fbLeiosDb, and return the
-  -- announcement to embed in the header. An honest forger only emits an
-  -- EB when it has txs to put in it; empty mempool ⇒ no EB ⇒ no
-  -- announcement (matches the original prototype). Persists the EB into
-  -- 'LeiosDb' before returning, so the closure is available locally
-  -- before the header carrying the announcement is finalised and
-  -- diffused — a peer that fetches our header will be able to pull the
-  -- closure from us in the same round-trip.
-  mkAndStoreEb :: m (Maybe (ForgedLeiosEb, EbAnnouncement))
-  mkAndStoreEb = case nonEmpty (fmap extractTx fbEbTxs) of
+  -- Produce an EB from 'fbEbTxs' and return it together with the announcement
+  -- to embed in the header. An honest forger only emits an EB when it has txs
+  -- to put in it; empty mempool ⇒ no EB ⇒ no announcement (matches the original
+  -- prototype).
+  mkEb :: m (Maybe (ForgedLeiosEb, EbAnnouncement))
+  mkEb = case nonEmpty (fmap extractTx fbEbTxs) of
     Nothing -> pure Nothing
     Just ebTxs -> do
       let forgedEb = forgeLeiosEb fbCurrentSlotNo ebTxs
@@ -174,24 +169,11 @@ forgeShelleyBlock hotKey cbl ForgeBlockArgs{..} = do
               { ebAnnouncementHash = ebHash
               , ebAnnouncementSize = ebSize
               }
-          ebPoint =
-            MkLeiosPoint
-              { pointSlotNo = fbCurrentSlotNo
-              , pointEbHash = ebHash
-              }
       traceWith fbLeiosTracer $
         TraceLeiosBlockForged
           { slot = fbCurrentSlotNo
           , eb = forgedEb.body
           , ebMeasure = ByteSize32 ebSize
           , mempoolRestMeasure = ByteSize32 0
           }
-      leiosDbInsertEbPoint fbLeiosDb ebPoint ebSize
-      void $ leiosDbInsertEbBody fbLeiosDb ebPoint forgedEb.body
-      void $ leiosDbInsertTxs fbLeiosDb forgedEb.txClosure
-      traceWith fbLeiosTracer $
-        TraceLeiosBlockStored
-          { slot = fbCurrentSlotNo
-          , eb = forgedEb.body
-          }
       pure (Just (forgedEb, ebAnn))

ouroboros-consensus-cardano/src/unstable-byron-testlib/Ouroboros/Consensus/ByronDual/Node.hs

@@ -67,13 +67,14 @@ dualByronBlockForging creds =
     , checkCanForge = checkCanForge . dualTopLevelConfigMain
     , forgeBlock = \ForgeBlockArgs{..} ->
         return $
-          forgeDualByronBlock
-            fbConfig
-            fbCurrentBlockNo
-            fbCurrentSlotNo
-            fbCurrentTickedLedgerState
-            fbRbTxs
-            fbIsLeader
+          flip (,) Nothing $
+            forgeDualByronBlock
+              fbConfig
+              fbCurrentBlockNo
+              fbCurrentSlotNo
+              fbCurrentTickedLedgerState
+              fbRbTxs
+              fbIsLeader
     , finalize = return ()
     }
  where

ouroboros-consensus-cardano/src/unstable-cardano-tools/Cardano/Tools/DBAnalyser/Run.hs

@@ -187,6 +187,7 @@ analyse dbaConfig args =
             shfs
             flavargs
             leiosDbHandle
+            (\_ -> pure ()) -- no LeiosTxCache in this tool
             $ ChainDB.defaultArgs
         -- Set @k=1@ to reduce the memory usage of the LedgerDB. We only ever
         -- go forward so we don't need to account for rollbacks.

ouroboros-consensus-cardano/src/unstable-cardano-tools/Cardano/Tools/DBSynthesizer/Forging.hs

@@ -111,7 +111,7 @@ runForge ::
   GenTxs blk ->
   LeiosDbConnection IO ->
   IO ForgeResult
-runForge epochSize_ nextSlot opts chainDB blockForging cfg genTxs leiosDb = do
+runForge epochSize_ nextSlot opts chainDB blockForging cfg genTxs _leiosDb = do
   putStrLn $ "--> epoch size: " ++ show epochSize_
   putStrLn $ "--> will process until: " ++ show opts
   -- Synthetic forging doesn't gather votes; supply a vote state with
@@ -233,7 +233,9 @@ runForge epochSize_ nextSlot opts chainDB blockForging cfg genTxs leiosDb = do
             tickedLedgerState
 
     -- Actually produce the block
-    newBlock <-
+    --
+    -- TODO the block may be accompanied by an EB, which is entirely ignored
+    (newBlock, _mForgedEb) <-
       lift $
         Block.forgeBlock
           blockForging'
@@ -246,7 +248,6 @@ runForge epochSize_ nextSlot opts chainDB blockForging cfg genTxs leiosDb = do
             , fbEbTxs = []
             , fbIsLeader = proof
             , fbChainDepState = Nothing
-            , fbLeiosDb = leiosDb
             , fbLeiosTracer = Trace.nullTracer
             , fbLeiosVoteState = leiosVoteState
             , fbMayLeiosCert = Nothing

ouroboros-consensus-cardano/src/unstable-cardano-tools/Cardano/Tools/DBSynthesizer/Run.hs

@@ -170,6 +170,7 @@ synthesize genTxs DBSynthesizerConfig{confOptions, confShelleyGenesis, confDbDir
           (Node.stdMkChainDbHasFS confDbDir)
           flavargs
           leiosDbHandle
+          (\_ -> pure ()) -- no LeiosTxCache in this tool
           $ ChainDB.defaultArgs
 
     mbfs <- mkForgers nullTracer

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs

@@ -136,6 +136,10 @@ import qualified Network.Mux as Mux
 import Network.TypedProtocol.Codec
 import Network.TypedProtocol.Peer (Peer (Effect))
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types
+  ( diffRelTime
+  , systemTimeCurrent
+  )
 import Ouroboros.Consensus.Config (DiffusionPipeliningSupport (..))
 import Ouroboros.Consensus.HeaderValidation (HeaderWithTime)
 import Ouroboros.Consensus.Ledger.SupportsMempool
@@ -157,7 +161,7 @@ import Ouroboros.Consensus.Storage.LedgerDB.Forker
   ( ResolveLeiosBlock
   )
 import Ouroboros.Consensus.Storage.Serialisation (SerialisedHeader)
-import Ouroboros.Consensus.Util (ShowProxy)
+import Ouroboros.Consensus.Util (ShowProxy, whenJust)
 import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Consensus.Util.Orphans ()
 import Ouroboros.Network.Block
@@ -351,6 +355,7 @@ mkHandlers ::
   ( IOLike m
   , MonadTime m
   , MonadTimer m
+  , ConvertRawHash blk
   , LedgerSupportsMempool blk
   , HasTxId (GenTx blk)
   , LedgerSupportsProtocol blk
@@ -397,8 +402,29 @@ mkHandlers
               , CsClient.tracer =
                   contramap (TraceLabelPeer peer) (Node.chainSyncClientTracer tracers)
               , CsClient.getDiffusionPipeliningSupport = getDiffusionPipeliningSupport
-              , CsClient.leiosCertRbCallback =
-                  Leios.leiosCertRbCallback (getLeiosOutstanding, getLeiosReady) peerVars
+              , CsClient.leiosMsgRollForwardCallback = \hdr hdrSlotTime cds -> do
+                  Leios.checkMsgRollForwardForLeiosOffers
+                    (getLeiosOutstanding, getLeiosReady)
+                    peerVars
+                    hdr
+                    cds
+                  -- Feed any EB this header announces into the central
+                  -- announcement state (relay + dedup + txCache), central-only:
+                  -- a roll-forward is not this peer announcing over LeiosNotify,
+                  -- so no PeerState is touched. Date it from the header slot's
+                  -- onset (its ChainSync arrival latency).
+                  whenJust (Leios.mkAnnouncingHeader hdr) $ \ancHdr -> do
+                    now <- systemTimeCurrent systemTime
+                    Leios.processAnnouncementCentrally
+                      (Node.leiosKernelTracer tracers)
+                      getLeiosCentralState
+                      (getLeiosOutstanding, getLeiosReady)
+                      getLeiosTxCache
+                      (Just peer)
+                      Leios.ReceivedViaChainSync
+                      Announcements.DoRelay
+                      (Just (diffRelTime now hdrSlotTime))
+                      ancHdr
               }
             dynEnv
       , hChainSyncServer = \peer _version ->
@@ -493,23 +519,20 @@ mkHandlers
                                 (Leios.ancHeader ancH)
                           )
                           -- central part of the processing
-                          ( \ancHdr (shouldRelay, age, anc'@(p, _sz)) -> do
+                          ( \ancHdr (shouldRelay, age, (p, _sz)) -> do
                               traceWith tracer $
                                 MkTraceLeiosPeer $
                                   "MsgLeiosBlockAnnouncement new: " <> Leios.prettyLeiosPoint p
-                              MVar.modifyMVar_ getLeiosCentralState $ \cst ->
-                                -- TODO OK to hold this the whole time we're writing to the LeiosNotify queues (NB those enqeues never block)?
-                                Announcements.onAnnouncementCentral
-                                  (contramap Leios.traceNewAnnouncement kernelTracer)
-                                  Leios.ancElId
-                                  ( \_elSt ->
-                                      Leios.recordAnnouncedEb (getLeiosOutstanding, getLeiosReady) anc'
-                                  )
-                                  cst
-                                  (Just peer)
-                                  shouldRelay
-                                  (Just age)
-                                  ancHdr
+                              Leios.processAnnouncementCentrally
+                                kernelTracer
+                                getLeiosCentralState
+                                (getLeiosOutstanding, getLeiosReady)
+                                getLeiosTxCache
+                                (Just peer)
+                                Leios.ReceivedViaLeiosNotify
+                                shouldRelay
+                                (Just age)
+                                ancHdr
                           )
                           peerSt0
                           anc
@@ -666,6 +689,7 @@ mkHandlers
                   (leiosPeerTracer peer)
                   ((== Terminate) <$> controlMessageSTM)
                   (getLeiosOutstanding, getLeiosReady)
+                  getLeiosTxCache
                   leiosConn
                   (Leios.MkPeerId peer)
                   reqVar
@@ -684,6 +708,7 @@ mkHandlers
       , getLeiosOutstanding
       , getLeiosReady
       , getLeiosCentralState
+      , getLeiosTxCache
       } = nodeKernel
 
     leiosPeerTracer peer = TraceLabelPeer peer `contramap` Node.leiosPeerTracer tracers