Implement the GDD governor (#1015)

This PR builds upon #979 to provide the GDD governor, the component
responsible for identifying and disconnecting peers offering sparser
chains than the best. This has the effect of unblocking the Limit on
Eagerness, since removing disagreeing peers allows the current selection
to advance.

The GDD governor runs in a background thread and evaluates candidate
chains whenever they change, or whenever a peer claims to have no more
headers, or whenever a peer starts sending headers beyond the forecast
horizon.

Whenever GDD disconnects peers, the chain selection is updated.

The implementation of the GDD governor is in commit
3fc6d28. A range of supporting changes
is provided as well.

-------------
Peer simulator improvements in f02d8df
include:
* Our simulator of peers now kills all miniprotocol threads whenever one
of the miniprotocols is terminated. The GDD governor might terminate a
ChainSync client, and this termination should also cause the termination
of the BlockFetch client.
* Our tests now also check that no threads are left lingering when the
peer simulator is terminated.
* Traces now print a timestamp when events occur between the start of
consecutive ticks. This need arose when Limit on Patience started
disconnecting peers because of exhausted buckets, and these events could
happen at any moment between ticks. See [this
comment](#990 (comment))
of #990 for an example trace.

-------------
Post-GDD improvements in ea12e12
include:
* Started tracing the information about BlockFetch timeouts and
BlockFetch logic decisions. This information became relevant when we had
to analyze why blocks wouldn't be downloaded from the honest peer, or
why a quiet adversarial peer wasn't being disconnected.
* Have the Limit on Eagerness recalculated when the change selection
changes. Limit on Eagerness does not depend on the chain selection, but
this mechanism has the desired side effect of retriggering chain
selection in succession until no more blocks are added to it. This
change is necessary after chain selection stopped being idempotent in
7ace045 (later on this same PR). This
change is included here to demonstrate that the GDD governor prevents
the node from being leashed, but otherwise we intend to recover
idempotency of chain selection in a later PR.

-------------
Tests for the GDD in b8af4b4 include:
* Added a new test to check that GDD actually disconnects nodes in small
examples. This test is subsumed by the already existing test "serve
adversarial branches", but it tries only small peer schedules which are
easier to approach on test failures.
* Enabled one of the leashing attack tests. This test was waiting for
the GDD governor to be implemented. It generates an honest peer schedule
for every peer, and then drops from the adversary schedules some ticks
chosen at random. This has the effect of having the adversaries
introduce delays to send headers or blocks that shouldn't stop the node
from reaching the honest tip by the end of the test.

There is one other leashing attack test called "time limited leashing
attack", which it is enabled in [a follow up
PR](#1035).

-------------
The classifiers added in cfefa6c keep
track of whether peers are being disconnected by the GDD governor, and
the LoP, and timeouts. It also keep tracks of adversaries who did a
rollback of their chain. The commit also implements a check ensuring no
peer is killed because of an unforeseen exception.

-------------
Chain generators implement the extended praos chain growth assumption
(EPCG) in d8c6546. This causes the
honest chain to have at least `k+1` blocks in every stability window.
The former Praos Chain Growth assumption required at least `k`, and in
practice chains with less than `k+1` blocks in a stability window are
still very rare. The EPCG assumption makes a difference to testing the
GDD governor.

We cannot let the almost caught up GDD governor disconnect honest peers
which disagree on a few final blocks. Therefore, we ask the governor to
require a disagreement of more than `k` blocks before deciding a
disconnection. But asking for more than `k` blocks without the EPCG
assumption means that the GDD governor might not be able to make
progress if the `k+1`st block is after the first stability window after
the intersection and after the forecast horizon. In this case, the
candidate fragments wouldn't be extended with the `k+1`st block, and the
GDD governor wouldn't realize that there is a discrepancy of more than
`k` blocks.

-------------
In c8468c2 we don't allow the chain
selection to fork more than k blocks from the honest chain. The honest
chain is considered to be the one containing the first intersection at
which candidate fragments fork into different chains.

It used to be the case that we allowed the selection to choose up to k
blocks after the intersection. But it turned out that if the selection
tip is still X blocks behind the intersection, allowing chain selection
to pick `X+k` blocks, there is no guarantee that the selection will only
add blocks from the candidate fragments. It could select more than `k`
blocks of an older fork in the chain DB, and this is catastrophic
because now the immutable tip would leave the honest chain.

-------------
bf3f09a refactors and properly bundles
the various state that a ChainSync client maintains.

ouroboros-consensus-cardano/changelog.d/20240328_165135_alexander.esgen_milestone_11_gdd_governor_squashed.md

@@ -0,0 +1,7 @@
+### Non-Breaking
+
+- Set Genesis window for Byron and Shelley.
+
+- Bump to `HardForkSpecificNodeToClientVersion3` for
+  `CardanoNodeToClientVersion12` to account for serialisation change of
+  `EraParams`.

ouroboros-consensus-cardano/src/byron/Ouroboros/Consensus/Byron/Ledger/Ledger.hs

@@ -43,6 +43,7 @@ module Ouroboros.Consensus.Byron.Ledger.Ledger (
 
 import qualified Cardano.Chain.Block as CC
 import qualified Cardano.Chain.Byron.API as CC
+import qualified Cardano.Chain.Common as Gen
 import qualified Cardano.Chain.Genesis as Gen
 import qualified Cardano.Chain.Update as Update
 import qualified Cardano.Chain.Update.Validation.Endorsement as UPE
@@ -278,6 +279,7 @@ byronEraParams genesis = HardFork.EraParams {
       eraEpochSize  = fromByronEpochSlots $ Gen.configEpochSlots genesis
     , eraSlotLength = fromByronSlotLength $ genesisSlotLength genesis
     , eraSafeZone   = HardFork.StandardSafeZone (2 * k)
+    , eraGenesisWin = GenesisWindow (2 * k)
     }
   where
     SecurityParam k = genesisSecurityParam genesis
@@ -288,6 +290,7 @@ byronEraParamsNeverHardForks genesis = HardFork.EraParams {
       eraEpochSize  = fromByronEpochSlots $ Gen.configEpochSlots genesis
     , eraSlotLength = fromByronSlotLength $ genesisSlotLength genesis
     , eraSafeZone   = HardFork.UnsafeIndefiniteSafeZone
+    , eraGenesisWin = GenesisWindow (2 * Gen.unBlockCount (Gen.configK genesis))
     }
 
 instance HasHardForkHistory ByronBlock where

ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Node.hs

@@ -447,7 +447,7 @@ pattern CardanoNodeToClientVersion11 =
 pattern CardanoNodeToClientVersion12 :: BlockNodeToClientVersion (CardanoBlock c)
 pattern CardanoNodeToClientVersion12 =
     HardForkNodeToClientEnabled
-      HardForkSpecificNodeToClientVersion2
+      HardForkSpecificNodeToClientVersion3
       (  EraNodeToClientEnabled ByronNodeToClientVersion1
       :* EraNodeToClientEnabled ShelleyNodeToClientVersion8
       :* EraNodeToClientEnabled ShelleyNodeToClientVersion8

ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Ledger.hs

@@ -134,6 +134,7 @@ shelleyEraParams genesis = HardFork.EraParams {
       eraEpochSize  = SL.sgEpochLength genesis
     , eraSlotLength = mkSlotLength $ SL.fromNominalDiffTimeMicro $ SL.sgSlotLength genesis
     , eraSafeZone   = HardFork.StandardSafeZone stabilityWindow
+    , eraGenesisWin = GenesisWindow stabilityWindow
     }
   where
     stabilityWindow =
@@ -147,7 +148,13 @@ shelleyEraParamsNeverHardForks genesis = HardFork.EraParams {
       eraEpochSize  = SL.sgEpochLength genesis
     , eraSlotLength = mkSlotLength $ SL.fromNominalDiffTimeMicro $ SL.sgSlotLength genesis
     , eraSafeZone   = HardFork.UnsafeIndefiniteSafeZone
+    , eraGenesisWin = GenesisWindow stabilityWindow
     }
+  where
+    stabilityWindow =
+        SL.computeStabilityWindow
+          (SL.sgSecurityParam genesis)
+          (SL.sgActiveSlotCoeff genesis)
 
 mkShelleyLedgerConfig ::
      SL.ShelleyGenesis (EraCrypto era)

ouroboros-consensus-cardano/src/unstable-cardano-testlib/Test/Consensus/Cardano/Generators.hs

@@ -554,6 +554,7 @@ instance Arbitrary History.EraParams where
       <$> (EpochSize <$> arbitrary)
       <*> arbitrary
       <*> arbitrary
+      <*> (GenesisWindow <$> arbitrary)
 
 instance Arbitrary History.EraSummary where
   arbitrary = History.EraSummary
@@ -624,9 +625,7 @@ instance c ~ MockCryptoCompatByron
       , (1, WithVersion
               <$> (getHardForkEnabledNodeToClientVersion <$> arbitrary)
               <*> genQueryAnytimeResultConway)
-      , (1, WithVersion
-              <$> (getHardForkEnabledNodeToClientVersion <$> arbitrary)
-              <*> genQueryHardForkResult)
+      , (1, genQueryHardForkResult)
       ]
     where
       injByron   (SomeResult q r) = SomeResult (QueryIfCurrentByron   q) (QueryResultSuccess r)
@@ -695,8 +694,15 @@ instance c ~ MockCryptoCompatByron
       genQueryAnytimeResultConway =
           SomeResult (QueryAnytimeConway GetEraStart) <$> arbitrary
 
-      genQueryHardForkResult :: Gen (SomeResult (CardanoBlock c))
+      genQueryHardForkResult ::
+           Gen (WithVersion (HardForkNodeToClientVersion (CardanoEras c))
+                            (SomeResult (CardanoBlock c)))
       genQueryHardForkResult = oneof
-          [ SomeResult (QueryHardFork GetInterpreter) <$> arbitrary
-          , SomeResult (QueryHardFork GetCurrentEra)  <$> arbitrary
+          [ WithVersion
+              <$> genWithHardForkSpecificNodeToClientVersion
+                    (>= HardForkSpecificNodeToClientVersion3)
+              <*> (SomeResult (QueryHardFork GetInterpreter) <$> arbitrary)
+          , WithVersion
+              <$> (getHardForkEnabledNodeToClientVersion <$> arbitrary)
+              <*> (SomeResult (QueryHardFork GetCurrentEra) <$> arbitrary)
           ]

ouroboros-consensus-diffusion/changelog.d/20240328_165134_alexander.esgen_milestone_11_gdd_governor_squashed.md

@@ -0,0 +1,3 @@
+### Breaking
+
+- Accounted for a refactoring of the ChainSync client parameters.

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

@@ -229,6 +229,7 @@ test-suite consensus-test
     Test.Consensus.Genesis.Setup.Classifiers
     Test.Consensus.Genesis.Setup.GenChains
     Test.Consensus.Genesis.Tests
+    Test.Consensus.Genesis.Tests.DensityDisconnect
     Test.Consensus.Genesis.Tests.LoE
     Test.Consensus.Genesis.Tests.LoP
     Test.Consensus.Genesis.Tests.LongRangeAttack
@@ -238,7 +239,6 @@ test-suite consensus-test
     Test.Consensus.HardFork.Combinator.B
     Test.Consensus.IOSimQSM.Test.StateMachine.Sequential
     Test.Consensus.Network.AnchoredFragment.Extras
-    Test.Consensus.Network.Driver.Limits.Extras
     Test.Consensus.Node
     Test.Consensus.PeerSimulator.BlockFetch
     Test.Consensus.PeerSimulator.ChainSync
@@ -252,6 +252,7 @@ test-suite consensus-test
     Test.Consensus.PeerSimulator.StateDiagram
     Test.Consensus.PeerSimulator.StateView
     Test.Consensus.PeerSimulator.Tests
+    Test.Consensus.PeerSimulator.Tests.LinkedThreads
     Test.Consensus.PeerSimulator.Tests.Rollback
     Test.Consensus.PeerSimulator.Tests.Timeouts
     Test.Consensus.PeerSimulator.Trace
@@ -304,6 +305,5 @@ test-suite consensus-test
     time,
     tree-diff,
     typed-protocols,
-    typed-protocols-examples,
     unstable-diffusion-testlib,
     vector,

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs

@@ -57,6 +57,8 @@ import           Ouroboros.Consensus.Block
 import           Ouroboros.Consensus.Ledger.SupportsMempool
 import           Ouroboros.Consensus.Ledger.SupportsProtocol
 import           Ouroboros.Consensus.MiniProtocol.BlockFetch.Server
+import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
+                     (ChainSyncStateView (..))
 import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client as CsClient
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Server
 import           Ouroboros.Consensus.Node.ExitPolicy
@@ -567,12 +569,11 @@ mkApps kernel Tracers {..} mkCodecs ByteLimits {..} genChainSyncTimeout lopBucke
         CsClient.bracketChainSyncClient
             (contramap (TraceLabelPeer them) (Node.chainSyncClientTracer (getTracers kernel)))
             (CsClient.defaultChainDbView (getChainDB kernel))
-            (getNodeCandidates kernel)
-            (getNodeIdlers     kernel)
+            (getChainSyncHandles kernel)
             them
             version
             lopBucketConfig
-            $ \varCandidate (startIdling, stopIdling) (pauseLoPBucket, resumeLoPBucket, grantLoPToken) -> do
+            $ \csState -> do
               chainSyncTimeout <- genChainSyncTimeout
               (r, trailing) <-
                 runPipelinedPeerWithLimits
@@ -589,12 +590,10 @@ mkApps kernel Tracers {..} mkCodecs ByteLimits {..} genChainSyncTimeout lopBucke
                           CsClient.version
                         , CsClient.controlMessageSTM
                         , CsClient.headerMetricsTracer = TraceLabelPeer them `contramap` reportHeader
-                        , CsClient.varCandidate
-                        , CsClient.startIdling
-                        , CsClient.stopIdling
-                        , CsClient.pauseLoPBucket
-                        , CsClient.resumeLoPBucket
-                        , CsClient.grantLoPToken
+                        , CsClient.setCandidate = csvSetCandidate csState
+                        , CsClient.idling = csvIdling csState
+                        , CsClient.loPBucket = csvLoPBucket csState
+                        , CsClient.setLatestSlot = csvSetLatestSlot csState
                         }
               return (ChainSyncInitiatorResult r, trailing)
 

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

@@ -37,7 +37,6 @@ import qualified Control.Monad.Class.MonadTimer.SI as SI
 import           Control.Tracer (Tracer, traceWith)
 import           Data.Functor ((<&>))
 import qualified Data.Map.Strict as Map
-import qualified Data.Set as Set
 import           Data.Time (NominalDiffTime)
 import qualified Ouroboros.Consensus.BlockchainTime.WallClock.Types as Clock
 import qualified Ouroboros.Consensus.HardFork.Abstract as HardFork
@@ -94,7 +93,7 @@ data GsmState =
     -- ^ We are caught-up.
   deriving (Eq, Show, Read)
 
-data GsmView m upstreamPeer selection candidate = GsmView {
+data GsmView m upstreamPeer selection chainSyncState = GsmView {
     antiThunderingHerd        :: Maybe StdGen
     -- ^ An initial seed used to randomly increase 'minCaughtUpDuration' by up
     -- to 15% every transition from Syncing to CaughtUp, in order to avoid a
@@ -103,7 +102,9 @@ data GsmView m upstreamPeer selection candidate = GsmView {
     -- 'Nothing' should only be used for testing.
   ,
     candidateOverSelection    ::
-        selection -> candidate -> CandidateVersusSelection
+        selection -> chainSyncState -> CandidateVersusSelection
+  ,
+    peerIsIdle                :: chainSyncState -> Bool
   ,
     durationUntilTooOld       :: Maybe (selection -> m DurationFromNow)
     -- ^ How long from now until the selection will be so old that the node
@@ -115,13 +116,10 @@ data GsmView m upstreamPeer selection candidate = GsmView {
     -- ^ Whether the two selections are equivalent for the purpose of the
     -- Genesis State Machine
   ,
-    getChainSyncCandidates    ::
-        STM m (Map.Map upstreamPeer (StrictTVar m candidate))
-    -- ^ The latest candidates from the upstream ChainSync peers
-  ,
-    getChainSyncIdlers        :: STM m (Set.Set upstreamPeer)
-    -- ^ The ChainSync peers whose latest message claimed that they have no
-    -- subsequent headers
+    getChainSyncStates        ::
+        STM m (Map.Map upstreamPeer (StrictTVar m chainSyncState))
+    -- ^ The current ChainSync state with the latest candidates from the
+    -- upstream peers
   ,
     getCurrentSelection       :: STM m selection
     -- ^ The node's current selection
@@ -224,7 +222,6 @@ gsmStateToLedgerJudgement = \case
 realGsmEntryPoints :: forall m upstreamPeer selection tracedSelection candidate.
      ( SI.MonadDelay m
      , SI.MonadTimer m
-     , Eq upstreamPeer
      )
   => (selection -> tracedSelection, Tracer m (TraceGsmEvent tracedSelection))
   -> GsmView m upstreamPeer selection candidate
@@ -241,14 +238,14 @@ realGsmEntryPoints tracerArgs gsmView = GsmEntryPoints {
         antiThunderingHerd
       ,
         candidateOverSelection
+      ,
+        peerIsIdle
       ,
         durationUntilTooOld
       ,
         equivalent
       ,
-        getChainSyncCandidates
-      ,
-        getChainSyncIdlers
+        getChainSyncStates
       ,
         getCurrentSelection
       ,
@@ -374,12 +371,11 @@ realGsmEntryPoints tracerArgs gsmView = GsmEntryPoints {
     blockUntilCaughtUp :: STM m (TraceGsmEvent tracedSelection)
     blockUntilCaughtUp = do
         -- STAGE 1: all ChainSync clients report no subsequent headers
-        idlers        <- getChainSyncIdlers
-        varsCandidate <- getChainSyncCandidates
+        varsState     <- getChainSyncStates
+        states        <- traverse StrictSTM.readTVar varsState
         check $
-                           0  < Map.size    varsCandidate
-          && Set.size idlers == Map.size    varsCandidate
-          &&          idlers == Map.keysSet varsCandidate
+             not (Map.null states)
+          && all peerIsIdle states
 
         -- STAGE 2: no candidate is better than the node's current
         -- selection
@@ -392,14 +388,14 @@ realGsmEntryPoints tracerArgs gsmView = GsmEntryPoints {
         -- block; general Praos reasoning ensures that won't take particularly
         -- long.
         selection  <- getCurrentSelection
-        candidates <- traverse StrictSTM.readTVar varsCandidate
+        candidates <- traverse StrictSTM.readTVar varsState
         let ok candidate =
                 WhetherCandidateIsBetter False
              == candidateOverSelection selection candidate
         check $ all ok candidates
 
         pure $ GsmEventEnterCaughtUp
-            (Set.size idlers)
+            (Map.size states)
             (cnvSelection selection)
 
         -- STAGE 3: the previous stages weren't so slow that the idler

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs

@@ -45,7 +45,6 @@ import           Data.List.NonEmpty (NonEmpty)
 import           Data.Map.Strict (Map)
 import           Data.Maybe (isJust, mapMaybe)
 import           Data.Proxy
-import           Data.Set (Set)
 import qualified Data.Text as Text
 import           Data.Void (Void)
 import           Ouroboros.Consensus.Block hiding (blockMatchesHeader)
@@ -61,6 +60,9 @@ import           Ouroboros.Consensus.Ledger.SupportsPeerSelection
 import           Ouroboros.Consensus.Ledger.SupportsProtocol
 import           Ouroboros.Consensus.Mempool
 import qualified Ouroboros.Consensus.MiniProtocol.BlockFetch.ClientInterface as BlockFetchClientInterface
+import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
+                     (ChainSyncClientHandle (..), ChainSyncState (..),
+                     viewChainSyncState)
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
                      (SomeHeaderInFutureCheck)
 import           Ouroboros.Consensus.Node.GSM (GsmNodeKernelArgs (..))
@@ -130,12 +132,8 @@ data NodeKernel m addrNTN addrNTC blk = NodeKernel {
       --
     , getLedgerStateJudgement :: STM m LedgerStateJudgement
 
-      -- | Read the current candidates
-    , getNodeCandidates       :: StrictTVar m (Map (ConnectionId addrNTN) (StrictTVar m (AnchoredFragment (Header blk))))
-
-      -- | Read the set of peers that have claimed to have no subsequent
-      -- headers beyond their current candidate
-    , getNodeIdlers           :: StrictTVar m (Set (ConnectionId addrNTN))
+      -- | The kill handle and exposed state for each ChainSync client.
+    , getChainSyncHandles     :: StrictTVar m (Map (ConnectionId addrNTN) (ChainSyncClientHandle m blk))
 
       -- | Read the current peer sharing registry, used for interacting with
       -- the PeerSharing protocol
@@ -202,8 +200,7 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
           , fetchClientRegistry
           , mempool
           , peerSharingRegistry
-          , varCandidates
-          , varIdlers
+          , varChainSyncHandles
           , varLedgerJudgement
           } = st
 
@@ -215,23 +212,23 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
 
         let gsm = GSM.realGsmEntryPoints gsmTracerArgs GSM.GsmView
               { GSM.antiThunderingHerd        = Just gsmAntiThunderingHerd
-              , GSM.candidateOverSelection    = \(headers, _lst) candidate ->
-                    case AF.intersectionPoint headers candidate of
+              , GSM.candidateOverSelection    = \(headers, _lst) state ->
+                    case AF.intersectionPoint headers (csCandidate state) of
                         Nothing -> GSM.CandidateDoesNotIntersect
                         Just{}  ->
                             GSM.WhetherCandidateIsBetter
                           $ -- precondition requires intersection
                             preferAnchoredCandidate
                                 (configBlock cfg)
                                 headers
-                                candidate
+                                (csCandidate state)
+              , GSM.peerIsIdle                = csIdling
               , GSM.durationUntilTooOld       =
                       gsmDurationUntilTooOld
                   <&> \wd (_headers, lst) ->
                         GSM.getDurationUntilTooOld wd (getTipSlot lst)
               , GSM.equivalent                = (==) `on` (AF.headPoint . fst)
-              , GSM.getChainSyncCandidates    = readTVar varCandidates
-              , GSM.getChainSyncIdlers        = readTVar varIdlers
+              , GSM.getChainSyncStates        = fmap cschState <$> readTVar varChainSyncHandles
               , GSM.getCurrentSelection       = do
                   headers        <- ChainDB.getCurrentChain  chainDB
                   extLedgerState <- ChainDB.getCurrentLedger chainDB
@@ -276,8 +273,7 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
       , getFetchClientRegistry  = fetchClientRegistry
       , getFetchMode            = readFetchMode blockFetchInterface
       , getLedgerStateJudgement = readTVar varLedgerJudgement
-      , getNodeCandidates       = varCandidates
-      , getNodeIdlers           = varIdlers
+      , getChainSyncHandles     = varChainSyncHandles
       , getPeerSharingRegistry  = peerSharingRegistry
       , getTracers              = tracers
       , setBlockForging         = \a -> atomically . LazySTM.putTMVar blockForgingVar $! a
@@ -308,8 +304,7 @@ data InternalState m addrNTN addrNTC blk = IS {
     , chainDB             :: ChainDB m blk
     , blockFetchInterface :: BlockFetchConsensusInterface (ConnectionId addrNTN) (Header blk) blk m
     , fetchClientRegistry :: FetchClientRegistry (ConnectionId addrNTN) (Header blk) blk m
-    , varCandidates       :: StrictTVar m (Map (ConnectionId addrNTN) (StrictTVar m (AnchoredFragment (Header blk))))
-    , varIdlers           :: StrictTVar m (Set (ConnectionId addrNTN))
+    , varChainSyncHandles :: StrictTVar m (Map (ConnectionId addrNTN) (ChainSyncClientHandle m blk))
     , mempool             :: Mempool m blk
     , peerSharingRegistry :: PeerSharingRegistry addrNTN m
     , varLedgerJudgement  :: StrictTVar m LedgerStateJudgement
@@ -337,8 +332,7 @@ initInternalState NodeKernelArgs { tracers, chainDB, registry, cfg
         gsmMarkerFileView
       newTVarIO j
 
-    varCandidates <- newTVarIO mempty
-    varIdlers     <- newTVarIO mempty
+    varChainSyncHandles <- newTVarIO mempty
     mempool       <- openMempool registry
                                  (chainDBLedgerInterface chainDB)
                                  (configLedger cfg)
@@ -349,7 +343,7 @@ initInternalState NodeKernelArgs { tracers, chainDB, registry, cfg
     fetchClientRegistry <- newFetchClientRegistry
 
     let getCandidates :: STM m (Map (ConnectionId addrNTN) (AnchoredFragment (Header blk)))
-        getCandidates = readTVar varCandidates >>= traverse readTVar
+        getCandidates = viewChainSyncState varChainSyncHandles csCandidate
 
     slotForgeTimeOracle <- BlockFetchClientInterface.initSlotForgeTimeOracle cfg chainDB
     let readFetchMode = BlockFetchClientInterface.readFetchModeDefault