Carry a forker in the mempool

Author: jasagredo

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

@@ -862,31 +862,29 @@ reproMempoolForge numBlks env = do
           <> "1 or 2 blocks at a time, not "
           <> show numBlks
 
-  mempool <-
-    Mempool.openMempoolWithoutSyncThread
-      Mempool.LedgerInterface
-        { Mempool.getCurrentLedgerState = ledgerState <$> LedgerDB.getVolatileTip ledgerDB
-        , Mempool.getLedgerTablesAtFor = \pt keys -> do
-            frk <- LedgerDB.getForkerAtTarget ledgerDB registry (SpecificPoint pt)
-            case frk of
-              Left _ -> pure Nothing
-              Right fr -> do
-                tbs <-
-                  Just . castLedgerTables
-                    <$> LedgerDB.forkerReadTables fr (castLedgerTables keys)
-                LedgerDB.forkerClose fr
-                pure tbs
-        }
-      lCfg
-      -- one mebibyte should generously accomodate two blocks' worth of txs
-      ( Mempool.MempoolCapacityBytesOverride $
-          LedgerSupportsMempool.ByteSize32 $
-            1024 * 1024
-      )
-      nullTracer
-
-  void $ processAll db registry GetBlock startFrom limit Nothing (process howManyBlocks mempool)
-  pure Nothing
+  withRegistry $ \reg -> do
+    mempool <-
+      Mempool.openMempoolWithoutSyncThread
+        reg
+        Mempool.LedgerInterface
+          { Mempool.getCurrentLedgerState = \reg' -> do
+              st <- LedgerDB.getVolatileTip ledgerDB
+              pure
+                ( ledgerState st
+                , fmap (LedgerDB.ledgerStateReadOnlyForker . LedgerDB.readOnlyForker)
+                    <$> LedgerDB.getForkerAtTarget ledgerDB reg' (SpecificPoint (castPoint $ getTip st))
+                )
+          }
+        lCfg
+        -- one mebibyte should generously accomodate two blocks' worth of txs
+        ( Mempool.MempoolCapacityBytesOverride $
+            LedgerSupportsMempool.ByteSize32 $
+              1024 * 1024
+        )
+        nullTracer
+
+    void $ processAll db registry GetBlock startFrom limit Nothing (process howManyBlocks mempool)
+    pure Nothing
  where
   AnalysisEnv
     { cfg

ouroboros-consensus-diffusion/src/unstable-diffusion-testlib/Test/ThreadNet/Network.hs

@@ -628,16 +628,15 @@ runThreadNetwork
       HasCallStack =>
       OracularClock m ->
       SlotNo ->
-      ResourceRegistry m ->
       (SlotNo -> STM m ()) ->
       STM m (Point blk) ->
       (ResourceRegistry m -> m (ReadOnlyForker' m blk)) ->
       Mempool m blk ->
       [GenTx blk] ->
       -- \^ valid transactions the node should immediately propagate
-      m ()
-    forkCrucialTxs clock s0 registry unblockForge getTipPoint mforker mempool txs0 = do
-      void $ forkLinkedThread registry "crucialTxs" $ withRegistry $ \reg -> do
+      m (Thread m Void)
+    forkCrucialTxs clock s0 unblockForge getTipPoint mforker mempool txs0 = do
+      testForkMempoolThread mempool "crucialTxs" $ withRegistry $ \reg -> do
         let loop (slot, mempFp) = do
               forker <- mforker reg
               extLedger <- atomically $ roforkerGetLedgerState forker
@@ -1150,15 +1149,17 @@ runThreadNetwork
       --
       -- TODO Is there a risk that this will block because the 'forkTxProducer'
       -- fills up the mempool too quickly?
-      forkCrucialTxs
-        clock
-        joinSlot
-        registry
-        unblockForge
-        (ledgerTipPoint . ledgerState <$> ChainDB.getCurrentLedger chainDB)
-        getForker
-        mempool
-        txs0
+      threadCrucialTxs <-
+        forkCrucialTxs
+          clock
+          joinSlot
+          unblockForge
+          (ledgerTipPoint . ledgerState <$> ChainDB.getCurrentLedger chainDB)
+          getForker
+          mempool
+          txs0
+
+      void $ allocate registry (\_ -> pure threadCrucialTxs) cancelThread
 
       forkTxProducer
         coreNodeId

ouroboros-consensus/changelog.d/20250620_132627_jasataco_mempool_carry_vh.md

@@ -0,0 +1,48 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+### Patch
+
+- The mempool will now carry its own forker instead of acquiring one on each
+  revalidation. This particularly implies that the mempool will no longer
+  re-sync under the hood while trying to add a transaction, and only the
+  background thread will perform such a re-sync.
+
+- The ledger state that triggers a re-sync in the background thread will be the
+  one used to revalidate the mempool. Previously, another read for the new tip
+  would be performed when starting the sync, which could lead to confusing
+  (although innocuous) situations in which:
+
+   - the tip changes from A to B
+   - the background thread sees the tip changed from A, so it records the new tip (B) and triggers the re-sync
+   - the tip changes again from B to C before the syncing process reads the tip again
+   - the mempool is re-synced with C
+   - the background thread would now see that the tip changed from B, so it records the new tip (C) and triggers the re-sync
+   - the mempool is re-synced **AGAIN** with C
+
+  This sequence of actions can't happen again.
+
+- The mempool now has its own registry in which it allocates forkers. The
+  background thread was moved to this inner registry such that it can access the
+  mempool internal registry, but an action to cancel it will still live in the
+  outer registry, to ensure the thread is closed before we attempt to close the
+  mempool internal registry. Otherwise we would run into a race condition if the
+  background thread would attempt a resync while the internal registry was being
+  closed.
+
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus/ouroboros-consensus.cabal

@@ -540,6 +540,7 @@ library unstable-mempool-test-utils
   exposed-modules: Test.Consensus.Mempool.Mocked
   build-depends:
     base,
+    resource-registry,
     contra-tracer,
     deepseq,
     ouroboros-consensus,

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/API.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE UndecidableInstances #-}
@@ -34,6 +35,7 @@ module Ouroboros.Consensus.Mempool.API
   , zeroTicketNo
   ) where
 
+import Control.ResourceRegistry
 import qualified Data.List.NonEmpty as NE
 import Ouroboros.Consensus.Block (ChainHash, SlotNo)
 import Ouroboros.Consensus.Ledger.Abstract
@@ -212,6 +214,12 @@ data Mempool m blk = Mempool
   -- Instead, we treat it the same way as a Mempool which is /at/
   -- capacity, i.e., we won't admit new transactions until some have been
   -- removed because they have become invalid.
+  , testForkMempoolThread :: forall a. String -> m a -> m (Thread m a)
+  -- ^ FOR TESTS ONLY
+  --
+  -- If we want to run a thread that can perform syncs in the mempool, it needs
+  -- to be registered in the mempool's internal registry. This function exposes
+  -- such functionality.
   }
 
 {-------------------------------------------------------------------------------

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TypeApplications #-}
@@ -39,9 +40,9 @@ module Ouroboros.Consensus.Mempool.Impl.Common
   , tickLedgerState
   ) where
 
-import Control.Concurrent.Class.MonadMVar (MVar, newMVar)
 import Control.Concurrent.Class.MonadSTM.Strict.TMVar (newTMVarIO)
 import Control.Monad.Trans.Except (runExcept)
+import Control.ResourceRegistry
 import Control.Tracer
 import qualified Data.Foldable as Foldable
 import qualified Data.List.NonEmpty as NE
@@ -61,8 +62,11 @@ import Ouroboros.Consensus.Mempool.TxSeq (TxSeq (..), TxTicket (..))
 import qualified Ouroboros.Consensus.Mempool.TxSeq as TxSeq
 import Ouroboros.Consensus.Storage.ChainDB (ChainDB)
 import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
+import Ouroboros.Consensus.Storage.LedgerDB.Forker
 import Ouroboros.Consensus.Util.Enclose (EnclosingTimed)
 import Ouroboros.Consensus.Util.IOLike hiding (newMVar)
+import Ouroboros.Consensus.Util.NormalForm.StrictMVar
+import Ouroboros.Network.Protocol.LocalStateQuery.Type
 
 {-------------------------------------------------------------------------------
   Internal State
@@ -188,26 +192,26 @@ initInternalState capacityOverride lastTicketNo cfg slot st =
 
 -- | Abstract interface needed to run a Mempool.
 data LedgerInterface m blk = LedgerInterface
-  { getCurrentLedgerState :: STM m (LedgerState blk EmptyMK)
-  -- ^ Get the current tip of the LedgerDB.
-  , getLedgerTablesAtFor ::
-      Point blk ->
-      LedgerTables (LedgerState blk) KeysMK ->
-      m (Maybe (LedgerTables (LedgerState blk) ValuesMK))
-  -- ^ Get values at the given point on the chain. Returns Nothing if the
-  -- anchor moved or if the state is not found on the ledger db.
+  { getCurrentLedgerState ::
+      ResourceRegistry m ->
+      STM m (LedgerState blk EmptyMK, m (Either GetForkerError (ReadOnlyForker m (LedgerState blk) blk)))
+  -- ^ Get the current tip of the LedgerDB and an action to get a forker there.
   }
 
 -- | Create a 'LedgerInterface' from a 'ChainDB'.
 chainDBLedgerInterface ::
-  IOLike m =>
-  ChainDB m blk -> LedgerInterface m blk
+  (IOLike m, IsLedger (LedgerState blk)) =>
+  ChainDB m blk ->
+  LedgerInterface m blk
 chainDBLedgerInterface chainDB =
   LedgerInterface
-    { getCurrentLedgerState =
-        ledgerState <$> ChainDB.getCurrentLedger chainDB
-    , getLedgerTablesAtFor = \pt keys ->
-        fmap castLedgerTables <$> ChainDB.getLedgerTablesAtFor chainDB pt (castLedgerTables keys)
+    { getCurrentLedgerState = \reg -> do
+        st <- ChainDB.getCurrentLedger chainDB
+        pure
+          ( ledgerState st
+          , fmap (fmap ledgerStateReadOnlyForker) $
+              ChainDB.getReadOnlyForkerAtPoint chainDB reg (SpecificPoint (castPoint $ getTip st))
+          )
     }
 
 {-------------------------------------------------------------------------------
@@ -219,10 +223,12 @@ chainDBLedgerInterface chainDB =
 -- different operations.
 data MempoolEnv m blk = MempoolEnv
   { mpEnvLedger :: LedgerInterface m blk
+  , mpEnvForker :: StrictMVar m (ReadOnlyForker m (LedgerState blk) blk)
   , mpEnvLedgerCfg :: LedgerConfig blk
+  , mpEnvRegistry :: ResourceRegistry m
   , mpEnvStateVar :: StrictTMVar m (InternalState blk)
-  , mpEnvAddTxsRemoteFifo :: MVar m ()
-  , mpEnvAddTxsAllFifo :: MVar m ()
+  , mpEnvAddTxsRemoteFifo :: StrictMVar m ()
+  , mpEnvAddTxsAllFifo :: StrictMVar m ()
   , mpEnvTracer :: Tracer m (TraceEventMempool blk)
   , mpEnvCapacityOverride :: MempoolCapacityBytesOverride
   }
@@ -236,25 +242,39 @@ initMempoolEnv ::
   LedgerConfig blk ->
   MempoolCapacityBytesOverride ->
   Tracer m (TraceEventMempool blk) ->
+  ResourceRegistry m ->
   m (MempoolEnv m blk)
-initMempoolEnv ledgerInterface cfg capacityOverride tracer = do
-  st <- atomically $ getCurrentLedgerState ledgerInterface
-  let (slot, st') = tickLedgerState cfg (ForgeInUnknownSlot st)
-  isVar <-
-    newTMVarIO $
-      initInternalState capacityOverride TxSeq.zeroTicketNo cfg slot st'
-  addTxRemoteFifo <- newMVar ()
-  addTxAllFifo <- newMVar ()
-  return
-    MempoolEnv
-      { mpEnvLedger = ledgerInterface
-      , mpEnvLedgerCfg = cfg
-      , mpEnvStateVar = isVar
-      , mpEnvAddTxsRemoteFifo = addTxRemoteFifo
-      , mpEnvAddTxsAllFifo = addTxAllFifo
-      , mpEnvTracer = tracer
-      , mpEnvCapacityOverride = capacityOverride
-      }
+initMempoolEnv ledgerInterface cfg capacityOverride tracer registry = do
+  (_, mpEnvRegistry) <- allocate registry (\_ -> unsafeNewRegistry) closeRegistry
+  initMempoolEnv' mpEnvRegistry
+ where
+  initMempoolEnv' reg = do
+    (st, meFrk) <- atomically $ getCurrentLedgerState ledgerInterface reg
+    eFrk <- meFrk
+    case eFrk of
+      -- This should happen very rarely, if between getting the state and getting
+      -- the forker, the ledgerdb has changed. We just retry here.
+      Left{} -> initMempoolEnv' reg
+      Right frk -> do
+        frkMVar <- newMVar frk
+        let (slot, st') = tickLedgerState cfg (ForgeInUnknownSlot st)
+        isVar <-
+          newTMVarIO $
+            initInternalState capacityOverride TxSeq.zeroTicketNo cfg slot st'
+        addTxRemoteFifo <- newMVar ()
+        addTxAllFifo <- newMVar ()
+        return
+          MempoolEnv
+            { mpEnvLedger = ledgerInterface
+            , mpEnvLedgerCfg = cfg
+            , mpEnvForker = frkMVar
+            , mpEnvRegistry = reg
+            , mpEnvStateVar = isVar
+            , mpEnvAddTxsRemoteFifo = addTxRemoteFifo
+            , mpEnvAddTxsAllFifo = addTxAllFifo
+            , mpEnvTracer = tracer
+            , mpEnvCapacityOverride = capacityOverride
+            }
 
 {-------------------------------------------------------------------------------
   Ticking the ledger state