Rewrite the LeiosFetch decision logic

ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Block.hs

@@ -1628,6 +1628,14 @@ instance
         (leiosClosureTxKeySets inner)
     _ -> emptyLedgerTables
 
+  leiosTxHashOfGenTx tx = case tx of
+    GenTxDijkstra inner -> leiosTxHashOfGenTx inner
+    _ -> Nothing
+
+  leiosTxBytesOfGenTx tx = case tx of
+    GenTxDijkstra inner -> leiosTxBytesOfGenTx inner
+    _ -> Nothing
+
 -----
 
 -- | We don't want to add the ResolveLeiosBlock sin-bin to SingleEraBlock, so we

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

@@ -11,6 +11,7 @@
 
 module Ouroboros.Consensus.Shelley.Ledger.Leios () where
 
+import Cardano.Binary (serialize')
 import qualified Cardano.Crypto.Hash as Crypto (hashToBytesShort)
 import Cardano.Ledger.Api (Tx)
 import Cardano.Ledger.Binary (decCBOR, decodeFullAnnotator)
@@ -38,7 +39,9 @@ import LeiosDemoLogic.Announcements.ElBimap (ElId (MkElId))
 import LeiosDemoTypes
   ( EbAnnouncement (..)
   , LeiosPoint (..)
+  , LeiosTx (..)
   , RbHash (..)
+  , hashLeiosTx
   )
 import Lens.Micro ((.~), (^.))
 import Ouroboros.Consensus.Block (ChainHash (..), blockPrevHash, toRawHash)
@@ -107,6 +110,13 @@ instance
   (PraosCrypto c, ShelleyCompatible (Praos c) DijkstraEra) =>
   ResolveLeiosBlock (ShelleyBlock (Praos c) DijkstraEra)
   where
+  -- The on-wire bytes and 'TxHash' a forged EB records for each tx (see
+  -- 'forgeLeiosEb'): 'serialize'' the tx, and hash exactly those bytes. Matching
+  -- this encoding is what lets the mempool key its txs by the same 'TxHash' an EB
+  -- lists, so the body-arrival mempool pull can find them.
+  leiosTxBytesOfGenTx (ShelleyTx _ tx) = Just (serialize' tx)
+  leiosTxHashOfGenTx (ShelleyTx _ tx) = Just (hashLeiosTx (MkLeiosTx (serialize' tx)))
+
   resolveLeiosClosure leiosDb ebHash = do
     mAnnouncedEb <-
       leiosDbLookupEbClosure

ouroboros-consensus-cardano/test/cardano-test/Test/ThreadNet/Leios.hs

@@ -250,7 +250,7 @@ prop_leios seed =
     _ -> Nothing
 
   acquiredPoints = Set.fromList . flip mapMaybe leiosTraces $ \case
-    TraceLeiosBlockTxsAcquired point -> Just point
+    TraceLeiosBlockTxsAcquired point _age -> Just point
     _ -> Nothing
 
   -- An EB forged at slot @s@ is required to diffuse iff it has at least

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

@@ -71,6 +71,7 @@ import Data.Functor ((<&>))
 import Data.Hashable (Hashable)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
+import Data.Maybe.Strict (StrictMaybe (SJust))
 import qualified Data.Primitive.MutVar as Prim
 import qualified Data.Sequence as Seq
 import qualified Data.Set as Set
@@ -144,6 +145,7 @@ import Ouroboros.Consensus.Config (DiffusionPipeliningSupport (..))
 import Ouroboros.Consensus.HeaderValidation (HeaderWithTime)
 import Ouroboros.Consensus.Ledger.SupportsMempool
 import Ouroboros.Consensus.Ledger.SupportsProtocol
+import Ouroboros.Consensus.Mempool.API (getLeiosTxIndex)
 import Ouroboros.Consensus.MiniProtocol.BlockFetch.Server
 import Ouroboros.Consensus.MiniProtocol.ChainSync.Client
   ( ChainSyncStateView (..)
@@ -159,6 +161,7 @@ import Ouroboros.Consensus.NodeKernel
 import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
 import Ouroboros.Consensus.Storage.LedgerDB.Forker
   ( ResolveLeiosBlock
+  , leiosTxBytesOfGenTx
   )
 import Ouroboros.Consensus.Storage.Serialisation (SerialisedHeader)
 import Ouroboros.Consensus.Util (ShowProxy, whenJust)
@@ -423,6 +426,7 @@ mkHandlers
                       (Just peer)
                       Leios.ReceivedViaChainSync
                       Announcements.DoRelay
+                      (SJust hdrSlotTime)
                       (Just (diffRelTime now hdrSlotTime))
                       ancHdr
               }
@@ -519,7 +523,7 @@ mkHandlers
                                 (Leios.ancHeader ancH)
                           )
                           -- central part of the processing
-                          ( \ancHdr (shouldRelay, age, (p, _sz)) -> do
+                          ( \ancHdr (shouldRelay, onset, age, (p, _sz)) -> do
                               traceWith tracer $
                                 MkTraceLeiosPeer $
                                   "MsgLeiosBlockAnnouncement new: " <> Leios.prettyLeiosPoint p
@@ -531,6 +535,7 @@ mkHandlers
                                 (Just peer)
                                 Leios.ReceivedViaLeiosNotify
                                 shouldRelay
+                                (SJust onset)
                                 (Just age)
                                 ancHdr
                           )
@@ -544,43 +549,17 @@ mkHandlers
                     Prim.writeMutVar peerStateVar (latestPruneSlot', peerSt2)
                   MsgLeiosBlockOffer point ebBytesSize -> do
                     traceWith tracer $ MkTraceLeiosPeer $ "MsgLeiosBlockOffer " <> Leios.prettyLeiosPoint point
-                    let MkLeiosPoint{pointEbHash = ebHash} = point
-                    -- TODO: EB announcements now record the authoritative
-                    -- (forger-signed) size via 'recordAnnouncedEb', but this
-                    -- offer handler is not integrated with them yet: it still
-                    -- builds fetch state directly from peer offers, whose sizes
-                    -- are not authoritative (the authoritative one lives in
-                    -- 'headerLeiosAnnouncement' on the parent RB header). Until
-                    -- the two are reconciled, the sanitisation below is the best
-                    -- we can do against malformed offers: drop a zero-sized
-                    -- offer outright (no honest forger ever announces a 0-byte
-                    -- EB) and refuse to overwrite an existing entry that shares
-                    -- the same content hash, so the first-seen (slot, size)
-                    -- wins. The per-peer 'offerings' below is still updated so
-                    -- the peer remains a valid serving candidate.
-                    MVar.modifyMVar_ getLeiosOutstanding $ \outstanding ->
-                      pure $
-                        if ebBytesSize == 0
-                          || Set.member ebHash (Leios.acquiredEbBodies outstanding)
-                          || any
-                            ((== ebHash) . pointEbHash)
-                            (Map.keys (Leios.missingEbBodies outstanding))
-                          then outstanding
-                          else
-                            outstanding
-                              { Leios.missingEbBodies =
-                                  Map.insert point ebBytesSize (Leios.missingEbBodies outstanding)
-                              }
-                    MVar.modifyMVar_ (Leios.offerings peerVars) $ \(offers1, offers2) -> do
-                      let !offers1' = Set.insert ebHash offers1
-                      pure (offers1', offers2)
-                    void $ MVar.tryPutMVar getLeiosReady ()
+                    -- TODO punish peer for a too-old offer, modulo clock/immtip skew.
+                    Leios.recordEbBodyOffer
+                      (getLeiosOutstanding, getLeiosReady)
+                      peerVars
+                      Leios.TxsClosureNotAlsoOffered
+                      (point, ebBytesSize)
                   MsgLeiosBlockTxsOffer p -> do
                     traceWith tracer $ MkTraceLeiosPeer $ "MsgLeiosBlockTxsOffer " <> Leios.prettyLeiosPoint p
-                    let MkLeiosPoint{pointEbHash = ebHash} = p
-                    MVar.modifyMVar_ (Leios.offerings peerVars) $ \(offers1, offers2) -> do
-                      let !offers2' = Set.insert ebHash offers2
-                      pure (offers1, offers2')
+                    -- A closure offer implies the body too.
+                    MVar.modifyMVar_ (Leios.offerings peerVars) $
+                      pure . Map.insertWith Leios.mergeOffer p Leios.TxsClosureAlsoOffered
                     void $ MVar.tryPutMVar getLeiosReady ()
                   MsgLeiosVotes vs -> do
                     traceWith tracer $ MkTraceLeiosPeer $ "MsgLeiosVotes " <> show vs
@@ -691,6 +670,11 @@ mkHandlers
                   (getLeiosOutstanding, getLeiosReady)
                   getLeiosTxCache
                   leiosConn
+                  systemTime
+                  ( Leios.mkMempoolPull
+                      (atomically (getLeiosTxIndex getMempool))
+                      (leiosTxBytesOfGenTx . txForgetValidated)
+                  )
                   (Leios.MkPeerId peer)
                   reqVar
                   responseQ
@@ -1130,7 +1114,7 @@ mkApps kernel rng Tracers{tTxLogicTracer = _, ..} mkCodecs ByteLimits{..} chainS
           csjConfig
           getDiffusionPipeliningSupport
         $ \csState ->
-          bracketLeiosPeer them $ \peerVars -> do
+          bracketLeiosPeer them isBigLedgerPeer $ \peerVars -> do
             (r, trailing) <-
               runPipelinedPeerWithLimitsRnd
                 (contramap (TraceLabelPeer them) tChainSyncTracer)
@@ -1399,17 +1383,18 @@ mkApps kernel rng Tracers{tTxLogicTracer = _, ..} mkCodecs ByteLimits{..} chainS
   -- teardown has refunded it.
   bracketLeiosPeer ::
     ConnectionId addrNTN ->
+    IsBigLedgerPeer ->
     (Leios.LeiosPeerVars m -> m a) ->
     m a
-  bracketLeiosPeer them =
+  bracketLeiosPeer them isBigLedgerPeer =
     bracket
       -- Get-or-create: any peer-vars mini-protocol can be the first to run and
       -- allocate; the others share the existing vars. No ref count: a hot peer's
       -- mini-protocols tear down together, so whichever exits first runs the full
       -- cleanup below, and the rest find it already gone (idempotent). A straggler
       -- that allocated its own entry cleans that up on its own exit.
       ( do
-          fresh <- Leios.newLeiosPeerVars
+          fresh <- Leios.newLeiosPeerVars isBigLedgerPeer
           atomically $ do
             peersVars <- LazySTM.readTVar (getLeiosPeersVars kernel)
             case Map.lookup pid peersVars of
@@ -1436,10 +1421,11 @@ mkApps kernel rng Tracers{tTxLogicTracer = _, ..} mkCodecs ByteLimits{..} chainS
     ExpandedInitiatorContext
       { eicConnectionId = them
       , eicControlMessage = controlMessageSTM
+      , eicIsBigLedgerPeer = isBigLedgerPeer
       }
     channel = do
       labelThisThread "LeiosNotifyClient"
-      bracketLeiosPeer them $ \peerVars -> do
+      bracketLeiosPeer them isBigLedgerPeer $ \peerVars -> do
         ((), trailing) <-
           runPipelinedPeerWithLimits
             (TraceLabelPeer them `contramap` tLeiosNotifyTracer)
@@ -1480,10 +1466,11 @@ mkApps kernel rng Tracers{tTxLogicTracer = _, ..} mkCodecs ByteLimits{..} chainS
     ExpandedInitiatorContext
       { eicConnectionId = them
       , eicControlMessage = controlMessageSTM
+      , eicIsBigLedgerPeer = isBigLedgerPeer
       }
     channel = do
       labelThisThread "LeiosFetchClient"
-      bracketLeiosPeer them $ \peerVars ->
+      bracketLeiosPeer them isBigLedgerPeer $ \peerVars ->
         withLeiosDb leiosDB $ \leiosConn -> do
           ((), trailing) <-
             runPipelinedPeerWithLimits

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node.hs

@@ -963,7 +963,8 @@ mkNodeKernelArgs
   leiosTxCache =
     do
       let (kaRng, rng') = splitGen rng
-          (psRng, _) = splitGen rng'
+          (psRng, rng'') = splitGen rng'
+          (lfRng, _) = splitGen rng''
       return
         NodeKernelArgs
           { tracers
@@ -997,6 +998,7 @@ mkNodeKernelArgs
           , txSubmissionInitDelay
           , leiosDB
           , leiosTxCache
+          , leiosFetchRng = lfRng
           }
 
 -- | We allow the user running the node to customise the 'NodeKernelArgs'