Label threads, queues and vars (#2172)

#### Motivation

Labels make it easier to track the origin and lifecycle of threads,
queues, and transactional variables, especially when used in simulations
(e.g. `IOSim`) or debugging tools.

#### 🔄 Changes

This PR introduces **labelled variants** of common concurrency
primitives and combinators to improve observability and debugging.

Replaced existing constructors/combinators with labelled versions:

* **STM / TVars / Queues**

  * `newTVar`         → `newLabelledTVar`
  * `newTVarIO`       → `newLabelledTVarIO`
  * `newEmptyTMVar`   → `newLabelledEmptyTMVar`
  * `newEmptyTMVarIO` → `newLabelledEmptyTMVarIO`
  * `newTQueue`       → `newLabelledTQueue`
  * `newTQueueIO`     → `newLabelledTQueueIO`
  * `newTBQueue`      → `newLabelledTBQueue`
  * `newTBQueueIO`    → `newLabelledTBQueueIO`

* **Async / Thread combinators**
  (based on `Control.Monad.Class.MonadFork.labelThisThread`)

  * `race`            → `raceLabelled`
  * `race_`           → `raceLabelled_`
  * `withAsync`       → `withAsyncLabelled`
  * `async`           → `asyncLabelled`
  * `concurrently`    → `concurrentlyLabelled`
  * `concurrently_`   → `concurrentlyLabelled_`

#### 🔍 Usage not found

* `newTMVar`, `newTMVarIO`
* `forkIO` and its variants

#### ⚠️ Unresolved

* `newEmptyMVar`
* `newMVar`

> **Reason:** Labelled versions exist in
>
> * [`MonadMVar`
docs](https://hackage.haskell.org/package/io-classes-1.8.0.1/docs/Control-Concurrent-Class-MonadMVar.html)
> * [`io-classes`
changelog](https://github.com/input-output-hk/io-sim/blob/aadd3602b9b75640b042a02e0e62d10dd399b610/io-classes/CHANGELOG.md?plain=1#L23)
>
> However, their `IO` implementations are no-ops (labels are only
effective under `IOSim`).
> We need to decide whether to wrap these in our codebase or leave them
unlabelled for now.

---

<!-- Consider each and tick it off one way or the other -->
* [X] CHANGELOG updated or not needed
* [X] Documentation updated or not needed
* [X] Haddocks updated or not needed
* [X] No new TODOs introduced or explained herafter

Author: noonio

CHANGELOG.md

@@ -65,12 +65,13 @@ changes.
 
 - **BREAKING** Rename `--script-info` option to `--hydra-script-catalogue` in the `hydra-node` CLI.
 
-Fix rotation log id consistency after restart by changing the rotation check to trigger only
+- Fix rotation log id consistency after restart by changing the rotation check to trigger only
 when the number of persisted `StateChanged` events exceeds the configured `--persistence-rotate-after` threshold.
   * This also prevents immediate rotation on startup when the threshold is set to 1.
   * `Checkpoint` event ids now match the suffix of their preceding rotated log file and the last `StateChanged` event id within it,
   preserving sequential order and making it easier to identify which rotated log file was used to compute it.
 
+- Label threads, queues and vars.
 
 ## [0.22.2] - 2025-06-30
 

hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs

@@ -11,7 +11,6 @@ import Blockfrost.Client (
 import Blockfrost.Client qualified as Blockfrost
 import Control.Concurrent.Class.MonadSTM (
   MonadSTM (readTVarIO),
-  newTVarIO,
   writeTVar,
  )
 import Control.Retry (RetryPolicyM, RetryStatus, constantDelay, retrying)
@@ -96,7 +95,7 @@ blockfrostClient tracer projectPath blockConfirmations = do
 
           let blockHash = fromChainPoint chainPoint genesisBlockHash
 
-          stateTVar <- newTVarIO (blockHash, mempty)
+          stateTVar <- newLabelledTVarIO "blockfrost-client-state" (blockHash, mempty)
           void $
             retrying (retryPolicy blockTime) shouldRetry $ \_ -> do
               loop tracer prj networkId blockTime observerHandler blockConfirmations stateTVar

hydra-cluster/bench/Bench/EndToEnd.hs

@@ -12,8 +12,6 @@ import Control.Concurrent.Class.MonadSTM (
   check,
   lengthTBQueue,
   modifyTVar,
-  newTBQueueIO,
-  newTVarIO,
   tryReadTBQueue,
   writeTBQueue,
   writeTVar,
@@ -83,7 +81,7 @@ bench startingNodeId timeoutSeconds workDir dataset = do
         putTextLn "Starting benchmark"
         let cardanoKeys = hydraNodeKeys dataset <&> \sk -> (getVerificationKey sk, sk)
         let hydraKeys = generateSigningKey . show <$> [1 .. toInteger (length cardanoKeys)]
-        statsTvar <- newTVarIO mempty
+        statsTvar <- newLabelledTVarIO "bench-stats" mempty
         scenarioData <- withOSStats workDir statsTvar $
           withCardanoNodeDevnet (contramap FromCardanoNode tracer) workDir $ \_ backend -> do
             let nodeSocket' = case Backend.getOptions backend of
@@ -261,13 +259,14 @@ withOSStats workDir tvar action =
     Nothing -> action
     Just _ ->
       withCreateProcess process{std_out = CreatePipe} $ \_stdin out _stderr _processHandle ->
-        race
-          ( do
+        raceLabelled
+          ( "os-stats-collect"
+          , do
               -- Write the header
               atomically $ writeTVar tvar [" | Time | Used | Free | ", "|------------------------------------|------|------|"]
               collectStats tvar out
           )
-          action
+          ("os-stats-action", action)
           >>= \case
             Left _ -> failure "dool process failed unexpectedly"
             Right a -> pure a
@@ -384,12 +383,15 @@ processTransactions clients clientDatasets = do
 
   clientProcessDataset (ClientDataset{txSequence}, client) clientId = do
     let numberOfTxs = length txSequence
-    submissionQ <- newTBQueueIO (fromIntegral numberOfTxs)
+    submissionQ <- newLabelledTBQueueIO "submission" (fromIntegral numberOfTxs)
     registry <- newRegistry
     atomically $ forM_ txSequence $ writeTBQueue submissionQ
-    ( submitTxs client registry submissionQ
-        `concurrently_` waitForAllConfirmations client registry (Set.fromList $ map txId txSequence)
-        `concurrently_` progressReport (hydraNodeId client) clientId numberOfTxs submissionQ
+    concurrentlyLabelled_
+      ("submit-txs", submitTxs client registry submissionQ)
+      ( "confirm-txs"
+      , concurrentlyLabelled_
+          ("wait-for-all-confirmations", waitForAllConfirmations client registry (Set.fromList $ map txId txSequence))
+          ("progress-report", progressReport (hydraNodeId client) clientId numberOfTxs submissionQ)
       )
       `catch` \(HUnitFailure sourceLocation reason) ->
         putStrLn ("Something went wrong while waiting for all confirmations: " <> formatLocation sourceLocation <> ": " <> formatFailureReason reason)
@@ -443,8 +445,8 @@ data Registry tx = Registry
 newRegistry ::
   IO (Registry Tx)
 newRegistry = do
-  processedTxs <- newTVarIO mempty
-  latestSnapshot <- newTVarIO 0
+  processedTxs <- newLabelledTVarIO "registry-processed-txs" mempty
+  latestSnapshot <- newLabelledTVarIO "registry-latest-snapshot" 0
   pure $ Registry{processedTxs, latestSnapshot}
 
 submitTxs ::

hydra-cluster/src/CardanoNode.hs

@@ -322,7 +322,9 @@ withCardanoNode tr stateDirectory args action = do
     withCreateProcess process{std_out = UseHandle out, std_err = CreatePipe} $
       \_stdin _stdout mError processHandle ->
         (`finally` cleanupSocketFile) $
-          race (checkProcessHasNotDied "cardano-node" processHandle mError) waitForNode
+          raceLabelled
+            ("check-cardano-node-process-not-died", checkProcessHasNotDied "cardano-node" processHandle mError)
+            ("wait-for-node", waitForNode)
             <&> either absurd id
  where
   CardanoNodeArgs{nodeSocket} = args

hydra-cluster/src/Hydra/Cluster/Scenarios.hs

@@ -1896,9 +1896,9 @@ threeNodesWithMirrorParty tracer workDir backend hydraScriptsTxId = do
         -- N1 & N3 commit the same thing at the same time
         -- XXX: one will fail but the head will still open
         aliceUTxO <- seedFromFaucet backend aliceCardanoVk 1_000_000 (contramap FromFaucet tracer)
-        race_
-          (requestCommitTx n1 aliceUTxO >>= Backend.submitTransaction backend)
-          (requestCommitTx n3 aliceUTxO >>= Backend.submitTransaction backend)
+        raceLabelled_
+          ("request-commit-tx-n1", requestCommitTx n1 aliceUTxO >>= Backend.submitTransaction backend)
+          ("request-commit-tx-n3", requestCommitTx n3 aliceUTxO >>= Backend.submitTransaction backend)
 
         -- N2 commits something
         bobUTxO <- seedFromFaucet backend bobCardanoVk 1_000_000 (contramap FromFaucet tracer)

hydra-cluster/src/HydraNode.hs

@@ -7,7 +7,7 @@ import Hydra.Prelude hiding (STM, delete)
 
 import CardanoNode (cliQueryProtocolParameters)
 import Control.Concurrent.Async (forConcurrently_)
-import Control.Concurrent.Class.MonadSTM (modifyTVar', newTVarIO, readTVarIO)
+import Control.Concurrent.Class.MonadSTM (modifyTVar', readTVarIO)
 import Control.Exception (Handler (..), IOException, catches)
 import Control.Lens ((?~))
 import Control.Monad.Class.MonadAsync (forConcurrently)
@@ -105,7 +105,7 @@ waitNoMatch delay client match = do
 -- | Wait up to some time for an API server output to match the given predicate.
 waitMatch :: HasCallStack => NominalDiffTime -> HydraClient -> (Aeson.Value -> Maybe a) -> IO a
 waitMatch delay client@HydraClient{tracer, hydraNodeId} match = do
-  seenMsgs <- newTVarIO []
+  seenMsgs <- newLabelledTVarIO "wait-match-seen-msgs" []
   timeout (realToFrac delay) (go seenMsgs) >>= \case
     Just x -> pure x
     Nothing -> do
@@ -453,9 +453,9 @@ withPreparedHydraNode tracer workDir hydraNodeId runOptions action =
 
     withProcessTerm cmd $ \p -> do
       -- NOTE: exit code thread gets cancelled if 'action' terminates first
-      race
-        (collectAndCheckExitCode p)
-        (withConnectionToNode tracer hydraNodeId action)
+      raceLabelled
+        ("collect-check-process-exit-code", collectAndCheckExitCode p)
+        ("with-connection-to-node", withConnectionToNode tracer hydraNodeId action)
         <&> either absurd id
  where
   collectAndCheckExitCode p = do

hydra-cluster/test/Test/BlockfrostChainSpec.hs

@@ -6,7 +6,7 @@ import Hydra.Prelude
 import Test.Hydra.Prelude
 
 import Cardano.Api.UTxO qualified as UTxO
-import Control.Concurrent.STM (newEmptyTMVarIO, takeTMVar)
+import Control.Concurrent.STM (takeTMVar)
 import Control.Concurrent.STM.TMVar (putTMVar)
 import Control.Exception (IOException)
 import Hydra.Chain (
@@ -164,7 +164,7 @@ withBlockfrostChainTest tracer config party action = do
           _ -> failure $ "unexpected chainBackendOptions: " <> show chainBackendOptions
       otherConfig -> failure $ "unexpected chainConfig: " <> show otherConfig
   ctx <- loadChainContext backend configuration party
-  eventMVar <- newEmptyTMVarIO
+  eventMVar <- newLabelledEmptyTMVarIO "blockfrost-chain-events"
 
   let callback event = atomically $ putTMVar eventMVar event
 

hydra-cluster/test/Test/ChainObserverSpec.hs

@@ -11,7 +11,7 @@ import Test.Hydra.Prelude
 
 import Cardano.Api.UTxO qualified as UTxO
 import CardanoNode (NodeLog, withCardanoNodeDevnet)
-import Control.Concurrent.Class.MonadSTM (modifyTVar', newTVarIO, readTVarIO)
+import Control.Concurrent.Class.MonadSTM (modifyTVar', readTVarIO)
 import Control.Lens ((^?))
 import Data.Aeson as Aeson
 import Data.Aeson.Lens (key, _JSON, _String)
@@ -114,7 +114,7 @@ chainObserverSees observer txType headId =
 
 awaitMatch :: HasCallStack => ChainObserverHandle -> DiffTime -> (Aeson.Value -> Maybe a) -> IO a
 awaitMatch chainObserverHandle delay f = do
-  seenMsgs <- newTVarIO []
+  seenMsgs <- newLabelledTVarIO "await-match-seen-msgs" []
   timeout delay (go seenMsgs) >>= \case
     Just x -> pure x
     Nothing -> do

hydra-cluster/test/Test/DirectChainSpec.hs

@@ -12,7 +12,7 @@ import CardanoClient (
   waitForUTxO,
  )
 import CardanoNode (NodeLog, withCardanoNodeDevnet)
-import Control.Concurrent.STM (newEmptyTMVarIO, takeTMVar)
+import Control.Concurrent.STM (takeTMVar)
 import Control.Concurrent.STM.TMVar (putTMVar)
 import Control.Lens ((<>~))
 import Data.List.Split (splitWhen)
@@ -548,7 +548,7 @@ withDirectChainTest tracer config party action = do
           _ -> failure $ "unexpected chainBackendOptions: " <> show chainBackendOptions
       otherConfig -> failure $ "unexpected chainConfig: " <> show otherConfig
   ctx <- loadChainContext backend configuration party
-  eventMVar <- newEmptyTMVarIO
+  eventMVar <- newLabelledEmptyTMVarIO "direct-chain-events"
 
   let callback event = atomically $ putTMVar eventMVar event
 

hydra-cluster/test/Test/EndToEndSpec.hs

@@ -676,9 +676,9 @@ spec = around (showLogsOnFailure "EndToEndSpec") $ do
           withClusterTempDir $ \tmpDir -> do
             withBackend (contramap FromCardanoNode tracer) tmpDir $ \_ backend -> do
               hydraScriptsTxId <- publishHydraScriptsAs backend Faucet
-              concurrently_
-                (initAndClose tmpDir tracer 0 hydraScriptsTxId backend)
-                (initAndClose tmpDir tracer 1 hydraScriptsTxId backend)
+              concurrentlyLabelled_
+                ("init-and-close-0", initAndClose tmpDir tracer 0 hydraScriptsTxId backend)
+                ("init-and-close-1", initAndClose tmpDir tracer 1 hydraScriptsTxId backend)
 
       it "alice inits a Head with incorrect keys preventing bob from observing InitTx" $ \tracer ->
         failAfter 60 $