Full rework of the BlockFetch logic for bulk sync mode

cabal.project

@@ -53,3 +53,13 @@ if impl(ghc >= 9.10)
   constraints:
     -- Earlier versions do not compile with ghc-9.10
     , plutus-ledger-api ^>=1.31
+
+source-repository-package
+  type: git
+  location: https://github.com/IntersectMBO/ouroboros-network
+  tag: 5c304e5adbd27907c675bd60a2282264a65f117c
+  --sha256: Jn3Qqlsirlyu23bsFN2SO054LIe2rcsHUW1tN8Pm1Ks=
+  subdir:
+    ouroboros-network
+    ouroboros-network-api
+    ouroboros-network-protocols

ouroboros-consensus-diffusion/changelog.d/20240807_100458_alexander.esgen_milestone_1.md

@@ -0,0 +1,3 @@
+### Breaking
+
+- Adapted to Genesis-related changes in `ouroboros-consensus` ([#1179](https://github.com/IntersectMBO/ouroboros-consensus/pull/1179)).

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

@@ -1,24 +1,32 @@
+{-# LANGUAGE DeriveGeneric       #-}
 {-# LANGUAGE DeriveTraversable   #-}
 {-# LANGUAGE DerivingStrategies  #-}
 {-# LANGUAGE LambdaCase          #-}
 {-# LANGUAGE NamedFieldPuns      #-}
 {-# LANGUAGE NumericUnderscores  #-}
+{-# LANGUAGE RecordWildCards     #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Ouroboros.Consensus.Node.Genesis (
     -- * 'GenesisConfig'
     GenesisConfig (..)
+  , GenesisConfigFlags (..)
   , LoEAndGDDConfig (..)
+  , defaultGenesisConfigFlags
   , disableGenesisConfig
   , enableGenesisConfigDefault
+  , mkGenesisConfig
     -- * NodeKernel helpers
   , GenesisNodeKernelArgs (..)
+  , LoEAndGDDNodeKernelArgs (..)
   , mkGenesisNodeKernelArgs
   , setGetLoEFragment
   ) where
 
 import           Control.Monad (join)
+import           Data.Maybe (fromMaybe)
 import           Data.Traversable (for)
+import           GHC.Generics (Generic)
 import           Ouroboros.Consensus.Block
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
                      (CSJConfig (..), CSJEnabledConfig (..),
@@ -32,51 +40,121 @@ import           Ouroboros.Consensus.Util.Args
 import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import qualified Ouroboros.Network.AnchoredFragment as AF
+import           Ouroboros.Network.BlockFetch
+                     (GenesisBlockFetchConfiguration (..))
 
 -- | Whether to en-/disable the Limit on Eagerness and the Genesis Density
 -- Disconnector.
 data LoEAndGDDConfig a =
     LoEAndGDDEnabled !a
   | LoEAndGDDDisabled
-  deriving stock (Show, Functor, Foldable, Traversable)
+  deriving stock (Eq, Generic, Show, Functor, Foldable, Traversable)
 
 -- | Aggregating the various configs for Genesis-related subcomponents.
-data GenesisConfig = GenesisConfig {
-    gcChainSyncLoPBucketConfig :: !ChainSyncLoPBucketConfig
+data GenesisConfig = GenesisConfig
+  { gcBlockFetchConfig         :: !GenesisBlockFetchConfiguration
+  , gcChainSyncLoPBucketConfig :: !ChainSyncLoPBucketConfig
   , gcCSJConfig                :: !CSJConfig
-  , gcLoEAndGDDConfig          :: !(LoEAndGDDConfig ())
+  , gcLoEAndGDDConfig          :: !(LoEAndGDDConfig LoEAndGDDParams)
+  } deriving stock (Eq, Generic, Show)
+
+-- | Genesis configuration flags and low-level args, as parsed from config file or CLI
+data GenesisConfigFlags = GenesisConfigFlags
+  { gcfEnableCSJ             :: Bool
+  , gcfEnableLoEAndGDD       :: Bool
+  , gcfEnableLoP             :: Bool
+  , gcfBlockFetchGracePeriod :: Maybe Integer
+  , gcfBucketCapacity        :: Maybe Integer
+  , gcfBucketRate            :: Maybe Integer
+  , gcfCSJJumpSize           :: Maybe Integer
+  , gcfGDDRateLimit          :: Maybe DiffTime
+  } deriving stock (Eq, Generic, Show)
+
+defaultGenesisConfigFlags :: GenesisConfigFlags
+defaultGenesisConfigFlags = GenesisConfigFlags
+  { gcfEnableCSJ              = True
+  , gcfEnableLoEAndGDD        = True
+  , gcfEnableLoP              = True
+  , gcfBlockFetchGracePeriod  = Nothing
+  , gcfBucketCapacity         = Nothing
+  , gcfBucketRate             = Nothing
+  , gcfCSJJumpSize            = Nothing
+  , gcfGDDRateLimit           = Nothing
   }
 
--- TODO justification/derivation from other parameters
 enableGenesisConfigDefault :: GenesisConfig
-enableGenesisConfigDefault = GenesisConfig {
-      gcChainSyncLoPBucketConfig = ChainSyncLoPBucketEnabled ChainSyncLoPBucketEnabledConfig {
-          csbcCapacity = 100_000 -- number of tokens
-        , csbcRate     = 500 -- tokens per second leaking, 1/2ms
-        }
-    , gcCSJConfig = CSJEnabled CSJEnabledConfig {
-          csjcJumpSize = 3 * 2160 * 20 -- mainnet forecast range
-        }
-    , gcLoEAndGDDConfig = LoEAndGDDEnabled ()
-    }
+enableGenesisConfigDefault = mkGenesisConfig $ Just $ defaultGenesisConfigFlags
 
 -- | Disable all Genesis components, yielding Praos behavior.
 disableGenesisConfig :: GenesisConfig
-disableGenesisConfig = GenesisConfig {
-      gcChainSyncLoPBucketConfig = ChainSyncLoPBucketDisabled
+disableGenesisConfig = mkGenesisConfig Nothing
+
+mkGenesisConfig :: Maybe GenesisConfigFlags -> GenesisConfig
+mkGenesisConfig Nothing = -- disable Genesis
+  GenesisConfig
+    { gcBlockFetchConfig = GenesisBlockFetchConfiguration
+        { gbfcGracePeriod = 0 -- no grace period when Genesis is disabled
+        }
+    , gcChainSyncLoPBucketConfig = ChainSyncLoPBucketDisabled
     , gcCSJConfig                = CSJDisabled
     , gcLoEAndGDDConfig          = LoEAndGDDDisabled
     }
+mkGenesisConfig (Just GenesisConfigFlags{..}) =
+  GenesisConfig
+    { gcBlockFetchConfig = GenesisBlockFetchConfiguration
+        { gbfcGracePeriod
+        }
+    , gcChainSyncLoPBucketConfig = if gcfEnableLoP
+        then ChainSyncLoPBucketEnabled ChainSyncLoPBucketEnabledConfig
+          { csbcCapacity
+          , csbcRate
+          }
+        else ChainSyncLoPBucketDisabled
+    , gcCSJConfig = if gcfEnableCSJ
+        then CSJEnabled CSJEnabledConfig
+          { csjcJumpSize
+          }
+        else CSJDisabled
+    , gcLoEAndGDDConfig = if gcfEnableLoEAndGDD
+        then LoEAndGDDEnabled LoEAndGDDParams{lgpGDDRateLimit}
+        else LoEAndGDDDisabled
+    }
+  where
+    -- TODO justification/derivation from other parameters
+    defaultBlockFetchGracePeriod = 10 -- seconds
+    defaultCapacity              = 100_000 -- number of tokens
+    defaultRate                  = 500 -- tokens per second leaking, 1/2ms
+    -- 3 * 2160 * 20 works in more recent ranges of slots, but causes syncing to
+    -- block in byron.
+    defaultCSJJumpSize           = 2 * 2160
+    defaultGDDRateLimit          = 1.0 -- seconds
+
+    gbfcGracePeriod = fromInteger $ fromMaybe defaultBlockFetchGracePeriod gcfBlockFetchGracePeriod
+    csbcCapacity              = fromInteger $ fromMaybe defaultCapacity gcfBucketCapacity
+    csbcRate                  = fromInteger $ fromMaybe defaultRate gcfBucketRate
+    csjcJumpSize              = fromInteger $ fromMaybe defaultCSJJumpSize gcfCSJJumpSize
+    lgpGDDRateLimit            = fromMaybe defaultGDDRateLimit gcfGDDRateLimit
+
+newtype LoEAndGDDParams = LoEAndGDDParams
+  { -- | How often to evaluate GDD. 0 means as soon as possible.
+    -- Otherwise, no faster than once every T seconds, where T is the
+    -- value of the field.
+    lgpGDDRateLimit :: DiffTime
+  } deriving stock (Eq, Generic, Show)
 
 -- | Genesis-related arguments needed by the NodeKernel initialization logic.
 data GenesisNodeKernelArgs m blk = GenesisNodeKernelArgs {
+    gnkaLoEAndGDDArgs :: !(LoEAndGDDConfig (LoEAndGDDNodeKernelArgs m blk))
+  }
+
+data LoEAndGDDNodeKernelArgs m blk = LoEAndGDDNodeKernelArgs {
     -- | A TVar containing an action that returns the 'ChainDB.GetLoEFragment'
     -- action. We use this extra indirection to update this action after we
     -- opened the ChainDB (which happens before we initialize the NodeKernel).
     -- After that, this TVar will not be modified again.
-    gnkaGetLoEFragment :: !(LoEAndGDDConfig (StrictTVar m (ChainDB.GetLoEFragment m blk)))
+    lgnkaLoEFragmentTVar :: !(StrictTVar m (ChainDB.GetLoEFragment m blk))
+  , lgnkaGDDRateLimit    :: DiffTime
   }
-
 -- | Create the initial 'GenesisNodeKernelArgs" (with a temporary
 -- 'ChainDB.GetLoEFragment' that will be replaced via 'setGetLoEFragment') and a
 -- function to update the 'ChainDbArgs' accordingly.
@@ -87,20 +165,24 @@ mkGenesisNodeKernelArgs ::
        , Complete ChainDbArgs m blk -> Complete ChainDbArgs m blk
        )
 mkGenesisNodeKernelArgs gcfg = do
-    gnkaGetLoEFragment <- for (gcLoEAndGDDConfig gcfg) $ \() ->
-        newTVarIO $ pure $
+    gnkaLoEAndGDDArgs <- for (gcLoEAndGDDConfig gcfg) $ \p -> do
+        loeFragmentTVar <- newTVarIO $ pure $
           -- Use the most conservative LoE fragment until 'setGetLoEFragment'
           -- is called.
           ChainDB.LoEEnabled $ AF.Empty AF.AnchorGenesis
-    let updateChainDbArgs = case gnkaGetLoEFragment of
+        pure LoEAndGDDNodeKernelArgs
+          { lgnkaLoEFragmentTVar = loeFragmentTVar
+          , lgnkaGDDRateLimit = lgpGDDRateLimit p
+          }
+    let updateChainDbArgs = case gnkaLoEAndGDDArgs of
           LoEAndGDDDisabled -> id
-          LoEAndGDDEnabled varGetLoEFragment -> \cfg ->
+          LoEAndGDDEnabled lgnkArgs -> \cfg ->
             cfg { ChainDB.cdbsArgs =
                   (ChainDB.cdbsArgs cfg) { ChainDB.cdbsLoE = getLoEFragment }
                 }
             where
-              getLoEFragment = join $ readTVarIO varGetLoEFragment
-    pure (GenesisNodeKernelArgs {gnkaGetLoEFragment}, updateChainDbArgs)
+              getLoEFragment = join $ readTVarIO $ lgnkaLoEFragmentTVar lgnkArgs
+    pure (GenesisNodeKernelArgs{gnkaLoEAndGDDArgs}, updateChainDbArgs)
 
 -- | Set 'gnkaGetLoEFragment' to the actual logic for determining the current
 -- LoE fragment.

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

@@ -31,14 +31,17 @@ import           Ouroboros.Consensus.MiniProtocol.BlockFetch.Server
                      (TraceBlockFetchServerEvent)
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
                      (InvalidBlockReason, TraceChainSyncClientEvent)
+import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client.Jumping as CSJumping
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Server
                      (TraceChainSyncServerEvent)
 import           Ouroboros.Consensus.MiniProtocol.LocalTxSubmission.Server
                      (TraceLocalTxSubmissionServerEvent (..))
 import           Ouroboros.Consensus.Node.GSM (TraceGsmEvent)
 import           Ouroboros.Network.Block (Tip)
-import           Ouroboros.Network.BlockFetch (FetchDecision,
-                     TraceFetchClientState, TraceLabelPeer)
+import           Ouroboros.Network.BlockFetch (TraceFetchClientState,
+                     TraceLabelPeer)
+import           Ouroboros.Network.BlockFetch.Decision.Trace
+                     (TraceDecisionEvent)
 import           Ouroboros.Network.KeepAlive (TraceKeepAliveClient)
 import           Ouroboros.Network.TxSubmission.Inbound
                      (TraceTxSubmissionInbound)
@@ -53,7 +56,7 @@ data Tracers' remotePeer localPeer blk f = Tracers
   { chainSyncClientTracer         :: f (TraceLabelPeer remotePeer (TraceChainSyncClientEvent blk))
   , chainSyncServerHeaderTracer   :: f (TraceLabelPeer remotePeer (TraceChainSyncServerEvent blk))
   , chainSyncServerBlockTracer    :: f (TraceChainSyncServerEvent blk)
-  , blockFetchDecisionTracer      :: f [TraceLabelPeer remotePeer (FetchDecision [Point (Header blk)])]
+  , blockFetchDecisionTracer      :: f (TraceDecisionEvent remotePeer (Header blk))
   , blockFetchClientTracer        :: f (TraceLabelPeer remotePeer (TraceFetchClientState (Header blk)))
   , blockFetchServerTracer        :: f (TraceLabelPeer remotePeer (TraceBlockFetchServerEvent blk))
   , txInboundTracer               :: f (TraceLabelPeer remotePeer (TraceTxSubmissionInbound  (GenTxId blk) (GenTx blk)))
@@ -68,6 +71,7 @@ data Tracers' remotePeer localPeer blk f = Tracers
   , consensusErrorTracer          :: f SomeException
   , gsmTracer                     :: f (TraceGsmEvent (Tip blk))
   , gddTracer                     :: f (TraceGDDEvent remotePeer blk)
+  , csjTracer                     :: f (CSJumping.TraceEvent remotePeer)
   }
 
 instance (forall a. Semigroup (f a))
@@ -91,6 +95,7 @@ instance (forall a. Semigroup (f a))
       , consensusErrorTracer          = f consensusErrorTracer
       , gsmTracer                     = f gsmTracer
       , gddTracer                     = f gddTracer
+      , csjTracer                     = f csjTracer
       }
     where
       f :: forall a. Semigroup a
@@ -122,6 +127,7 @@ nullTracers = Tracers
     , consensusErrorTracer          = nullTracer
     , gsmTracer                     = nullTracer
     , gddTracer                     = nullTracer
+    , csjTracer                     = nullTracer
     }
 
 showTracers :: ( Show blk
@@ -156,6 +162,7 @@ showTracers tr = Tracers
     , consensusErrorTracer          = showTracing tr
     , gsmTracer                     = showTracing tr
     , gddTracer                     = showTracing tr
+    , csjTracer                     = showTracing tr
     }
 
 {-------------------------------------------------------------------------------