Skip to content

Commit 525a3ef

Browse files
dnadalesdnadales
committed
Fetch missing CertRB closures on late-joining nodes
When ChainSel filters a CertRB because its EB closure is missing, drive a fetch through LeiosFetch using each peer's ChainSync candidate fragment as a fallback peer source. * Expose cdbPendingEBs via ChainDB.getPendingCertRBs. * pendingEbReconciler in NodeKernel mirrors the pending set into Leios missingEbBodies with size 0; it never overwrites offer-supplied sizes and only removes its own size-0 entries. * leiosFetchLogic walks per-peer ChainSync candidate fragments, extracts certified EB hashes via certifiedEbFromHeader, and passes a per-peer Set EbHash to leiosFetchLogicIteration. * choosePeerEb and choosePeerTx fall back to candidate-derived peers when no peer has offered the EB body / tx-closure. A peer whose candidate contains the CertRB must have validated the closure locally, so it must also hold both the body and the txs. * Relax the response-size check in msgLeiosBlock when the expected size is 0; the hash check remains authoritative.
1 parent 75ef78b commit 525a3ef

5 files changed

Lines changed: 124 additions & 29 deletions

File tree

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

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ import Data.Functor ((<&>))
5656
import Data.Hashable (Hashable)
5757
import Data.List.NonEmpty (NonEmpty)
5858
import qualified Data.List.NonEmpty as NE
59-
import Data.Maybe (isJust, mapMaybe)
59+
import Data.Maybe (fromMaybe, isJust, mapMaybe)
6060
import Data.Proxy
6161
import qualified Data.Text as Text
6262
import Data.Void (Void)
@@ -165,6 +165,8 @@ import Control.Concurrent.Class.MonadSTM.Strict (readTChan)
165165
import qualified Data.ByteString as BS
166166
import Data.Map (Map)
167167
import qualified Data.Map as Map
168+
import Data.Set (Set)
169+
import qualified Data.Set as Set
168170
import LeiosDemoDb
169171
( LeiosDbConnection
170172
, LeiosDbHandle (..)
@@ -176,9 +178,11 @@ import LeiosDemoDb
176178
import qualified LeiosDemoDb as LeiosDb
177179
import qualified LeiosDemoLogic as Leios
178180
import LeiosDemoTypes
179-
( ForgedLeiosEb
181+
( EbHash
182+
, ForgedLeiosEb
180183
, LeiosOutstanding
181184
, LeiosPeerVars
185+
, LeiosPoint (..)
182186
, LeiosVote (..)
183187
, TraceLeiosKernel (..)
184188
, VoterId (..)
@@ -419,6 +423,39 @@ initNodeKernel
419423
getLeiosOutstanding <- MVar.newMVar Leios.emptyLeiosOutstanding -- TODO init from DB
420424
getLeiosReady <- MVar.newEmptyMVar
421425

426+
-- Mirror cdbPendingEBs into Leios missingEbBodies with size 0 so that the
427+
-- LeiosFetch client tries to fetch closures for CertRBs that ChainSel is
428+
-- holding back.
429+
lastAppliedPendingEBs <- newTVarIO Map.empty
430+
void $
431+
forkLinkedThread registry "NodeKernel.pendingEbReconciler" $
432+
forever $ do
433+
(added, removed) <- atomically $ do
434+
new <- ChainDB.getPendingCertRBs chainDB
435+
old <- readTVar lastAppliedPendingEBs
436+
when (new == old) retry
437+
writeTVar lastAppliedPendingEBs new
438+
pure (Map.difference new old, Map.difference old new)
439+
MVar.modifyMVar_ getLeiosOutstanding $ \outstanding -> do
440+
let bodies0 = Leios.missingEbBodies outstanding
441+
-- Add entries from new pending CertRBs; never overwrite an
442+
-- existing non-zero (offer-supplied) size.
443+
bodies1 =
444+
Map.foldlWithKey'
445+
(\acc point _ -> Map.alter (Just . fromMaybe 0) point acc)
446+
bodies0
447+
added
448+
-- Drop entries we no longer need; keep offer-driven entries
449+
-- (non-zero size) untouched.
450+
bodies2 =
451+
Map.foldlWithKey'
452+
(\acc point _ ->
453+
Map.update (\sz -> if sz == 0 then Nothing else Just sz) point acc)
454+
bodies1
455+
removed
456+
pure outstanding{Leios.missingEbBodies = bodies2}
457+
void $ MVar.tryPutMVar getLeiosReady ()
458+
422459
void $
423460
forkLinkedThread registry "NodeKernel.leiosFetchLogic" $ do
424461
leiosConn <- allocate_ registry (LeiosDb.open leiosDB) LeiosDb.close
@@ -429,6 +466,21 @@ initNodeKernel
429466
iterationStart <- getMonotonicTime
430467
leiosPeersVars <- MVar.readMVar getLeiosPeersVars
431468
offerings <- mapM (MVar.readMVar . Leios.offerings) leiosPeersVars
469+
-- Per-peer certified EBs derived from ChainSync candidate fragments,
470+
-- used as a fallback peer source for fetching EB closures that no
471+
-- peer has explicitly offered.
472+
candidateCertEbs <- atomically $ do
473+
handles <- cschcMap varChainSyncHandles
474+
fmap (Map.mapKeysMonotonic Leios.MkPeerId) $
475+
forM handles $ \handle -> do
476+
state <- readTVar (cschState handle)
477+
pure $
478+
Set.fromList
479+
[ pointEbHash
480+
| hwt <- AF.toOldestFirst (csCandidate state)
481+
, Just MkLeiosPoint{pointEbHash} <-
482+
[LedgerDB.certifiedEbFromHeader (hwtHeader hwt)]
483+
]
432484
newDecisions <- MVar.modifyMVar getLeiosOutstanding $ \outstanding -> do
433485
-- Filter outstanding work against DB before running fetch iteration.
434486
-- This removes EBs and TXs we already have (e.g., from forging or other peers).
@@ -438,6 +490,7 @@ initNodeKernel
438490
Leios.leiosFetchLogicIteration
439491
Leios.demoLeiosFetchStaticEnv
440492
offerings
493+
candidateCertEbs
441494
filteredOutstanding
442495
pure (outstanding', newDecisions)
443496
traceWith tracer $ MkTraceLeiosKernel $ "leiosFetchLogic: decided"

‎ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic.hs‎

Lines changed: 52 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -262,9 +262,13 @@ leiosFetchLogicIteration ::
262262
Ord pid =>
263263
LeiosFetchStaticEnv ->
264264
Map (PeerId pid) (Set EbHash, Set EbHash) ->
265+
-- | Per-peer certified EB hashes inferred from ChainSync candidate
266+
-- fragments. Used as a fallback peer source when no peer has offered the
267+
-- EB body (e.g. for CertRBs whose closure pre-dates the local node).
268+
Map (PeerId pid) (Set EbHash) ->
265269
LeiosOutstanding pid ->
266270
(LeiosOutstanding pid, LeiosFetchDecisions pid)
267-
leiosFetchLogicIteration env offerings =
271+
leiosFetchLogicIteration env offerings candidateCertEbs =
268272
\acc ->
269273
go1 acc emptyLeiosFetchDecisions $
270274
expand $
@@ -330,17 +334,25 @@ leiosFetchLogicIteration env offerings =
330334

331335
choosePeerEb :: Set (PeerId pid) -> LeiosOutstanding pid -> EbHash -> Maybe (PeerId pid)
332336
choosePeerEb peerIds acc ebHash =
333-
foldr (\a _ -> Just a) Nothing $
334-
[ peerId
335-
| (peerId, (ebHashes, _ebHashes)) <-
336-
Map.toList $ -- TODO prioritize/shuffle?
337-
(`Map.withoutKeys` peerIds) $ -- not already requested from this peer
338-
offerings
339-
, Map.findWithDefault 0 peerId (Leios.requestedBytesSizePerPeer acc)
340-
<= Leios.maxRequestedBytesSizePerPeer env
341-
, -- peer can be sent more requests
342-
ebHash `Set.member` ebHashes -- peer has offered this EB body
343-
]
337+
case pickFrom (Map.map fst offerings) of
338+
Just peerId -> Just peerId
339+
-- No peer has offered this EB body; fall back to peers whose ChainSync
340+
-- candidate fragment includes the CertRB that depends on this EB.
341+
Nothing -> pickFrom candidateCertEbs
342+
where
343+
pickFrom :: Map (PeerId pid) (Set EbHash) -> Maybe (PeerId pid)
344+
pickFrom source =
345+
foldr (\a _ -> Just a) Nothing $
346+
[ peerId
347+
| (peerId, ebHashes) <-
348+
Map.toList $ -- TODO prioritize/shuffle?
349+
(`Map.withoutKeys` peerIds) $ -- not already requested from this peer
350+
source
351+
, Map.findWithDefault 0 peerId (Leios.requestedBytesSizePerPeer acc)
352+
<= Leios.maxRequestedBytesSizePerPeer env
353+
, -- peer can be sent more requests
354+
ebHash `Set.member` ebHashes
355+
]
344356

345357
goTx2 ::
346358
LeiosOutstanding pid ->
@@ -388,20 +400,29 @@ leiosFetchLogicIteration env offerings =
388400
BytesSize ->
389401
Maybe (PeerId pid, Map EbHash Int)
390402
choosePeerTx peerIds acc txOffsets targetTxBytesSize =
391-
foldr (\a _ -> Just a) Nothing $
392-
[ (peerId, Map.map fst txOffsets')
393-
| (peerId, (_ebIds, ebIds)) <-
394-
Map.toList $ -- TODO prioritize/shuffle?
395-
(`Map.withoutKeys` peerIds) $ -- not already requested from this peer
396-
offerings
397-
, Map.findWithDefault 0 peerId (Leios.requestedBytesSizePerPeer acc)
398-
<= Leios.maxRequestedBytesSizePerPeer env
399-
, -- peer can be sent more requests
400-
let txOffsets' = txOffsets `Map.restrictKeys` ebIds
401-
, case Map.lookupMax txOffsets' of
402-
Nothing -> False
403-
Just (_ebHash, (_txOffset, txBytesSize)) -> targetTxBytesSize == txBytesSize -- peer has offered at least one EB closure that includes this tx with the same size
404-
]
403+
case pickFrom (Map.map snd offerings) of
404+
Just hit -> Just hit
405+
-- Same fallback rationale as 'choosePeerEb': if a peer's ChainSync
406+
-- candidate contains a CertRB for this EB, the peer must have validated
407+
-- the full closure locally and therefore also has the txs.
408+
Nothing -> pickFrom candidateCertEbs
409+
where
410+
pickFrom :: Map (PeerId pid) (Set EbHash) -> Maybe (PeerId pid, Map EbHash Int)
411+
pickFrom source =
412+
foldr (\a _ -> Just a) Nothing $
413+
[ (peerId, Map.map fst txOffsets')
414+
| (peerId, ebIds) <-
415+
Map.toList $ -- TODO prioritize/shuffle?
416+
(`Map.withoutKeys` peerIds) $ -- not already requested from this peer
417+
source
418+
, Map.findWithDefault 0 peerId (Leios.requestedBytesSizePerPeer acc)
419+
<= Leios.maxRequestedBytesSizePerPeer env
420+
, -- peer can be sent more requests
421+
let txOffsets' = txOffsets `Map.restrictKeys` ebIds
422+
, case Map.lookupMax txOffsets' of
423+
Nothing -> False
424+
Just (_ebHash, (_txOffset, txBytesSize)) -> targetTxBytesSize == txBytesSize
425+
]
405426

406427
packRequests ::
407428
LeiosFetchStaticEnv ->
@@ -561,7 +582,11 @@ msgLeiosBlock ktracer tracer (outstandingVar, readyVar) db peerId req eb = do
561582
let MkLeiosPoint _ebSlot ebHash = point
562583
do
563584
let ebBytesSize' = leiosEbBytesSize eb
564-
when (ebBytesSize' /= ebBytesSize) $ do
585+
-- A 0 expected size signals a ChainSel-driven request (see
586+
-- 'pendingEbReconciler'): we don't know the EB body size up front because
587+
-- 'hbMayCertifiedEb' only carries the 'LeiosPoint'. The hash check below
588+
-- is the authoritative integrity check.
589+
when (ebBytesSize /= 0 && ebBytesSize' /= ebBytesSize) $ do
565590
error $ "MsgLeiosBlock size mismatch: " <> show (ebBytesSize', ebBytesSize)
566591
let ebHash' = hashLeiosEb eb
567592
when (ebHash' /= ebHash) $ do

‎ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/API.hs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,8 +74,10 @@ module Ouroboros.Consensus.Storage.ChainDB.API
7474

7575
import Control.Monad (void)
7676
import Control.ResourceRegistry
77+
import Data.Map.Strict (Map)
7778
import Data.Typeable (Typeable)
7879
import GHC.Generics (Generic)
80+
import LeiosDemoTypes (LeiosPoint)
7981
import Ouroboros.Consensus.Block
8082
import Ouroboros.Consensus.HeaderStateHistory
8183
( HeaderStateHistory (..)
@@ -383,6 +385,10 @@ data ChainDB m blk = ChainDB
383385
, getChainSelStarvation :: STM m ChainSelStarvation
384386
-- ^ Whether ChainSel is currently starved, or when was last time it
385387
-- stopped being starved.
388+
, getPendingCertRBs :: STM m (Map LeiosPoint (HeaderHash blk))
389+
-- ^ CertRBs filtered from ChainSel candidates because their EB closure
390+
-- is not yet available locally. Keyed by the missing EB's 'LeiosPoint',
391+
-- value is the CertRB's header hash.
386392
, getLedgerTablesAtFor ::
387393
Point blk ->
388394
LedgerTables (ExtLedgerState blk) KeysMK ->

‎ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl.hs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,7 @@ openDBInternal leiosDb args launchBgTasks = runWithTempRegistry $ do
275275
, newFollower = Follower.newFollower h
276276
, getIsInvalidBlock = getEnvSTM h Query.getIsInvalidBlock
277277
, getChainSelStarvation = getEnvSTM h Query.getChainSelStarvation
278+
, getPendingCertRBs = getEnvSTM h Query.getPendingCertRBs
278279
, closeDB = closeDB h
279280
, isOpen = isOpen h
280281
, getCurrentLedger = getEnvSTM h Query.getCurrentLedger

‎ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Query.hs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,15 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Query
3030
, getAnyKnownBlock
3131
, getAnyKnownBlockComponent
3232
, getChainSelStarvation
33+
, getPendingCertRBs
3334
) where
3435

3536
import Cardano.Ledger.BaseTypes (unNonZero)
3637
import Control.ResourceRegistry (ResourceRegistry)
38+
import Data.Map.Strict (Map)
3739
import qualified Data.Map.Strict as Map
3840
import qualified Data.Set as Set
41+
import LeiosDemoTypes (LeiosPoint)
3942
import Ouroboros.Consensus.Block
4043
import Ouroboros.Consensus.Config
4144
import Ouroboros.Consensus.HeaderStateHistory
@@ -194,6 +197,13 @@ getChainSelStarvation ::
194197
STM m ChainSelStarvation
195198
getChainSelStarvation CDB{..} = readTVar cdbChainSelStarvation
196199

200+
getPendingCertRBs ::
201+
forall m blk.
202+
IOLike m =>
203+
ChainDbEnv m blk ->
204+
STM m (Map LeiosPoint (HeaderHash blk))
205+
getPendingCertRBs cdb = readTVar (cdbPendingEBs cdb)
206+
197207
getIsValid ::
198208
forall m blk.
199209
(IOLike m, HasHeader blk) =>

0 commit comments

Comments
 (0)