Carry a forker in the mempool (#1565)

# Description

- 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 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.

Author: jasagredo

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

@@ -77,6 +77,7 @@ import Ouroboros.Consensus.Ledger.SupportsProtocol
   )
 import Ouroboros.Consensus.Ledger.Tables.Utils
 import qualified Ouroboros.Consensus.Mempool as Mempool
+import Ouroboros.Consensus.Mempool.Impl.Common
 import Ouroboros.Consensus.Protocol.Abstract (LedgerView)
 import Ouroboros.Consensus.Storage.Common (BlockComponent (..))
 import Ouroboros.Consensus.Storage.ImmutableDB (ImmutableDB)
@@ -862,31 +863,30 @@ 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 $
+                MempoolLedgerDBView
+                  (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
@@ -991,7 +991,7 @@ reproMempoolForge numBlks env = do
           LedgerDB.tryFlush ledgerDB
 
           -- this flushes blk from the mempool, since every tx in it is now on the chain
-          void $ Mempool.syncWithLedger mempool
+          void $ Mempool.testSyncWithLedger mempool
 
 {-------------------------------------------------------------------------------
   Auxiliary: processing all blocks in the DB

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
@@ -679,7 +678,7 @@ runThreadNetwork
               -- avoid the race in which we wake up before the mempool's
               -- background thread wakes up by mimicking it before we do
               -- anything else
-              void $ syncWithLedger mempool
+              void $ testSyncWithLedger mempool
 
               loop fps'
         loop (s0, [])
@@ -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,33 @@
+<!--
+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 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
+
+- Removed `getLedgerTablesAtFor` from the ChainDB API. Clients now have to
+  actually open a forker and manage it.

ouroboros-consensus/ouroboros-consensus.cabal

@@ -543,6 +543,7 @@ library unstable-mempool-test-utils
     contra-tracer,
     deepseq,
     ouroboros-consensus,
+    resource-registry,
     strict-stm,
 
 library unstable-tutorials

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,8 +35,9 @@ 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.Block (ChainHash, Point, SlotNo)
 import Ouroboros.Consensus.Ledger.Abstract
 import Ouroboros.Consensus.Ledger.SupportsMempool
 import qualified Ouroboros.Consensus.Mempool.Capacity as Cap
@@ -109,11 +111,11 @@ data Mempool m blk = Mempool
   --
   -- Note that transactions that are invalid will /never/ be added to the
   -- mempool. However, it is possible that, at a given point in time,
-  -- transactions which were valid in an older ledger state but are invalid
-  -- in the current ledger state, could exist within the mempool until they
-  -- are revalidated and dropped from the mempool via a call to
-  -- 'syncWithLedger' or by the background thread that watches the ledger
-  -- for changes.
+  -- transactions which were valid in an older ledger state but are invalid in
+  -- the current ledger state, could exist within the mempool until they are
+  -- revalidated and dropped from the mempool via a call to by the background
+  -- thread that watches the ledger for changes or by 'testSyncWithLedger' in
+  -- testing scenarios.
   --
   -- This action returns one of two results.
   --
@@ -161,22 +163,6 @@ data Mempool m blk = Mempool
   -- persistence.
   , removeTxsEvenIfValid :: NE.NonEmpty (GenTxId blk) -> m ()
   -- ^ Manually remove the given transactions from the mempool.
-  , syncWithLedger :: m (MempoolSnapshot blk)
-  -- ^ Sync the transactions in the mempool with the current ledger state
-  --  of the 'ChainDB'.
-  --
-  -- The transactions that exist within the mempool will be revalidated
-  -- against the current ledger state. Transactions which are found to be
-  -- invalid with respect to the current ledger state, will be dropped
-  -- from the mempool, whereas valid transactions will remain.
-  --
-  -- We keep this in @m@ instead of @STM m@ to leave open the possibility
-  -- of persistence. Additionally, this makes it possible to trace the
-  -- removal of invalid transactions.
-  --
-  -- n.b. in our current implementation, when one opens a mempool, we
-  -- spawn a thread which performs this action whenever the 'ChainDB' tip
-  -- point changes.
   , getSnapshot :: STM m (MempoolSnapshot blk)
   -- ^ Get a snapshot of the current mempool state. This allows for
   -- further pure queries on the snapshot.
@@ -212,6 +198,33 @@ 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.
+  , testSyncWithLedger :: m (MempoolSnapshot blk)
+  -- ^ ONLY FOR TESTS
+  --
+  -- Sync the transactions in the mempool with the current ledger state
+  --  of the 'ChainDB'.
+  --
+  -- The transactions that exist within the mempool will be revalidated
+  -- against the current ledger state. Transactions which are found to be
+  -- invalid with respect to the current ledger state, will be dropped
+  -- from the mempool, whereas valid transactions will remain.
+  --
+  -- We keep this in @m@ instead of @STM m@ to leave open the possibility
+  -- of persistence. Additionally, this makes it possible to trace the
+  -- removal of invalid transactions.
+  --
+  -- n.b. in our current implementation, when one opens a mempool, we
+  -- spawn a thread which performs this action whenever the 'ChainDB' tip
+  -- point changes.
+  , 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.
+  --
+  -- The 'String' passed will be used as the thread label, and the @m a@ will be
+  -- the action forked in the thread.
   }
 
 {-------------------------------------------------------------------------------
@@ -353,4 +366,5 @@ data MempoolSnapshot blk = MempoolSnapshot
   , snapshotStateHash :: ChainHash (TickedLedgerState blk)
   -- ^ The resulting state currently in the mempool after applying the
   -- transactions
+  , snapshotPoint :: Point blk
   }