Fix resource registry handle in V2 Forker handles

cabal.project

@@ -14,7 +14,7 @@ repository cardano-haskell-packages
 -- update either of these.
 index-state:
   -- Bump this if you need newer packages from Hackage
-  , hackage.haskell.org 2025-09-26T20:57:57Z
+  , hackage.haskell.org 2025-10-23T13:39:53Z
   -- Bump this if you need newer packages from CHaP
   , cardano-haskell-packages 2025-10-01T14:54:25Z
 

ouroboros-consensus-diffusion/changelog.d/20251023_152349_javier.sagredo_fix_resource_registry.md

@@ -0,0 +1,23 @@
+<!--
+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
+
+- Bump to `resource-registry ^>= 0.2`.
+
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus-diffusion/ouroboros-consensus-diffusion.cabal

@@ -95,7 +95,7 @@ library
     ouroboros-network-framework ^>=0.19,
     ouroboros-network-protocols ^>=0.15,
     random,
-    resource-registry ^>=0.1,
+    resource-registry ^>=0.2,
     safe-wild-cards ^>=1.0,
     serialise ^>=0.2,
     text,

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

@@ -1158,7 +1158,7 @@ runThreadNetwork
           mempool
           txs0
 
-      void $ allocate registry (\_ -> pure threadCrucialTxs) cancelThread
+      void $ allocateThread registry (\_ -> pure threadCrucialTxs)
 
       forkTxProducer
         coreNodeId

ouroboros-consensus/changelog.d/20251023_152234_javier.sagredo_fix_resource_registry.md

@@ -0,0 +1,22 @@
+<!--
+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
+
+- Bump to `resource-registry ^>= 0.2`.
+
+### Non-Breaking
+
+- Committing a forker will move the handles to the registry of the LedgerDB. The
+  discarded fork will be queued to be released by the `garbageCollect` logic.
+
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus/changelog.d/20251029_113308_javier.sagredo_fix_mempool_dangling_forker.md

@@ -0,0 +1,25 @@
+<!--
+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 sync thread was allocated in the top level registry in order to
+  ensure it would be cancelled before the mempool registry was shutting
+  down. This was solved in `resource-registry-0.2.0.0`.
+
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus/ouroboros-consensus.cabal

@@ -344,7 +344,7 @@ library
     psqueues ^>=0.2.3,
     quiet ^>=0.2,
     rawlock ^>=0.1.1,
-    resource-registry ^>=0.1,
+    resource-registry ^>=0.2,
     semialign >=1.1,
     serialise ^>=0.2,
     singletons,
@@ -393,7 +393,7 @@ library ouroboros-consensus-lsm
     ouroboros-consensus,
     primitive,
     random,
-    resource-registry ^>=0.1,
+    resource-registry ^>=0.2,
     serialise ^>=0.2,
     streaming,
     text,

ouroboros-consensus/src/ouroboros-consensus-lsm/Ouroboros/Consensus/Storage/LedgerDB/V2/LSM.hs

@@ -64,6 +64,7 @@ import Data.Void
 import Database.LSMTree (Salt, Session, Table)
 import qualified Database.LSMTree as LSM
 import GHC.Generics
+import GHC.Stack (HasCallStack)
 import NoThunks.Class
 import Ouroboros.Consensus.Block
 import Ouroboros.Consensus.Ledger.Abstract
@@ -167,21 +168,22 @@ newLSMLedgerTablesHandle ::
   , IndexedMemPack (l EmptyMK) (TxOut l)
   ) =>
   Tracer m LedgerDBV2Trace ->
-  ResourceRegistry m ->
   (ResourceKey m, UTxOTable m) ->
   m (LedgerTablesHandle m l)
-newLSMLedgerTablesHandle tracer rr (resKey, t) = do
+newLSMLedgerTablesHandle tracer (origResKey, t) = do
   traceWith tracer TraceLedgerTablesHandleCreate
+  tv <- newTVarIO origResKey
   pure
     LedgerTablesHandle
-      { close = implClose resKey
-      , duplicate = implDuplicate rr t tracer
+      { close = implClose tv
+      , duplicate = \rr -> implDuplicate rr t tracer
       , read = implRead t
       , readRange = implReadRange t
       , readAll = implReadAll t
       , pushDiffs = implPushDiffs t
       , takeHandleSnapshot = implTakeHandleSnapshot t
       , tablesSize = pure Nothing
+      , transfer = atomically . writeTVar tv
       }
 
 {-# INLINE implClose #-}
@@ -192,8 +194,9 @@ newLSMLedgerTablesHandle tracer rr (resKey, t) = do
 {-# INLINE implPushDiffs #-}
 {-# INLINE implTakeHandleSnapshot #-}
 
-implClose :: IOLike m => ResourceKey m -> m ()
-implClose = Monad.void . release
+implClose :: (HasCallStack, IOLike m) => StrictTVar m (ResourceKey m) -> m ()
+implClose tv =
+  Monad.void $ release =<< readTVarIO tv
 
 implDuplicate ::
   ( IOLike m
@@ -203,17 +206,17 @@ implDuplicate ::
   ResourceRegistry m ->
   UTxOTable m ->
   Tracer m LedgerDBV2Trace ->
-  m (LedgerTablesHandle m l)
+  m (ResourceKey m, LedgerTablesHandle m l)
 implDuplicate rr t tracer = do
-  table <-
+  (rk, table) <-
     allocate
       rr
       (\_ -> LSM.duplicate t)
       ( \t' -> do
           traceWith tracer TraceLedgerTablesHandleClose
           LSM.closeTable t'
       )
-  newLSMLedgerTablesHandle tracer rr table
+  (rk,) <$> newLSMLedgerTablesHandle tracer (rk, table)
 
 implRead ::
   forall m l.
@@ -461,7 +464,7 @@ loadSnapshot tracer rr ccfg fs session ds =
     case pointToWithOriginRealPoint (castPoint (getTip extLedgerSt)) of
       Origin -> throwE InitFailureGenesis
       NotOrigin pt -> do
-        values <-
+        (rk, values) <-
           lift $
             allocate
               rr
@@ -481,7 +484,7 @@ loadSnapshot tracer rr ccfg fs session ds =
           $ InitFailureRead
             ReadSnapshotDataCorruption
         (,pt)
-          <$> lift (empty extLedgerSt values (newLSMLedgerTablesHandle tracer rr))
+          <$> lift (empty extLedgerSt (rk, values) (newLSMLedgerTablesHandle tracer))
 
 -- | Create the initial LSM table from values, which should happen only at
 -- Genesis.
@@ -495,18 +498,16 @@ tableFromValuesMK ::
   LedgerTables l ValuesMK ->
   m (ResourceKey m, UTxOTable m)
 tableFromValuesMK tracer rr session st (LedgerTables (ValuesMK values)) = do
-  res@(_, table) <-
+  (rk, table) <-
     allocate
       rr
-      ( \_ ->
-          LSM.newTableWith (LSM.defaultTableConfig{LSM.confFencePointerIndex = LSM.OrdinaryIndex}) session
-      )
+      (\_ -> LSM.newTable session)
       ( \tb -> do
           traceWith tracer TraceLedgerTablesHandleClose
           LSM.closeTable tb
       )
   mapM_ (go table) $ chunks 1000 $ Map.toList values
-  pure res
+  pure (rk, table)
  where
   go table items =
     LSM.inserts table $
@@ -600,7 +601,7 @@ instance
   newHandleFromValues trcr reg res st = do
     table <-
       tableFromValuesMK trcr reg (sessionResource res) (forgetLedgerTables st) (ltprj st)
-    newLSMLedgerTablesHandle trcr reg table
+    newLSMLedgerTablesHandle trcr table
 
   snapshotManager _ res = Ouroboros.Consensus.Storage.LedgerDB.V2.LSM.snapshotManager (sessionResource res)
 
@@ -731,7 +732,7 @@ mkLSMYieldArgs fp snapName mkFS mkGen _ reg = do
             (LSM.SnapshotLabel $ T.pack "UTxO table")
       )
       LSM.closeTable
-  YieldLSM 1000 <$> newLSMLedgerTablesHandle nullTracer reg tb
+  YieldLSM 1000 <$> newLSMLedgerTablesHandle nullTracer tb
 
 -- | Create Sink arguments for LSM
 mkLSMSinkArgs ::

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

@@ -41,7 +41,7 @@ openMempool ::
   m (Mempool m blk)
 openMempool topLevelRegistry ledger cfg capacityOverride tracer = do
   env <- initMempoolEnv ledger cfg capacityOverride tracer topLevelRegistry
-  forkSyncStateOnTipPointChange topLevelRegistry env
+  forkSyncStateOnTipPointChange env
   return $ mkMempool env
 
 -- | Spawn a thread which syncs the 'Mempool' state whenever the 'LedgerState'
@@ -53,11 +53,10 @@ forkSyncStateOnTipPointChange ::
   , HasTxId (GenTx blk)
   , ValidateEnvelope blk
   ) =>
-  ResourceRegistry m ->
   MempoolEnv m blk ->
   m ()
-forkSyncStateOnTipPointChange topLevelRegistry menv = do
-  w <-
+forkSyncStateOnTipPointChange menv =
+  void $
     forkLinkedWatcher
       (mpEnvRegistry menv)
       "Mempool.syncStateOnTipPointChange"
@@ -67,12 +66,6 @@ forkSyncStateOnTipPointChange topLevelRegistry menv = do
         , wNotify = action
         , wReader = getCurrentTip
         }
-
-  -- With this allocation on the top level registry, we make sure that we first
-  -- stop the watcher thread before closing the mempool registry, as otherwise
-  -- we would run into a race condition (the thread might try to re-sync and
-  -- allocate a forker on the mempool registry which would be closing down).
-  void $ allocate topLevelRegistry (\_ -> pure w) cancelThread
  where
   action :: MempoolLedgerDBView m blk -> m ()
   action _a =