diff --git a/cabal.project b/cabal.project index 1db4270776..8dc4d6b2d5 100644 --- a/cabal.project +++ b/cabal.project @@ -56,6 +56,15 @@ source-repository-package cardano-diffusion network-mux +-- Points to nfrisby-pr93-lookahead-merge-commit tag +source-repository-package + type: git + location: https://github.com/IntersectMBO/typed-protocols + tag: 9b4627221ae5d649f2303c6926a8dba3f9934658 + --sha256: sha256-55skVYIR0WjXMwJEirLgF0HFHUpFLBppuV9gvTWqoOI= + subdir: + typed-protocols + -- Points to ouroboros-ledger/leios-prototype source-repository-package type: git diff --git a/ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Block.hs b/ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Block.hs index b2be40c98e..c4e753dc08 100644 --- a/ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Block.hs +++ b/ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/Block.hs @@ -5,12 +5,15 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE PolyKinds #-} +{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneKindSignatures #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TypeOperators #-} +{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE ViewPatterns #-} + {-# OPTIONS_GHC -Wno-orphans #-} module Ouroboros.Consensus.Cardano.Block @@ -214,12 +217,17 @@ module Ouroboros.Consensus.Cardano.Block , EraMismatch (..) ) where +import Control.Monad.Except (throwError, withExcept) import Data.Kind +import Data.Proxy (Proxy (..)) +import Data.Functor.Product (Product (..)) import Data.SOP.BasicFunctors +import Data.SOP.Constraint (All, SListI) import Data.SOP.Functors -import Data.SOP.Index (Index (..)) +import Data.SOP.Index (Index (..), hcizipWith) +import qualified Data.SOP.Match as Match import Data.SOP.Strict -import Ouroboros.Consensus.Block (BlockProtocol) +import Ouroboros.Consensus.Block (BlockProtocol, SlotNo) import Ouroboros.Consensus.Byron.Ledger.Block (ByronBlock) import Ouroboros.Consensus.HardFork.Combinator import Ouroboros.Consensus.HardFork.Combinator.AcrossEras @@ -245,6 +253,8 @@ import Ouroboros.Consensus.Shelley.Ledger.Ledger ( ShelleyPartialLedgerConfig (..) ) import Ouroboros.Consensus.Shelley.Ledger.Leios () +import Ouroboros.Consensus.Shelley.Protocol.TPraos () +import Ouroboros.Consensus.Shelley.Protocol.Praos () import Ouroboros.Consensus.Storage.LedgerDB (ResolveLeiosBlock (..)) import Ouroboros.Consensus.TypeFamilyWrappers @@ -1537,6 +1547,7 @@ instance , ShelleyCompatible (Praos c) DijkstraEra , HasCanonicalTxIn (CardanoEras c) , HasHardForkTxOut (CardanoEras c) + , CanHardFork (CardanoEras c) ) => ResolveLeiosBlock (HardForkBlock (CardanoEras c)) where @@ -1596,6 +1607,13 @@ instance HeaderDijkstra dHdr -> headerLeiosAnnouncement dHdr _ -> Nothing + -- Compositional dispatch over the era stack (even though in practice only + -- Dijkstra carries an announcement, so only its instance is ever exercised). + headerElId (HardForkHeader (OneEraHeader ns)) = + hcollapse $ hcmap (Proxy @ResolveLeiosBlock) (K . headerElId) ns + + validateAnnouncementChainDepState = update @(CardanoEras c) + headerContainsLeiosCert hdr = case hdr of HeaderDijkstra dHdr -> headerContainsLeiosCert dHdr _ -> False @@ -1611,3 +1629,64 @@ instance (IS (IS (IS (IS (IS (IS (IS IZ))))))) (leiosClosureTxKeySets inner) _ -> emptyLedgerTables + +----- + +-- | We don't want to add the ResolveLeiosBlock sin-bin to SingleEraBlock, so we +-- use this ad-hoc class instead. +class (SingleEraBlock blk, ResolveLeiosBlock blk) => AnnouncementEraBlock blk +instance (SingleEraBlock blk, ResolveLeiosBlock blk) => AnnouncementEraBlock blk + +-- | Adapted from "Ouroboros.Consensus.HardFork.Combinator.Protocol.update" +update :: + forall xs. + (CanHardFork xs, All AnnouncementEraBlock xs) => + ConsensusConfig (HardForkProtocol xs) -> + OneEraValidateView xs -> + SlotNo -> + Ticked (HardForkChainDepState xs) -> + Except (HardForkValidationErr xs) () +update + HardForkConsensusConfig{..} + (OneEraValidateView view) + slot + (TickedHardForkChainDepState chainDepState ei) = + case State.match view chainDepState of + Left mismatch -> + throwError $ + HardForkValidationErrWrongEra . MismatchEraInfo $ + Match.bihcmap + proxySingle + singleEraInfo + (LedgerEraInfo . chainDepStateInfo . State.currentState) + mismatch + Right matched -> + hcollapse + . hcizipWith (Proxy :: Proxy AnnouncementEraBlock) (updateEra ei slot) cfgs + $ matched + where + cfgs = getPerEraConsensusConfig hardForkConsensusConfigPerEra + +-- | Adapted from "Ouroboros.Consensus.HardFork.Combinator.Protocol.updateEra" +updateEra :: + forall xs blk. + (SListI xs, SingleEraBlock blk, ResolveLeiosBlock blk) => + EpochInfo (Except PastHorizonException) -> + SlotNo -> + Index xs blk -> + WrapPartialConsensusConfig blk -> + Product WrapValidateView (Ticked :.: WrapChainDepState) blk -> + K (Except (HardForkValidationErr xs) ()) blk +updateEra + ei + slot + index + cfg + (Pair view (Comp chainDepState)) = + K $ + withExcept (injectValidationErr index) $ + validateAnnouncementChainDepState @blk + (completeConsensusConfig' ei cfg) + (unwrapValidateView view) + slot + (unwrapTickedChainDepState chainDepState) diff --git a/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Block.hs b/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Block.hs index a6d28e37d9..97f9e0a957 100644 --- a/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Block.hs +++ b/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Block.hs @@ -210,7 +210,7 @@ data instance Header (ShelleyBlock proto era) = ShelleyHeader deriving Generic deriving instance ShelleyCompatible proto era => Show (Header (ShelleyBlock proto era)) -deriving instance ShelleyCompatible proto era => Eq (Header (ShelleyBlock proto era)) +deriving instance ShelleyCompatible proto era => Eq (Header (ShelleyBlock proto era)) -- TODO sound to use only 'shelleyHeaderHash'? deriving instance ShelleyCompatible proto era => NoThunks (Header (ShelleyBlock proto era)) instance diff --git a/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Leios.hs b/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Leios.hs index 04462d292e..4f8559e575 100644 --- a/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Leios.hs +++ b/ouroboros-consensus-cardano/src/shelley/Ouroboros/Consensus/Shelley/Ledger/Leios.hs @@ -11,12 +11,14 @@ module Ouroboros.Consensus.Shelley.Ledger.Leios () where +import qualified Cardano.Crypto.Hash as Crypto (hashToBytesShort) import Cardano.Ledger.Api (Tx) import Cardano.Ledger.Binary (decCBOR, decodeFullAnnotator) import qualified Cardano.Ledger.Block as Core import Cardano.Ledger.Core (TopTx, injectFailure) import qualified Cardano.Ledger.Core as Core import Cardano.Ledger.Dijkstra.BlockBody (leiosCertBlockBodyL) +import Cardano.Ledger.Hashes (KeyHash (..)) import qualified Cardano.Ledger.Shelley.API as SL import Cardano.Ledger.Shelley.Rules (ledgerPpL) import qualified Cardano.Ledger.Shelley.UTxO as SL @@ -31,13 +33,29 @@ import Data.Maybe.Strict (strictMaybeToMaybe) import Data.Proxy (Proxy (..)) import qualified Data.Sequence.Strict as StrictSeq import LeiosDemoDb (leiosDbLookupEbClosure) -import LeiosDemoTypes (EbAnnouncement (..), LeiosPoint (..), RbHash (..)) +import LeiosDemoLogic.Announcements.ElBimap (ElId (MkElId)) +import LeiosDemoTypes + ( EbAnnouncement (..) + , LeiosPoint (..) + , RbHash (..) + ) import Lens.Micro ((.~), (^.)) import Ouroboros.Consensus.Block (ChainHash (..), blockPrevHash, toRawHash) import Ouroboros.Consensus.Ledger.Abstract (getTipSlot) import Ouroboros.Consensus.Ledger.SupportsMempool (getTransactionKeySets) import Ouroboros.Consensus.Ledger.Tables (stowLedgerTables, unstowLedgerTables) -import Ouroboros.Consensus.Protocol.Praos (Praos, PraosCrypto, PraosState (..)) +import Ouroboros.Consensus.Protocol.Praos + ( ConsensusConfig (..) + , Praos + , PraosCrypto + , PraosParams (..) + , PraosState (..) + , Ticked (..) + , WhetherToUpperBoundOCERT (..) + , doValidateKESSignatureWorker + , doValidateVRFSignature + ) +import Ouroboros.Consensus.Protocol.Praos.Views (plvPoolDistr) import Ouroboros.Consensus.Protocol.Praos.Header ( Header (..) , HeaderBody (..) @@ -195,6 +213,38 @@ instance where Header{headerBody} = shelleyHeaderRaw hdr + headerElId hdr = + MkElId + headerBody.hbSlotNo + (Crypto.hashToBytesShort . unKeyHash . SL.hashKey $ headerBody.hbVk) + where + Header{headerBody} = shelleyHeaderRaw hdr + + -- The announcement is validated out-of-context against a (possibly lagging) + -- immutable tip, so we skip the OCERT counter's upper bound + -- ('DoNotUpperBoundOCERT'): a counter ahead of our recorded view is honestly + -- explained by that lag. The election proof (VRF) and the signature (KES + + -- opcert, including the counter's revocation lower bound) are checked in full. + validateAnnouncementChainDepState cfg hv _slot tcs = do + -- validate the claimed election + doValidateVRFSignature + (praosStateEpochNonce cs) + pd + (praosLeaderF prms) + hv + -- authenticate the message + doValidateKESSignatureWorker + DoNotUpperBoundOCERT + (praosMaxKESEvo prms) + (praosSlotsPerKESPeriod prms) + pd + (praosStateOCertCounters cs) + hv + where + prms = praosParams cfg + cs = tickedPraosStateChainDepState tcs + SL.PoolDistr pd _ = plvPoolDistr (tickedPraosStateLedgerView tcs) + protocolStateLeiosAnnouncement st = do ann <- strictMaybeToMaybe $ praosStateLeiosAnnouncement st pure diff --git a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs index 2ac1d0428e..bcef399ee8 100644 --- a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs +++ b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs @@ -8,6 +8,7 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} {-# OPTIONS_GHC -Wno-orphans #-} -- | Intended for qualified import @@ -60,9 +61,10 @@ import qualified Control.Concurrent.Class.MonadSTM as LazySTM import Control.Concurrent.Class.MonadSTM.Strict (readTChan) import qualified Control.Concurrent.Class.MonadSTM.Strict.TVar as TVar.Unchecked import Control.DeepSeq (NFData) -import Control.Monad (forM_, void) +import Control.Monad (forM_, forever, void, when) import Control.Monad.Class.MonadTime.SI (MonadTime) import Control.Monad.Class.MonadTimer.SI (MonadTimer) +import Control.Monad.Except (runExceptT) import Control.ResourceRegistry import Control.Tracer import Data.ByteString.Lazy (ByteString) @@ -70,6 +72,8 @@ import Data.Functor ((<&>)) import Data.Hashable (Hashable) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map +import qualified Data.Primitive.MutVar as Prim +import qualified Data.Sequence as Seq import qualified Data.Set as Set import Data.Void (Void) import LeiosDemoDb @@ -79,6 +83,7 @@ import LeiosDemoDb , withLeiosDb ) import qualified LeiosDemoLogic as Leios +import qualified LeiosDemoLogic.Announcements as Announcements import LeiosDemoOnlyTestFetch ( LeiosFetch , LeiosFetchClientPeerPipelined @@ -95,7 +100,7 @@ import LeiosDemoOnlyTestFetch import LeiosDemoOnlyTestNotify ( LeiosNotify , LeiosNotifyClientPeerPipelined - , LeiosNotifyServerPeer + , LeiosNotifyServerPeerLookahead , Message ( MsgLeiosBlockAnnouncement , MsgLeiosBlockOffer @@ -107,7 +112,8 @@ import LeiosDemoOnlyTestNotify , codecLeiosNotifyId , leiosNotifyClientPeerPipelined , leiosNotifyMiniProtocolNum - , leiosNotifyServerPeer + , leiosNotifyServerPeerLookahead + , runLookaheadFixedSenderPeerWithLimits , timeLimitsLeiosNotify , toLeiosNotifyClientPeerPipelined ) @@ -150,8 +156,6 @@ import Ouroboros.Consensus.NodeKernel import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB import Ouroboros.Consensus.Storage.LedgerDB.Forker ( ResolveLeiosBlock - , headerContainsLeiosCert - , headerLeiosAnnouncement ) import Ouroboros.Consensus.Storage.Serialisation (SerialisedHeader) import Ouroboros.Consensus.Util (ShowProxy) @@ -325,7 +329,10 @@ data Handlers m addr blk = Handlers , hLeiosNotifyServer :: NodeToNodeVersion -> ConnectionId addr -> - LeiosNotifyServerPeer LeiosPoint (Header blk) LeiosVote m () + m + ( LeiosNotifyServerPeerLookahead LeiosPoint (Header blk) LeiosVote m () + , m Void + ) , hLeiosFetchClient :: LeiosDbConnection m -> NodeToNodeVersion -> @@ -364,6 +371,7 @@ mkHandlers , miniProtocolParameters , getDiffusionPipeliningSupport , txSubmissionInitDelay + , systemTime } nodeKernel@NodeKernel { getChainDB @@ -453,44 +461,79 @@ mkHandlers let tracer = leiosPeerTracer peer kernelTracer = Node.leiosKernelTracer tracers LeiosVoteState{addVote} = leiosVoteState + -- Per-upstream-peer announcement accountability (dedup and + -- equivocation counting). The pipelined-client handler is a stateless + -- callback, so this state lives in a (single-threaded) ref. + peerStateVar <- Prim.newMutVar (SlotNo 0, Announcements.emptyPeerState) pure $ leiosNotifyClientPeerPipelined ( atomically controlMessageSTM <&> \case Terminate -> Left () - _ -> Right 100 {- TODO magic number -} + _ -> Right Leios.lEIOSNOTIFYPIPELINEDEPTH ) ( pure $ \case MsgLeiosBlockAnnouncement hdr -> do - traceWith tracer $ - MkTraceLeiosPeer $ - unwords - [ "MsgLeiosBlockAnnouncement" - , "(ignored)" - , show (headerPoint hdr) - , "containsCert=" <> show (headerContainsLeiosCert hdr) - , "announcedEb=" <> case headerLeiosAnnouncement hdr of - Nothing -> "none" - Just (ebPoint, ebSize) -> - Leios.prettyEbHash (Leios.pointEbHash ebPoint) <> " size=" <> show ebSize - ] + -- A header relayed as an announcement must carry one; + -- disconnect the peer otherwise. + anc <- case Leios.mkAnnouncingHeader hdr of + Nothing -> throwIO Leios.ExnLeiosBlockAnnouncementMissing + Just x -> pure x + immLedger <- atomically $ ChainDB.getImmutableLedger getChainDB + (latestPruneSlot, peerSt0) <- Prim.readMutVar peerStateVar + res <- + runExceptT $ + Announcements.onAnnouncement + (contramap Leios.tracePeerAnnouncement tracer) + Leios.ancElId + ( \ancH -> + Leios.announcementValidity + systemTime + chainSyncFutureCheck + getTopLevelConfig + immLedger + (Leios.ancHeader ancH) + ) + -- central part of the processing + ( \ancHdr (shouldRelay, age, anc'@(p, _sz)) -> do + traceWith tracer $ + MkTraceLeiosPeer $ "MsgLeiosBlockAnnouncement new: " <> Leios.prettyLeiosPoint p + MVar.modifyMVar_ getLeiosCentralState $ \cst -> -- TODO OK to hold this the whole time we're writing to the LeiosNotify queues (NB those enqeues never block)? + Announcements.onAnnouncementCentral + (contramap Leios.traceNewAnnouncement kernelTracer) + Leios.ancElId + ( \_elSt -> + Leios.recordAnnouncedEb (getLeiosOutstanding, getLeiosReady) anc' + ) + cst + (Just peer) + shouldRelay + (Just age) + ancHdr + ) + peerSt0 + anc + peerSt1 <- case res of + Left err -> throwIO $ Leios.ReactToAnnouncementError err + Right x -> pure x + let (!latestPruneSlot', !peerSt2) = + Leios.prunePeerStateToImmTip immLedger latestPruneSlot peerSt1 + Prim.writeMutVar peerStateVar (latestPruneSlot', peerSt2) MsgLeiosBlockOffer point ebBytesSize -> do traceWith tracer $ MkTraceLeiosPeer $ "MsgLeiosBlockOffer " <> Leios.prettyLeiosPoint point let MkLeiosPoint{pointEbHash = ebHash} = point - -- FIXME: EB announcements are not implemented. The - -- fetch state is built entirely from peer offers, - -- which is the wrong source of truth — the - -- authoritative size lives in - -- 'headerLeiosAnnouncement' on the parent RB - -- header (signed by the forger). Until announcement - -- handling lands, the sanitisation below is the - -- best we can do against malformed offers: - -- drop a zero-sized offer outright (no honest - -- forger ever announces a 0-byte EB) and refuse to - -- overwrite an existing entry that shares the same - -- content hash, so the first-seen (slot, size) - -- wins. The per-peer 'offerings' below is still - -- updated so the peer remains a valid serving - -- candidate. + -- TODO: EB announcements now record the authoritative + -- (forger-signed) size via 'recordAnnouncedEb', but this + -- offer handler is not integrated with them yet: it still + -- builds fetch state directly from peer offers, whose sizes + -- are not authoritative (the authoritative one lives in + -- 'headerLeiosAnnouncement' on the parent RB header). Until + -- the two are reconciled, the sanitisation below is the best + -- we can do against malformed offers: drop a zero-sized + -- offer outright (no honest forger ever announces a 0-byte + -- EB) and refuse to overwrite an existing entry that shares + -- the same content hash, so the first-seen (slot, size) + -- wins. The per-peer 'offerings' below is still updated so + -- the peer remains a valid serving candidate. MVar.modifyMVar_ getLeiosOutstanding $ \outstanding -> pure $ if ebBytesSize == 0 @@ -529,25 +572,50 @@ mkHandlers traceWith kernelTracer TraceLeiosCertified{rbHash = Leios.announcingRbHash vote} _ -> pure () ) - , hLeiosNotifyServer = \_version _peer -> Effect $ do + , hLeiosNotifyServer = \_version peer -> do chan <- subscribeEbNotifications leiosDB - let processEbNotification :: - STM - m - ( LeiosDemoOnlyTestNotify.Message - (LeiosNotify LeiosPoint (Header blk) LeiosVote) - LeiosDemoOnlyTestNotify.StBusy - LeiosDemoOnlyTestNotify.StIdle - ) - processEbNotification = - readTChan chan >>= \case - AcquiredEb point ebSize -> - pure $ MsgLeiosBlockOffer point ebSize - AcquiredEbTxs point -> - pure $ MsgLeiosBlockTxsOffer point - LeiosVoteSubscription{getNextVote} <- subscribeVotes leiosVoteState - let processVote :: + + -- This peer's single outgoing LeiosNotify queue and its credit + -- counter, shared by all three sources (announcements, offers, votes) + -- and served strictly FIFO. The node-wide relay logic + -- ('onAnnouncementCentral') appends announcements through the + -- 'QueueAnnouncementView' we register below; offers and votes are + -- moved in by the 'pump'. + -- + -- TODO the EB-offer and vote sources should eventually /register/ + -- this peer with those components too, rather than the pump draining + -- fresh per-peer subscriptions. + credits <- TVar.Unchecked.newTVarIO (0 :: Int) + queue <- TVar.Unchecked.newTVarIO Seq.empty + + -- The view the central relay logic uses to enqueue announcements + let qav = + Announcements.MkQueueAnnouncementView + credits + (\q anc -> q Seq.|> MsgLeiosBlockAnnouncement (Leios.ancHeader anc)) + queue + + let -- 'incr' adds a credit per received request; 'next' hands the + -- sender thread the next queued message (FIFO across all three + -- sources); 'pump' moves offers/votes into the queue, dropping a + -- message when there are no free credits. + incr = atomically $ do + n <- TVar.Unchecked.readTVar credits + if n == Leios.lEIOSNOTIFYPIPELINEDEPTH + then pure LeiosDemoOnlyTestNotify.ExcessiveRequests + else do + TVar.Unchecked.writeTVar credits $! n + 1 + pure LeiosDemoOnlyTestNotify.NotExcessiveRequests + next = atomically $ do + q <- TVar.Unchecked.readTVar queue + case Seq.viewl q of + Seq.EmptyL -> LazySTM.retry + msg Seq.:< q' -> do + TVar.Unchecked.writeTVar queue q' + pure msg + + pumpNext :: STM m ( LeiosDemoOnlyTestNotify.Message @@ -555,13 +623,32 @@ mkHandlers LeiosDemoOnlyTestNotify.StBusy LeiosDemoOnlyTestNotify.StIdle ) - processVote = do - vote <- getNextVote - pure $ MsgLeiosVotes [vote] + pumpNext = + ( readTChan chan >>= \case + AcquiredEb point ebSize -> + pure $ MsgLeiosBlockOffer point ebSize + AcquiredEbTxs point -> + pure $ MsgLeiosBlockTxsOffer point + ) + <|> (getNextVote <&> \vote -> MsgLeiosVotes [vote]) - pure . leiosNotifyServerPeer $ - atomically $ - processEbNotification <|> processVote + pump = + ( do + MVar.modifyMVar_ getLeiosCentralState $ + pure . Announcements.insertPeerCentral peer qav + forever $ atomically $ do + msg <- pumpNext + c <- TVar.Unchecked.readTVar credits + when (c > 0) $ do + TVar.Unchecked.writeTVar credits $! c - 1 + q <- TVar.Unchecked.readTVar queue + TVar.Unchecked.writeTVar queue (q Seq.|> msg) + ) + `finally` ( MVar.modifyMVar_ getLeiosCentralState $ + pure . Announcements.deletePeerCentral peer + ) + + pure (leiosNotifyServerPeerLookahead incr next, pump) , hLeiosFetchClient = \leiosConn _version controlMessageSTM peer peerVars -> toLeiosFetchClientPeerPipelined $ Effect $ do let reqVar = Leios.requestsToSend peerVars -- Queue for responses received by the pipelined-peer collector @@ -594,6 +681,7 @@ mkHandlers , getLeiosVoteState = leiosVoteState , getLeiosOutstanding , getLeiosReady + , getLeiosCentralState } = nodeKernel leiosPeerTracer peer = TraceLabelPeer peer `contramap` Node.leiosPeerTracer tracers @@ -1342,13 +1430,18 @@ mkApps kernel rng Tracers{tTxLogicTracer = _, ..} mkCodecs ByteLimits{..} chainS m ((), Maybe bLN) aLeiosNotifyServer version ResponderContext{rcConnectionId = them} channel = do labelThisThread "LeiosNotifyServer" - runPeerWithLimits - (TraceLabelPeer them `contramap` tLeiosNotifyTracer) - (cLeiosNotifyCodec (mkCodecs version)) - blLeiosNotify - timeLimitsLeiosNotify - channel - $ hLeiosNotifyServer version them + (peer, pump) <- hLeiosNotifyServer version them + -- The pump lives exactly as long as the peer; 'link' surfaces a pump + -- crash instead of silently leaving the peer unable to send. + withAsync pump $ \pumpThread -> do + link pumpThread + runLookaheadFixedSenderPeerWithLimits + (TraceLabelPeer them `contramap` tLeiosNotifyTracer) + (cLeiosNotifyCodec (mkCodecs version)) + blLeiosNotify + timeLimitsLeiosNotify + channel + peer aLeiosFetchClient :: NodeToNodeVersion -> diff --git a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs index 33e3a56a59..9f74bac41b 100644 --- a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs +++ b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs @@ -63,6 +63,7 @@ import LeiosDemoDb ) import qualified LeiosDemoDb as LeiosDb import qualified LeiosDemoLogic as Leios +import qualified LeiosDemoLogic.Announcements as Announcements import LeiosDemoTypes ( LeiosOutstanding , LeiosPeerVars @@ -120,6 +121,7 @@ import qualified Ouroboros.Consensus.Storage.ChainDB.Init as InitChainDB import Ouroboros.Consensus.Util.AnchoredFragment ( preferAnchoredCandidate ) +import Ouroboros.Consensus.Util (whenJust) import Ouroboros.Consensus.Util.EarlyExit hiding (callTraceSameThread) import Ouroboros.Consensus.Util.IOLike import Ouroboros.Consensus.Util.LeakyBucket @@ -244,6 +246,9 @@ data NodeKernel m addrNTN addrNTC blk = NodeKernel , getLeiosReady :: MVar.MVar m () -- ^ Filled by anyone who makes a change that might unblock a new -- fetch decision; the fetch logic 'MVar.takeMVar's before it runs. + , getLeiosCentralState :: + MVar.MVar m (Announcements.CentralState m (ConnectionId addrNTN) (Leios.AnnouncingHeader blk)) + -- ^ Node-wide EB-announcement state } -- | Arguments required when initializing a node @@ -325,6 +330,7 @@ initNodeKernel , varGsmState , leiosOutstanding = getLeiosOutstanding , leiosReady = getLeiosReady + , leiosCentralState = getLeiosCentralState , leiosPeersVars = getLeiosPeersVars , leiosVoteState } = st @@ -539,6 +545,19 @@ initNodeKernel leiosVoteState (topLevelConfigVotingKey cfg) + void $ + forkLinkedWatcher registry "NodeKernel.leiosPruneAnnouncements" $ + Watcher + { wFingerprint = id + , wInitial = Nothing + , wReader = getTipSlot . ledgerState <$> ChainDB.getImmutableLedger chainDB + , wNotify = \case + Origin -> pure () + NotOrigin immTipSlot -> + MVar.modifyMVar_ getLeiosCentralState $ + pure . Announcements.pruneCentralState immTipSlot + } + return NodeKernel { getChainDB = chainDB @@ -566,9 +585,11 @@ initNodeKernel , getLeiosPeersVars = getLeiosPeersVars , getLeiosOutstanding = getLeiosOutstanding , getLeiosReady = getLeiosReady + , getLeiosCentralState = getLeiosCentralState } where blockForgingController :: + Ord remotePeer => InternalState m remotePeer localPeer blk -> STM m [MkBlockForging m blk] -> m Void @@ -615,6 +636,8 @@ data InternalState m addrNTN addrNTC blk = IS , -- Leios fetch-logic state; consumed in 'initNodeKernel'. leiosOutstanding :: MVar.MVar m (LeiosOutstanding (ConnectionId addrNTN)) , leiosReady :: MVar.MVar m () + , leiosCentralState :: + MVar.MVar m (Announcements.CentralState m (ConnectionId addrNTN) (Leios.AnnouncingHeader blk)) , leiosPeersVars :: LazySTM.TVar m (Map.Map (Leios.PeerId (ConnectionId addrNTN)) (LeiosPeerVars m)) , leiosVoteState :: LeiosVoteState m @@ -673,6 +696,7 @@ initInternalState leiosPeersVars <- LazySTM.newTVarIO Map.empty leiosOutstanding <- MVar.newMVar Leios.emptyLeiosOutstanding leiosReady <- MVar.newEmptyMVar + leiosCentralState <- MVar.newMVar Announcements.emptyCentralState let readFetchMode = BlockFetchClientInterface.readFetchModeDefault @@ -710,7 +734,7 @@ toConsensusMode = \case forkBlockForging :: forall m addrNTN addrNTC blk. - (IOLike m, RunNode blk) => + (IOLike m, RunNode blk, Ord addrNTN) => InternalState m addrNTN addrNTC blk -> MkBlockForging m blk -> m (Thread m Void) @@ -745,12 +769,32 @@ forkBlockForging IS{..} (MkBlockForging blockForgingM) = leiosVoteState bf leiosConn + announceForgedBlock currentSlot ) where label :: String label = "NodeKernel.blockForging" + -- Concurrently (fire-and-forget) relay this node's own freshly-forged EB + -- announcement, if any, to downstream peers via LeiosNotify. 'forge' invokes + -- this right after forging and before adoption, so adoption never gates + -- getting the announcement onto the wire. + announceForgedBlock :: Header blk -> m () + announceForgedBlock forgedHeader = + whenJust (Leios.mkAnnouncingHeader forgedHeader) $ \anc -> + void $ async $ + MVar.modifyMVar_ leiosCentralState $ \cst -> + Announcements.onAnnouncementCentral + (contramap Leios.traceNewAnnouncement (leiosKernelTracer tracers)) + Leios.ancElId + (\_elSt -> pure ()) -- we forged the EB; nothing to fetch locally + cst + Nothing -- the source is this node, not an upstream peer + Announcements.DoRelay -- our newly forged block can't be too old + Nothing -- no wall-clock lateness for a locally-forged announcement + anc + -- 'LeiosDbConnection' is not thread-safe, so we open one per -- forge-credentials thread (and close it when the thread exits). allocateForging = do diff --git a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel/Forge.hs b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel/Forge.hs index 1fb992c6c1..f6c0300ffb 100644 --- a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel/Forge.hs +++ b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel/Forge.hs @@ -97,9 +97,13 @@ forge :: LeiosVoteState m -> BlockForging m blk -> LeiosDbConnection m -> + -- | Invoked with the freshly-forged block's header, after forging and + -- /before/ adoption, so the caller can act on the new block (e.g. concurrently + -- announce its EB) without adoption gating it. + (Header blk -> m ()) -> SlotNo -> WithEarlyExit m () -forge forgeEventTracer forgeStateInfoTracer leiosTracer forgeCCtx cfg chainDB mempool leiosVoteState blockForging leiosConn currentSlot = do +forge forgeEventTracer forgeStateInfoTracer leiosTracer forgeCCtx cfg chainDB mempool leiosVoteState blockForging leiosConn afterForge currentSlot = do let trace :: TraceForgeEvent blk -> WithEarlyExit m () trace = lift @@ -257,6 +261,10 @@ forge forgeEventTracer forgeStateInfoTracer leiosTracer forgeCCtx cfg chainDB me snapSize rbTxsSize + -- Hand the freshly-forged block's header to the caller before adoption, so it + -- can act on it (e.g. concurrently announce its EB) without adoption gating it. + lift $ afterForge (getHeader newBlock) + forgeTrace'Via (const ()) "add-block-to-chaindb" diff --git a/ouroboros-consensus-protocol/src/ouroboros-consensus-protocol/Ouroboros/Consensus/Protocol/Praos.hs b/ouroboros-consensus-protocol/src/ouroboros-consensus-protocol/Ouroboros/Consensus/Protocol/Praos.hs index fc151bc1e0..75606faeac 100644 --- a/ouroboros-consensus-protocol/src/ouroboros-consensus-protocol/Ouroboros/Consensus/Protocol/Praos.hs +++ b/ouroboros-consensus-protocol/src/ouroboros-consensus-protocol/Ouroboros/Consensus/Protocol/Praos.hs @@ -31,6 +31,8 @@ module Ouroboros.Consensus.Protocol.Praos -- * For testing purposes , doValidateKESSignature + , doValidateKESSignatureWorker + , WhetherToUpperBoundOCERT (..) , doValidateVRFSignature ) where @@ -672,6 +674,20 @@ validateKESSignature ocertCounters = doValidateKESSignature praosMaxKESEvo praosSlotsPerKESPeriod lvPoolDistr ocertCounters +-- | Whether 'doValidateKESSignatureWorker' enforces the OCERT counter's /upper/ +-- bound, i.e. rejects (with 'CounterOverIncrementedOCERT') a counter more than +-- one greater than the one we have recorded for this issuer. +-- +-- Normal header validation enforces it ('UpperBoundOCERT'). Validating a +-- relayed Leios announcement out-of-context against a (possibly lagging) +-- immutable tip legitimately sees counters that have run ahead of our recorded +-- view, so that path skips it ('DoNotUpperBoundOCERT'). The counter's /lower/ +-- bound ('CounterTooSmallOCERT', a revoked key) is enforced either way. +data WhetherToUpperBoundOCERT + = UpperBoundOCERT + | DoNotUpperBoundOCERT + deriving (Eq, Show) + -- NOTE: This function is much easier to test than 'validateKESSignature' because we don't need to -- construct a 'PraosConfig' nor 'LedgerView' to test it. doValidateKESSignature :: @@ -682,7 +698,20 @@ doValidateKESSignature :: Map (KeyHash SL.BlockIssuer) Word64 -> Views.HeaderView c -> Except (PraosValidationErr c) () -doValidateKESSignature praosMaxKESEvo praosSlotsPerKESPeriod stakeDistribution ocertCounters b = +doValidateKESSignature = doValidateKESSignatureWorker UpperBoundOCERT + +-- | The worker underlying 'doValidateKESSignature', parameterized by whether to +-- enforce the OCERT counter's upper bound (see 'WhetherToUpperBoundOCERT'). +doValidateKESSignatureWorker :: + PraosCrypto c => + WhetherToUpperBoundOCERT -> + Word64 -> + Word64 -> + Map (KeyHash SL.StakePool) SL.IndividualPoolStake -> + Map (KeyHash SL.BlockIssuer) Word64 -> + Views.HeaderView c -> + Except (PraosValidationErr c) () +doValidateKESSignatureWorker whetherToUpperBound praosMaxKESEvo praosSlotsPerKESPeriod stakeDistribution ocertCounters b = do c0 <= kp ?! KESBeforeStartOCERT c0 kp kp_ < c0_ + fromIntegral praosMaxKESEvo ?! KESAfterEndOCERT kp c0 praosMaxKESEvo @@ -701,7 +730,9 @@ doValidateKESSignature praosMaxKESEvo praosSlotsPerKESPeriod stakeDistribution o throwError $ NoCounterForKeyHashOCERT hk Just m -> do m <= n ?! CounterTooSmallOCERT m n - n <= m + 1 ?! CounterOverIncrementedOCERT m n + case whetherToUpperBound of + UpperBoundOCERT -> n <= m + 1 ?! CounterOverIncrementedOCERT m n + DoNotUpperBoundOCERT -> pure () where oc@(OCert vk_hot n c0@(KESPeriod c0_) tau) = Views.hvOCert b (VKey vkcold) = Views.hvVK b diff --git a/ouroboros-consensus.cabal b/ouroboros-consensus.cabal index 3cf8155362..61579da93e 100644 --- a/ouroboros-consensus.cabal +++ b/ouroboros-consensus.cabal @@ -106,6 +106,9 @@ library LeiosDemoDb.Trace LeiosDemoException LeiosDemoLogic + LeiosDemoLogic.Announcements + LeiosDemoLogic.Announcements.ElBimap + LeiosDemoLogic.Announcements.Validate LeiosDemoOnlyTestFetch LeiosDemoOnlyTestNotify LeiosDemoTypes @@ -395,7 +398,7 @@ library network-mux, nonempty-containers, nothunks ^>=0.2 || ^>=0.3, - ouroboros-network:{api, api-tests-lib, protocols} ^>=1.1, + ouroboros-network:{api, api-tests-lib, framework, protocols} ^>=1.1, pretty-simple, primitive, psqueues ^>=0.2.3, @@ -723,6 +726,7 @@ test-suite consensus-test Test.Consensus.Util.Versioned Test.LeiosDemoDb Test.LeiosDemoLogic + Test.LeiosDemoLogic.Announcements Test.LeiosDemoTypes Test.LeiosUtils.CallTrace Test.LeiosVoteState @@ -1144,6 +1148,7 @@ library diffusion network-mux ^>=0.10, ouroboros-consensus:{ouroboros-consensus, protocol}, ouroboros-network:{api, framework, ouroboros-network, protocols}, + primitive, random ^>=1.3, resource-registry, safe-wild-cards ^>=1.0, diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic.hs index 287439aa49..d92a14ba7e 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic.hs @@ -1,11 +1,15 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeApplications #-} +{-# LANGUAGE UndecidableInstances #-} module LeiosDemoLogic (module LeiosDemoLogic) where @@ -17,13 +21,14 @@ import Control.Concurrent.Class.MonadSTM.Strict (StrictTVar) import qualified Control.Concurrent.Class.MonadSTM.Strict as StrictSTM import Control.Monad (forM_, when) import Control.Monad.Class.MonadThrow (Exception, catch, throwIO) +import Control.Monad.Except (runExcept) import Control.Monad.Primitive (PrimMonad, PrimState) import Control.Tracer (Tracer, traceWith) import qualified Data.Bits as Bits import qualified Data.ByteString as BS import Data.DList (DList) import qualified Data.DList as DList -import Data.Functor (void) +import Data.Functor ((<&>), void) import Data.IntMap (IntMap) import qualified Data.IntMap as IntMap import qualified Data.IntSet as IntSet @@ -34,6 +39,7 @@ import Data.Sequence (Seq) import qualified Data.Sequence as Seq import Data.Set (Set) import qualified Data.Set as Set +import Data.Time.Clock (NominalDiffTime) import qualified Data.Vector.Strict as V import qualified Data.Vector.Strict.Mutable as MV import Data.Word (Word16, Word64) @@ -47,9 +53,18 @@ import LeiosDemoDb , leiosDbInsertTxs , leiosDbLookupEbBody ) +import qualified LeiosDemoLogic.Announcements as Announcements +import LeiosDemoLogic.Announcements.ElBimap (ElId) +import LeiosDemoLogic.Announcements.Validate + ( AnnouncementInvalidity + , validateAnnouncementHeader + ) import qualified LeiosDemoOnlyTestFetch as LF import LeiosDemoTypes - ( BytesSize + ( AnnouncementEquivocation (..) + , AnnouncementFields (..) + , AnnouncementSource (..) + , BytesSize , EbHash (..) , LeiosBlockRequest (..) , LeiosBlockTxsRequest (..) @@ -70,7 +85,24 @@ import LeiosDemoTypes , maxTxsPerEb ) import qualified LeiosDemoTypes as Leios -import Ouroboros.Consensus.Block (BlockProtocol, Header) +import Ouroboros.Consensus.Block + ( BlockProtocol + , HasHeader + , Header + , WithOrigin (NotOrigin) + , headerHash + ) +import Ouroboros.Consensus.BlockchainTime.WallClock.Types + ( SystemTime + , diffRelTime + , systemTimeCurrent + ) +import Ouroboros.Consensus.Config (TopLevelConfig, configLedger) +import Ouroboros.Consensus.Ledger.Abstract (getTipSlot) +import Ouroboros.Consensus.Ledger.Basics (EmptyMK) +import Ouroboros.Consensus.Ledger.Extended (ExtLedgerState, ledgerState) +import Ouroboros.Consensus.Ledger.SupportsProtocol (LedgerSupportsProtocol) +import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck as InFutureCheck import Ouroboros.Consensus.Protocol.Abstract (ChainDepState) import Ouroboros.Consensus.Storage.LedgerDB.Forker (ResolveLeiosBlock (..)) import Ouroboros.Consensus.Util.IOLike (IOLike) @@ -951,3 +983,216 @@ leiosCertRbCallback kernelVars peerVars hdr cds = when (headerContainsLeiosCert hdr) $ forM_ (protocolStateLeiosAnnouncement @blk cds) $ \announcement -> leiosCertRbOffer kernelVars peerVars announcement + +----- + +-- The pure logic for handling an inbound 'MsgLeiosBlockAnnouncement'. The +-- effectful glue (reading the immutable tip, the 'Announcements.PeerState' ref, +-- 'MVar' updates, tracing, and 'throwIO') lives in the NodeToNode client, which +-- invokes 'Announcements.onAnnouncement' with these pieces. + +-- | 'Header blk' as a relayed LeiosNotify announcement, paired with the +-- announcement data parsed from it (see 'mkAnnouncingHeader'). The 'Eq' instance +-- compares by header hash: that is the one identity used for announcement dedup +-- and equivocation counting (see 'Announcements.onAnnouncement'). +data AnnouncingHeader blk = + -- | INVARIANT: 'ancHeader' includes an announcement whose fields are + -- 'ancAnnouncementFields' + UnsafeMkAnnouncingHeader { + ancHeader :: !(Header blk) + , + ancAnnouncementFields :: !AnnouncementFields + } + +instance HasHeader (Header blk) => Eq (AnnouncingHeader blk) where + a == b = headerHash (ancHeader a) == headerHash (ancHeader b) + +-- | Interpret a header as a relayed EB announcement, or 'Nothing' if it carries +-- no announcement (so it should not have been relayed as one). Parsing the +-- announcement once here keeps it total for all later consumers (e.g. tracing). +mkAnnouncingHeader :: ResolveLeiosBlock blk => Header blk -> Maybe (AnnouncingHeader blk) +mkAnnouncingHeader h = + headerLeiosAnnouncement h <&> \(MkLeiosPoint _ebSlot ebHash, ebBodySize) -> + UnsafeMkAnnouncingHeader h (MkAnnouncementFields (headerElId h) ebHash ebBodySize) + +-- | The election of an 'AnnouncingHeader'. +ancElId :: AnnouncingHeader blk -> ElId +ancElId = announcementElection . ancAnnouncementFields + +-- | Thrown when a peer misbehaves on the announcement protocol; the ensuing +-- thread death disconnects the peer. It carries the +-- 'Announcements.ErrAnnouncement' verbatim (the @blk@ is existential); every +-- such error is a disconnect, since the only invalidities that used to be +-- tolerated — opcert issue numbers ahead of the immutable tip — are now +-- accepted outright by 'validateAnnouncementHeader'. +data ExnInvalidLeiosAnnouncement = + forall blk. + ReactToAnnouncementError (Announcements.ErrAnnouncement (AnnouncementInvalidity blk)) + +deriving instance Show ExnInvalidLeiosAnnouncement + +instance Exception ExnInvalidLeiosAnnouncement + +-- | Thrown when a peer relays a 'MsgLeiosBlockAnnouncement' whose header carries +-- no EB announcement (so 'mkAnnouncingHeader' returns 'Nothing'); the ensuing thread +-- death disconnects the peer. +data ExnLeiosBlockAnnouncementMissing = ExnLeiosBlockAnnouncementMissing + deriving Show + +instance Exception ExnLeiosBlockAnnouncementMissing + +-- | The @validate@ callback for 'Announcements.onAnnouncement'. +-- +-- First apply ChainSync's in-future check to the announced slot's wall-clock +-- onset (reusing the node's own 'InFutureCheck.SomeHeaderInFutureCheck'): +-- a far-future slot raises 'InFutureCheck.HeaderArrivalException' (disconnecting +-- the peer), a near-future slot blocks until the slot's onset (Ouroboros +-- Chronos) — blocking the per-peer handler is acceptable, as a (near-)future +-- announcement is the peer's fault. +-- +-- Returns the announcement's data and whether to relay it downstream (see +-- 'Announcements.ShouldRelay' and 'maxAnnouncementAgeSend'), if the announcement +-- is valid. Per the 'Announcements.onAnnouncement' contract, 'Left Nothing' +-- signals the too-old rejection (see 'maxAnnouncementAgeRecv') and 'Left Just' +-- any other invalidity. +announcementValidity :: + (IOLike m, LedgerSupportsProtocol blk, ResolveLeiosBlock blk) => + SystemTime m -> + InFutureCheck.SomeHeaderInFutureCheck m blk -> + TopLevelConfig blk -> + ExtLedgerState blk EmptyMK -> + Header blk -> + m + ( Either + (Maybe (AnnouncementInvalidity blk)) + (Announcements.ShouldRelay, NominalDiffTime, (LeiosPoint, BytesSize)) + ) +announcementValidity systemTime futureCheck cfg immLedger hdr = do + onset <- case futureCheck of + InFutureCheck.SomeHeaderInFutureCheck hifc -> do + arrival <- InFutureCheck.recordHeaderArrival hifc hdr + judgment <- + either throwIO pure $ + runExcept $ + InFutureCheck.judgeHeaderArrival + hifc + (configLedger cfg) + (ledgerState immLedger) + arrival + arrivalResult <- InFutureCheck.handleHeaderArrival hifc judgment + either throwIO pure (runExcept arrivalResult) + -- The in-future check has delayed this thread until 'onset' if the + -- slot was near-future, so 'now' is at or after 'onset' and the age + -- is non-negative. + now <- systemTimeCurrent systemTime + let age = diffRelTime now onset + pure $ + -- 'Left Nothing' signals the too-old rejection to 'Announcements.onAnnouncement' + -- (which raises 'Announcements.ErrTooOld'); only this function holds the wall + -- clock, so it owns that check. + if age > maxAnnouncementAgeRecv + then Left Nothing + else + let shouldRelay = + if age <= maxAnnouncementAgeSend + then Announcements.DoRelay + else Announcements.DoNotRelay + in case validateAnnouncementHeader cfg immLedger hdr of + Left inv -> Left (Just inv) + Right v -> Right (shouldRelay, age, v) + +-- | Record a validated, newly-announced EB body as missing, with its +-- authoritative (forger-signed) size. First-seen wins: a no-op if the body is +-- already acquired or already recorded. +recordAnnouncedEb :: + IOLike m => + ( MVar m (LeiosOutstanding pid) + , MVar m () + ) -> + (LeiosPoint, BytesSize) -> + m () +recordAnnouncedEb (outstandingVar, readyVar) (point, ebBytesSize) = do + changed <- MVar.modifyMVar outstandingVar (pure . upd) + when changed $ void $ MVar.tryPutMVar readyVar () + where + MkLeiosPoint _ebSlot ebHash = point + + upd outstanding = if Set.member ebHash (Leios.acquiredEbBodies outstanding) + || any ((== ebHash) . pointEbHash) (Map.keys (Leios.missingEbBodies outstanding)) + then (outstanding, False) + else flip (,) True $ + outstanding + { Leios.missingEbBodies = + Map.insert point ebBytesSize (Leios.missingEbBodies outstanding) + } + +prunePeerStateToImmTip :: + LedgerSupportsProtocol blk => + ExtLedgerState blk EmptyMK -> + SlotNo -> + Announcements.PeerState anc -> + (SlotNo, Announcements.PeerState anc) +prunePeerStateToImmTip immLedger latestPruneSlot peerSt = + case getTipSlot (ledgerState immLedger) of + NotOrigin immTipSlot + | latestPruneSlot < immTipSlot -> (immTipSlot, Announcements.prunePeerState immTipSlot peerSt) + _ -> (latestPruneSlot, peerSt) + +-- | The just-counted announcement's fields, and whether it equivocates a prior +-- header announcing the same election. +announcementTraceFields :: + Announcements.ElState (AnnouncingHeader blk) -> + (AnnouncementEquivocation, AnnouncementFields) +announcementTraceFields = \case + Announcements.OneAnnouncement a -> + (NoEquivocation, ancAnnouncementFields a) + Announcements.TwoAnnouncements _a1 a2 -> + (Equivocation, ancAnnouncementFields a2) + +-- | Render an 'Announcements' per-peer announcement event as a 'TraceLeiosPeer'. +tracePeerAnnouncement :: + Announcements.TraceLeiosNotifyPeerEvent (AnnouncingHeader blk) -> + TraceLeiosPeer +tracePeerAnnouncement (Announcements.TracePeerAnnouncement elSt) = + let (equivocation, fields) = announcementTraceFields elSt + in TraceLeiosPeerAnnouncement equivocation fields + +-- | Render an 'Announcements' node-wide announcement event as a +-- 'TraceLeiosKernel'. +traceNewAnnouncement :: + Announcements.TraceLeiosNotifyEvent peer (AnnouncingHeader blk) -> + TraceLeiosKernel +traceNewAnnouncement (Announcements.TraceNewAnnouncement mbPeer _elId elSt age) = + let (equivocation, fields) = announcementTraceFields elSt + in TraceLeiosAnnouncementAccepted + (maybe ForgedLocally (const ReceivedFromPeer) mbPeer) + equivocation + fields + age + +lEIOSNOTIFYPIPELINEDEPTH :: Int +lEIOSNOTIFYPIPELINEDEPTH = 100 -- TODO magic number + +-- | Do not relay (to downstream peers) an announcement whose slot's wall-clock +-- onset is older than this. See 'Announcements.ShouldRelay'. +-- +-- Must be comfortably less than 'maxAnnouncementAgeRecv', so that an +-- announcement an honest node relays just before this bound still arrives at +-- the downstream peer within that peer's larger receive bound, even after +-- transmission time and clock skew. +-- +-- TODO magic number; should be a config/RunNode option +maxAnnouncementAgeSend :: NominalDiffTime +maxAnnouncementAgeSend = 300 -- 5 minutes + +-- | Disconnect an upstream peer that relays an announcement whose slot's +-- wall-clock onset is older than this. See 'Announcements.ErrTooOld'. +-- +-- Comfortably greater than 'maxAnnouncementAgeSend', so that an honest peer +-- (which stops relaying at that smaller bound) is never disconnected on account +-- of transmission time or clock skew. +-- +-- TODO magic number; should be a config/RunNode option... or even a protocol +-- parameter? +maxAnnouncementAgeRecv :: NominalDiffTime +maxAnnouncementAgeRecv = 600 -- 10 minutes diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements.hs new file mode 100644 index 0000000000..e585bee16e --- /dev/null +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements.hs @@ -0,0 +1,378 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} + +{- | + +TO BE IMPORTED QUALIFIED + +The Leios node must relay EbAnnouncements promptly. +CIP-0164 is designed around the idea that an EbAnnouncement at an honest stake pool will have propagated to all other honest stake pools within L_hdr, and suggests a value of 1 second. +On a network that spans the globe, that's a very tight window. +But each EbAnnouncement is slightly less than 1 kB, so it does seem plausible. + +In the current design of CIP-0164, the existing Praos header (aka RbHeader) was extended with new fields. +One of those fields is an optional tuple: EbBody hash and EbBody size. +That's the announcement. +As part of the RbHeader, it is adjacent to the election proof that justifies the EbAnnouncement and it is signed by the electee. +Because that election proof necessarily includes the slot of the election, every EbAnnouncement has an age. +In particular, that age is used to evict too-old announcements so that the existing stochastic bound on the number of elections younger than some age also bounds the number of in-memory EbAnnouncements: zero (no announcement), one (unequivocal announcement), or two (equivocal announcements) per sufficiently-young election. + +The LeiosNotify mini protocol in CIP-0164 already includes a notification MsgLeiosAnnouncement, whose payload is the RbHeader. +It is sent as one of the possible responses to the generic MsgLeiosNotificationRequestNext message. +The healthy honest node tries to maintain ~hundreds of those requests outstanding at all times, so that the upstream peer can always enqueue a notification immediately. +The intended timeline one a single connection from an honest node X to an honest node Y is as follows. + +- X either receives or itself issues a new valid EbAnnouncement. +- If that's the first or second announcement X has seen for that election, then X tries to relay that announcement to its downstream peers, including Y. +- X will be able to send to Y, because X should have received at least one more MsgLeiosNotificationRequestNext message from Y than X has already replied to. +- When Y receives the announcement from X, it confirms that X hasn't already sent this announcement or any two distinct (equivocating) announcements for that same election, disconnecting if it has. +- Y also needs to validate that the announcement is well-signed and that its election proof is valid, disconnecting if either is invalid. + +TO BE IMPORTED QUALIFIED + +-} +module LeiosDemoLogic.Announcements (module LeiosDemoLogic.Announcements) where + +import Cardano.Slotting.Slot (SlotNo) +import Data.Time.Clock (NominalDiffTime) +import Control.Concurrent.Class.MonadSTM (MonadSTM, atomically) +import Control.Concurrent.Class.MonadSTM.Strict.TVar (StrictTVar, readTVar, writeTVar) +import Control.Monad (foldM, void) +import Control.Monad.Except (ExceptT, throwError) +import Control.Monad.Trans (lift) +import Control.Tracer (Tracer, traceWith) +import Data.Functor.Compose (Compose (..)) +import qualified Data.Map.Strict as Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Set (Set) +import qualified Data.Set as Set +import LeiosDemoLogic.Announcements.ElBimap + +----- +-- The state of each LeiosNotify client +----- + +-- | State maintained for a single LeiosNotify upstream peer +data PeerState anc = + MkPeerState { + live :: !(Strict.Map ElId (ElState anc)) + } + +emptyPeerState :: PeerState anc +emptyPeerState = MkPeerState Map.empty + +-- | State maintained within 'PeerState' for each election +data ElState anc = + -- | The peer has sent one announcement + -- + -- INVARIANT: has valid signature and election proof + OneAnnouncement !anc + | + -- | The peer has sent two announcements + -- + -- INVARIANT: the left announcement was received first + -- + -- INVARIANT: both have valid signature and election proof + -- + -- INVARIANT: they are different announcements + TwoAnnouncements !anc !anc + +firstAnnouncement :: ElState anc -> anc +firstAnnouncement = \case + OneAnnouncement x -> x + TwoAnnouncements x1 _x2 -> x1 + +secondAnnouncement :: ElState anc -> Maybe anc +secondAnnouncement = \case + OneAnnouncement _x -> Nothing + TwoAnnouncements _x1 x2 -> Just x2 + +-- | Called whenever the ChainDB's immutable tip advances to a new slot +-- +-- NOTE this pruning should happen ~60 s after the immutable tip +-- advances. That accommodates clock skew, transmission time, etc +-- with only an insignificant increase in the stochastic bound on +-- election count +prunePeerState :: SlotNo -> PeerState anc -> PeerState anc +prunePeerState immTipSlot st = + MkPeerState { live = live' } + where + (_pruned, live') = Map.spanAntitone tooOld (live st) + + -- NB strict comparison, so that the immtip's own announcement is + -- not pruned + tooOld (MkElId elSlot _poolId) = elSlot < immTipSlot + +-- | Behaviors of a LeiosNotify upstream peer's announcement stream +-- that an honest node rejects +data ErrAnnouncement invalidity = + -- | The peer had already sent this same announcement before + ErrRepeat + | + -- | The peer had already sent two announcements for this election + ErrThird + | + -- | This announcement is invalid + ErrInvalid !invalidity + | + -- | This announcement's election is too old (its wall-clock age exceeds + -- what a client tolerates from an upstream peer). Unlike 'ErrInvalid', this + -- verdict depends on the wall clock rather than on the announcement itself, + -- so 'onAnnouncement' cannot decide it directly; it is signalled by the + -- validation callback returning @Left Nothing@ (see 'onAnnouncement'). + ErrTooOld + deriving Show + +-- | Returns 'Nothing' if this election was already 'TwoAnnouncements' or if this +-- announcement was already received +extendLive :: + forall anc invalidity. + Eq anc => + ElId -> + anc -> + PeerState anc -> + Either (ErrAnnouncement invalidity) (ElState anc, PeerState anc) +extendLive elId anc st = + getCompose + $ fmap + (\live' -> MkPeerState { live = live' }) + (Map.alterF (inj . upd) elId (live st)) + where + -- 0 -> 1 ok + -- 1 -> 2 ok when unequal + -- 2 -> 3 not ok + upd :: Maybe (ElState anc) -> Either (ErrAnnouncement x) (ElState anc) + upd = \case + Nothing -> Right $ OneAnnouncement anc + Just (OneAnnouncement x) -> + if x == anc then Left ErrRepeat else + Right $ TwoAnnouncements x anc + Just TwoAnnouncements{} -> Left ErrThird + + inj :: Functor f => f a -> Compose f ((,) a) (Maybe a) + inj = Compose . fmap (\x -> (x, Just x)) -- success is never a delete + +----- +-- The per-peer logic for receiving a MsgLeiosAnnouncement +----- + +data TraceLeiosNotifyPeerEvent anc = + TracePeerAnnouncement !(ElState anc) + +onAnnouncement :: + (Eq anc, Monad m) => + Tracer m (TraceLeiosNotifyPeerEvent anc) -> + (anc -> ElId) -> + (anc -> m (Either (Maybe invalidity) validated)) -> + -- ^ How to validate the announcement + -- + -- Return @Left Nothing@ when the announcement violates the too-old + -- restriction (its election's wall-clock age exceeds what a client + -- tolerates); 'onAnnouncement' then raises 'ErrTooOld'. This callback owns + -- that check because only it holds the wall clock. Return @Left (Just inv)@ + -- for any other rejection (raised as 'ErrInvalid'), and @Right@ when valid. + (anc -> validated -> m ()) -> + -- ^ How this central logic should react to a new announcement from + -- this peer + -- + -- ASSUMPTION: this is often a no-op (except maybe tracing), because + -- the same announcement has already been received from other peers. + PeerState anc -> + anc -> + ExceptT (ErrAnnouncement invalidity) m (PeerState anc) +onAnnouncement tracer getEl validate process st anc = do + (elSt, st') <- case extendLive (getEl anc) anc st of + Left err -> throwError err + Right x -> pure x + lift $ traceWith tracer $ TracePeerAnnouncement elSt + -- do the more expensive validation only after the trivial + -- counting checks + lift (validate anc) >>= \case + Left Nothing -> throwError ErrTooOld + Left (Just err) -> throwError $ ErrInvalid err + Right x -> do + lift $ process anc x + pure st' + +----- +-- The central logic for receiving a MsgLeiosAnnouncement +----- + +-- | State maintained for a node's own LeiosNotify behaviors +data CentralState m peer anc = + MkCentralState { + -- | An isomorph of `PeerState` that models the node itself as + -- an upstream peer of its downstream peers. This allows it to + -- ignore an announcement exactly when sending it would cause + -- the recipient's 'extendLive' to raise an error. + selfPeer :: !(PeerState anc) + , + queues :: !(Strict.Map peer (QueueAnnouncementView m anc)) + , + -- | Which peers we have already sent announcements for this + -- election + -- + -- We only send equivocation proofs to them. + -- + -- We don't need to track whether or not we've sent them an + -- equivocation. + gate :: !(ElBimap peer) + } + +emptyCentralState :: CentralState m peer anc +emptyCentralState = MkCentralState emptyPeerState Map.empty emptyElBimap + +-- | A downstream peer's send queue and its count of available credits +data QueueAnnouncementView m anc = + forall q. + MkQueueAnnouncementView + !(StrictTVar m Int) + !(q -> anc -> q) + !(StrictTVar m q) + +data TraceLeiosNotifyEvent peer anc = + -- | The final field is how late the announcement was — seconds from the + -- election slot's wall-clock onset to when this node counted it — when + -- known (a relayed announcement has it; a locally-forged one does not). + TraceNewAnnouncement !(Maybe peer) !ElId !(ElState anc) !(Maybe NominalDiffTime) + +-- | Called whenever the ChainDB's immutable tip advances to a new slot +-- +-- Unlike 'prunePeerState', a delay is undesirable here. +pruneCentralState :: + Ord peer => SlotNo -> CentralState m peer anc -> CentralState m peer anc +pruneCentralState immTipSlot st = + MkCentralState { + selfPeer = prunePeerState immTipSlot (selfPeer st) -- NB no delay + , + queues = queues st + , + gate = pruneElBimap immTipSlot (gate st) + } + +insertPeerCentral :: + Ord peer => + peer -> + QueueAnnouncementView m anc -> + CentralState m peer anc -> + CentralState m peer anc +insertPeerCentral peer qav st = + MkCentralState { + selfPeer = selfPeer st + , + queues = Map.insert peer qav (queues st) + , + gate = gate st + } + +deletePeerCentral :: + Ord peer => + peer -> + CentralState m peer anc -> + CentralState m peer anc +deletePeerCentral peer st = + MkCentralState { + selfPeer = selfPeer st + , + queues = Map.delete peer (queues st) + , + gate = deleteElBimapR peer (gate st) + } + +-- | Whether 'onAnnouncementCentral' should relay an announcement downstream. +-- +-- An honest node uses 'DoNotRelay' once an announcement's slot is old enough +-- that a peer whose immutable tip has advanced slightly further would +-- disconnect the relayer for relaying a below-its-immutable-tip announcement; +-- the caller decides this from the announcement's wall-clock age. Local +-- processing is unaffected either way. +data ShouldRelay = DoRelay | DoNotRelay + deriving (Eq, Show) + +-- | The nub of the callback argument to 'onAnnouncement' +-- +-- NOTE: This is also called by the block forging thread when this node issues +-- its own announcement (with 'Nothing' as the source peer). +-- +-- TODO: headers arriving via the ChainSync 'MsgRollForward' carry announcements +-- too, and so should also feed the 'CentralState' (for relay and dedup); they +-- do not yet. +onAnnouncementCentral :: + forall m peer anc. + (MonadSTM m, Ord peer, Eq anc) => + Tracer m (TraceLeiosNotifyEvent peer anc) -> + (anc -> ElId) -> + (ElState anc -> m ()) -> + -- ^ Notify other components about a /new/ announcement + -- + -- For example, notify the voting thread; it cares about both new + -- announcements and also equivocations. + -- + -- Maybe also update a cache used to dedup header validation + -- computations. Etc. + CentralState m peer anc -> + Maybe peer -> + -- ^ The upstream peer the announcement came from, or 'Nothing' if this node + -- is itself the source (e.g. its own block forging). + ShouldRelay -> + Maybe NominalDiffTime -> + -- ^ How late the announcement was (see 'TraceNewAnnouncement'). + anc -> + m (CentralState m peer anc) +onAnnouncementCentral tracer getEl publishLocally st peer shouldRelay age anc = + case extendLive el anc (selfPeer st) of + Left{} -> pure st -- complete noop for duplicates + Right (elSt, selfPeer') -> do + traceWith tracer $ TraceNewAnnouncement peer el elSt age + -- urgently relay + newPeers <- case shouldRelay of + DoNotRelay -> pure Set.empty + DoRelay -> send elSt (queues st) + -- signal other components + publishLocally elSt + pure MkCentralState { + selfPeer = selfPeer' + , + queues = queues st + , + gate = insertElBimapLs el newPeers (gate st) + } + where + el = getEl anc + + -- returns new peers to add to 'gate' for this election + send :: ElState anc -> Strict.Map peer (QueueAnnouncementView m anc) -> m (Set peer) + send elSt qs = case elSt of + OneAnnouncement{} -> foldM enqueue Set.empty (Map.assocs qs) + TwoAnnouncements{} -> do + -- Only send equivocation proofs to peers we've already + -- sent the first announcement to. Otherwise they'd be + -- interpreting it as /our/ /first/ announcement for that + -- election. TODO introduce a single message for the pair + -- OR send both announcements if they currently have at + -- least two credits. + -- + -- And 'gate' doesn't change. + mapM_ + (void . tryEnqueue) + (Map.restrictKeys qs $ lookupElBimapL el $ gate st) + pure Set.empty + + enqueue :: Set peer -> (peer, QueueAnnouncementView m anc) -> m (Set peer) + enqueue !acc (peer', ancQueue) = + fmap + (\enqueued -> if not enqueued then acc else Set.insert peer' acc) + (tryEnqueue ancQueue) + + tryEnqueue :: QueueAnnouncementView m anc -> m Bool + tryEnqueue (MkQueueAnnouncementView free snoc tvar) = atomically $ do + n <- readTVar free + -- just drop it if there are currently no credits + if n <= 0 then pure False else do + writeTVar free $! n - 1 + q <- readTVar tvar + writeTVar tvar $! snoc q anc + pure True diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/ElBimap.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/ElBimap.hs new file mode 100644 index 0000000000..f378564df8 --- /dev/null +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/ElBimap.hs @@ -0,0 +1,136 @@ +-- | A bidirectional map, both halves kept strict, relating each +-- `ElId` to a set of `a` and back. +-- +module LeiosDemoLogic.Announcements.ElBimap (module LeiosDemoLogic.Announcements.ElBimap) where + +import Cardano.Slotting.Slot (SlotNo) +import qualified Data.ByteString.Base16 as BS16 +import qualified Data.ByteString.Char8 as BS8 +import Data.ByteString.Short (ShortByteString, fromShort) +import qualified Data.Foldable +import qualified Data.Map.Strict as Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Set (Set) +import qualified Data.Set as Set +import Data.Set.NonEmpty (NESet) +import qualified Data.Set.NonEmpty as NESet + +-- | The slot number and pool id of an election +data ElId = + -- | The bytes are the hash of the pool's cold verification key + -- + -- TODO should it be the hash, or just the key itself? + MkElId !SlotNo !ShortByteString + deriving (Eq) + +-- | Renders the pool id as hex rather than as a raw byte string. +instance Show ElId where + showsPrec p (MkElId slot poolId) = + showParen (p > 10) $ + showString "MkElId " + . showsPrec 11 slot + . showChar ' ' + . showString (BS8.unpack (BS16.encode (fromShort poolId))) + +-- | INVARIANT: lexicographically starts with 'SlotNo' +instance Ord ElId + where + compare (MkElId slL idL) (MkElId slR idR) = + compare slL (slR :: SlotNo) + <> compare idL idR + +-- | A many-to-many relation between 'ElId' (left) and @a@ (right), +-- indexed in both directions +-- +-- INVARIANT: the halves agree --- @r@ occurs in @forwardHalf ! l@ iff +-- @l@ occurs in @inverseHalf ! r@. +data ElBimap a = + MkElBimap { + forwardHalf :: !(Strict.Map ElId (NESet a)) + , + inverseHalf :: !(Strict.Map a (NESet ElId)) + } + deriving (Show) + +emptyElBimap :: ElBimap a +emptyElBimap = MkElBimap Map.empty Map.empty + +insertElBimap :: Ord a => ElId -> a -> ElBimap a -> ElBimap a +insertElBimap l r bm = + MkElBimap { + forwardHalf = Map.insertWith (<>) l (NESet.singleton r) (forwardHalf bm) + , + inverseHalf = Map.insertWith (<>) r (NESet.singleton l) (inverseHalf bm) + } + +-- | Insert an election paired with each element of a set +insertElBimapLs :: Ord a => ElId -> Set a -> ElBimap a -> ElBimap a +insertElBimapLs l rs bm = + case NESet.nonEmptySet rs of + Nothing -> bm + Just rs' -> MkElBimap { + forwardHalf = Map.insertWith (<>) l rs' (forwardHalf bm) + , + inverseHalf = Data.Foldable.foldl' addToInverse (inverseHalf bm) rs + } + where + addToInverse inv r = Map.insertWith (<>) r (NESet.singleton l) inv + +lookupElBimapL :: ElId -> ElBimap a -> Set a +lookupElBimapL l bm = + maybe Set.empty NESet.toSet + $ Map.lookup l + $ forwardHalf bm + +lookupElBimapR :: Ord a => a -> ElBimap a -> Set ElId +lookupElBimapR r bm = + maybe Set.empty NESet.toSet + $ Map.lookup r + $ inverseHalf bm + +-- | Remove an election and every pair it was part of +deleteElBimapL :: Ord a => ElId -> ElBimap a -> ElBimap a +deleteElBimapL l bm = + case Map.lookup l (forwardHalf bm) of + Nothing -> bm + Just rs -> MkElBimap { + forwardHalf = Map.delete l (forwardHalf bm) + , + inverseHalf = Data.Foldable.foldl' dropFromInverse (inverseHalf bm) rs + } + where + dropFromInverse inv r = Map.update (NESet.nonEmptySet . NESet.delete l) r inv + +-- | Remove a right key and every pair it was part of +deleteElBimapR :: Ord a => a -> ElBimap a -> ElBimap a +deleteElBimapR r bm = + case Map.lookup r (inverseHalf bm) of + Nothing -> bm + Just ls -> MkElBimap { + forwardHalf = Data.Foldable.foldl' dropFromForward (forwardHalf bm) ls + , + inverseHalf = Map.delete r (inverseHalf bm) + } + where + dropFromForward fwd l = Map.update (NESet.nonEmptySet . NESet.delete r) l fwd + +-- | Remove every election older than the given slot and every pair +-- those elections were part of +-- +-- Called whenever the ChainDB's immutable tip advances to a new slot. +pruneElBimap :: Ord a => SlotNo -> ElBimap a -> ElBimap a +pruneElBimap immTipSlot bm = + MkElBimap { + forwardHalf = forwardHalf' + , + inverseHalf = Map.foldlWithKey' dropElection (inverseHalf bm) pruned + } + where + (pruned, forwardHalf') = Map.spanAntitone tooOld (forwardHalf bm) + + dropElection inv el rs = Data.Foldable.foldl' (dropPair el) inv rs + + dropPair el inv r = Map.update (NESet.nonEmptySet . NESet.delete el) r inv + + -- NB strict comparison, so that the immtip's announcement remains + tooOld (MkElId elSlot _poolId) = elSlot < immTipSlot diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/Validate.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/Validate.hs new file mode 100644 index 0000000000..2ff90a9369 --- /dev/null +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoLogic/Announcements/Validate.hs @@ -0,0 +1,195 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} + +-- | Validation of a relayed Leios EB announcement, which is carried as an RB +-- 'Header'. +-- +-- The announcement can arrive out-of-order with respect to the local chain, so we +-- run only the /protocol-level/ header validation (see 'validateHeaderProtocol') +-- against the immutable tip's ledger state, forecast to the announced slot; the +-- envelope (chain-extension) check is deliberately skipped. As of Dijkstra the +-- protocol-level header rules already fold in the announcement's checks (the +-- EbBody size bound etc.), so this is the whole announcement validation. +-- +-- For a caught-up node the immutable tip is always within the forecast horizon +-- of a fresh announcement (by appeal to Praos Chain Growth); still-syncing +-- nodes do not request LeiosNotify notifications, so they never reach here. +module LeiosDemoLogic.Announcements.Validate + ( AnnouncementInvalidity (..) + , validateAnnouncementHeader + ) where + +import Control.Monad (when) +import Control.Monad.Except (runExcept, throwError, withExcept) +import LeiosDemoTypes (LeiosPoint, BytesSize) +import Ouroboros.Consensus.Block + ( BlockProtocol + , Header + , WithOrigin (NotOrigin) + , blockSlot + , validateView + ) +import Ouroboros.Consensus.Config + ( TopLevelConfig + , configBlock + , configConsensus + , configLedger + ) +import Ouroboros.Consensus.Forecast (OutsideForecastRange, forecastFor) +import Ouroboros.Consensus.HeaderValidation + ( ValidateEnvelope (..) + , tickHeaderState + , tickedHeaderStateChainDep + ) +import Ouroboros.Consensus.Ledger.Abstract (getTipSlot) +import Ouroboros.Consensus.Ledger.Basics (EmptyMK) +import Ouroboros.Consensus.Ledger.Extended (ExtLedgerState (..)) +import Ouroboros.Consensus.Ledger.SupportsProtocol + ( LedgerSupportsProtocol + , ledgerViewForecastAt + ) +import Ouroboros.Consensus.Protocol.Abstract (ValidationErr) +import Ouroboros.Consensus.Storage.LedgerDB.Forker + ( ResolveLeiosBlock + , headerLeiosAnnouncement + , validateAnnouncementChainDepState + ) + +-- | Reasons a LeiosNotify client would reject its upstream peer's announcement, +-- regardless of other messages they had sent +data AnnouncementInvalidity blk + = -- | The announced slot is beyond the forecast horizon from the immutable + -- tip. This should not occur for a caught-up node (by appeal to Praos Chain + -- Growth), so it is either a bogus far-future slot or we are falling behind; + -- either way disconnecting is acceptable (if we are behind, then either we + -- are unhealthy or the peer is not serving us well). We cannot simply ignore + -- it, because there is an unbounded supply of far-future slots. + OutsideHorizon !OutsideForecastRange + | -- | The announced slot is before the immutable tip (the forecast anchor), + -- so it cannot be forecast\/validated at all. This is a separate case from + -- 'OutsideHorizon' because 'forecastFor' does not report a below-anchor + -- slot as 'OutsideForecastRange' (that only bounds the future + -- end)---instead, a below-anchor slot violates 'forecastFor''s + -- precondition. For a caught-up node a slot so stale is as bogus as one in + -- the far-future. + -- + -- See 'LeiosDemoLogic.Announcements.ShouldRelay' for how honest + -- servers avoid triggering this case despite clock + -- skew\/transmission delays\/buffering, etc. + -- + -- The 'LeiosDemoLogic.Announcements.ErrTooOld' bound rejects any + -- announcement older than 'maxAnnouncementAgeRecv' (minutes), which is + -- always far younger than the immutable tip (hours), so this case should + -- never actually fire. It is kept as a distinct error to throw in case that + -- branch is somehow reached, eg on an oddly-configured testnet. + SlotBeforeImmutableTip + | -- | The header failed the relaxed, out-of-context protocol-level validation + -- (election proof and signature). See 'validateAnnouncementChainDepState' + -- for which check is relaxed (the OCIN upper bound) and + -- 'validateAnnouncementHeader' for which are skipped (the chain-extension + -- envelope checks). + HeaderInvalid !(ValidationErr (BlockProtocol blk)) + | -- | The RbHeader failed an envelope check — chiefly, it exceeds the + -- protocol's max header size (see 'validateAnnouncementHeader'). + RbHeaderEnvelopeInvalid !(OtherHeaderEnvelopeError blk) + | -- | The header carries no EB announcement, so it should not have been + -- relayed as a 'MsgLeiosBlockAnnouncement' at all. + NoAnnouncement + +-- | NB 'HeaderInvalid' and 'RbHeaderEnvelopeInvalid' do not render their +-- wrapped errors, so that this instance is unconstrained in @blk@ (avoiding +-- @Show@ constraints on those errors that would have to be threaded through the +-- node). The wrapped values still carry them. +instance Show (AnnouncementInvalidity blk) where + show ai = case ai of + OutsideHorizon r -> "OutsideHorizon (" <> show r <> ")" + SlotBeforeImmutableTip -> "SlotBeforeImmutableTip" + HeaderInvalid{} -> "HeaderInvalid " + RbHeaderEnvelopeInvalid{} -> "RbHeaderEnvelopeInvalid " + NoAnnouncement -> "NoAnnouncement" + +-- | Protocol-level validation of an announced RB 'Header' against the immutable +-- tip's ledger state (forecast to the header's slot). Envelope check skipped; +-- see the module header. +-- +-- The operational certficiate (opcert) issue number (OCIN) is checked only as a +-- lower bound: any number at least the immutable tip's counter is accepted (the +-- over-increment upper bound, which the strict protocol check would enforce, is +-- skipped — see 'validateAnnouncementChainDepState' and +-- 'WhetherToUpperBoundOCERT'), and a lower one is rejected as a revoked key. +-- +-- OCINs are otherwise ignored. In effect, this logic is assuming that all OCINs +-- are controlled by the pool owner. That's patentedly contrary the intended +-- purpose of OCINs, so it needs justification; hence this comment. +-- +-- The crux is a Catch 22 if we consider different OCINs as different +-- identities. We must either treat all of those identities +-- _independently_ (ie as distinct elections) or _prioritize_ the +-- greater OCINs (which seems intuitive). The problem is that the +-- adversary can create arbitrarily many OCINs for its own pools. And +-- then it can abuse either choice we make: either it gets to multiply +-- the Leios load on the network per election, or it can cause +-- arbitrary "partitions" of the network, with one clique certifying a +-- lower OCIN's announcement but the other clique completely ignoring +-- that announcement. +-- +-- The current behavior is to accept (and relay!) any OCIN at least as +-- great as the counter in our immutable tip's ledger state. The only +-- downside to this is that an increment OCIN doesn't revoke the old +-- opcert _for Leios_ until the increment is on the immutable tip +-- (Praos is still immediate). So a leaked hot key means the attacker +-- can equivocate all of the victim's announcements until the victim +-- notices, lands a new opcert on chain, and then waits for that +-- opcert to become immutable (~12 hr, <= ~36 hr). Not ideal, but +-- tolerable. +-- +-- Returns the output of 'headerLeiosAnnouncement'. +validateAnnouncementHeader :: + forall blk. + ( LedgerSupportsProtocol blk + , ResolveLeiosBlock blk + ) => + TopLevelConfig blk -> + ExtLedgerState blk EmptyMK -> + Header blk -> + Either (AnnouncementInvalidity blk) (LeiosPoint, BytesSize) +validateAnnouncementHeader cfg extLedger hdr = + runExcept $ do + x <- case headerLeiosAnnouncement hdr of + Nothing -> throwError NoAnnouncement + Just x -> pure x + -- 'forecastFor' does not reject a slot below its anchor (the immutable + -- tip's slot) — that is a precondition violation, not 'OutsideForecastRange' + -- — so guard it explicitly before forecasting. + when (NotOrigin slot < getTipSlot (ledgerState extLedger)) $ + throwError SlotBeforeImmutableTip + ledgerView <- + withExcept OutsideHorizon $ + forecastFor + (ledgerViewForecastAt (configLedger cfg) (ledgerState extLedger)) + slot + -- Reject an RbHeader bigger than the protocol allows: it is the + -- announcement's transmission unit. This is the full Shelley envelope check, + -- so it also bounds the declared RB body size and rejects an obsolete node. + -- Those two are harmless extras — a legitimate announcement's RbHeader always + -- passes them — and, being LedgerView-derived, they are meaningful + -- out-of-context (unlike the chain-extension checks, which we skip). + withExcept RbHeaderEnvelopeInvalid $ + additionalEnvelopeChecks cfg ledgerView hdr + -- The out-of-context, relaxed protocol-level validation: the election proof + -- and the signature, but not the RB-header-specific checks nor the checks + -- that our lagging tip would spuriously trip (see + -- 'validateAnnouncementChainDepState'). Any error it returns is a genuine + -- rejection. + let tickedHeaderState = + tickHeaderState (configConsensus cfg) ledgerView slot (headerState extLedger) + withExcept HeaderInvalid $ + validateAnnouncementChainDepState @blk + (configConsensus cfg) + (validateView (configBlock cfg) hdr) + slot + (tickedHeaderStateChainDep tickedHeaderState) + pure x + where + slot = blockSlot hdr diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoOnlyTestNotify.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoOnlyTestNotify.hs index f97a37abcb..67d2cf7fed 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoOnlyTestNotify.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoOnlyTestNotify.hs @@ -6,11 +6,13 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PolyKinds #-} +{-# LANGUAGE QuantifiedConstraints #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE StandaloneKindSignatures #-} {-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TypeOperators #-} module LeiosDemoOnlyTestNotify ( LeiosNotify (..) @@ -23,10 +25,15 @@ module LeiosDemoOnlyTestNotify , timeLimitsLeiosNotify , LeiosNotifyClientPeerPipelined , LeiosNotifyServerPeer + , LeiosNotifyServerPeerLookahead , leiosNotifyClientPeer , leiosNotifyClientPeerPipelined , leiosNotifyServerPeer + , leiosNotifyServerPeerLookahead + , WhetherExcessiveRequests (..) , toLeiosNotifyClientPeerPipelined + + , runLookaheadFixedSenderPeerWithLimits ) where import qualified Codec.CBOR.Decoding as CBOR @@ -65,16 +72,20 @@ import Network.TypedProtocol.Core , Nat (..) , Protocol (..) , ReflRelativeAgency (..) + , SenderVariability (..) , StateAgency , natToInt ) import Network.TypedProtocol.Peer ( Peer (..) + , PeerLookaheadFixedSender (..) , PeerPipelined (..) , Receiver (..) + , Sender (..) ) import Ouroboros.Network.Protocol.Limits - ( ProtocolSizeLimits (..) + ( BearerBytes + , ProtocolSizeLimits (..) , ProtocolTimeLimits (..) , smallByteLimit , waitForever @@ -82,6 +93,18 @@ import Ouroboros.Network.Protocol.Limits import Ouroboros.Network.Util.ShowProxy (ShowProxy (..)) import Text.Printf (printf) +-- for runLookaheadFixedSenderPeerWithLimits +import Control.Monad.Class.MonadAsync +import Control.Monad.Class.MonadFork +import Control.Monad.Class.MonadSTM +import Control.Monad.Class.MonadThrow +import Control.Monad.Class.MonadTimer.SI +import Control.Tracer (Tracer (..)) +import Network.Mux.Timeout (withTimeoutSerial) +import Network.TypedProtocol.Driver (runLookaheadFixedSenderPeerWithDriver) +import Ouroboros.Network.Channel +import Ouroboros.Network.Driver.Limits (TraceSendRecv, driverWithLimits) + ----- leiosNotifyMiniProtocolNum :: Mux.MiniProtocolNum @@ -403,6 +426,9 @@ leiosNotifyClientPeer checkDone = type LeiosNotifyServerPeer point announcement vote m a = Peer (LeiosNotify point announcement vote) AsServer NonPipelined StIdle m () +type LeiosNotifyServerPeerLookahead point announcement vote m a = + PeerLookaheadFixedSender (LeiosNotify point announcement vote) AsServer StIdle m () + leiosNotifyServerPeer :: forall m point announcement vote. Monad m => @@ -438,6 +464,10 @@ data C = MkC data WhetherDraining = AlreadyDraining | NotYetDraining +-- | Whether incrementing the count of outstanding LeiosNotify requests found the +-- peer already at its pipelining bound, i.e. requesting more than it is allowed. +data WhetherExcessiveRequests = ExcessiveRequests | NotExcessiveRequests + leiosNotifyClientPeerPipelined :: forall m point announcement vote a. PrimMonad m => @@ -503,3 +533,95 @@ leiosNotifyClientPeerPipelined checkDone k0 = Collect Nothing (\MkC -> drainThePipe x m) + +leiosNotifyServerPeerLookahead :: + forall m point announcement vote. + MonadThrow m => + m WhetherExcessiveRequests -> + m (Message (LeiosNotify point announcement vote) StBusy StIdle) -> + -- ^ blocks until the next reply (announcement\/offer\/vote) is ready + PeerLookaheadFixedSender (LeiosNotify point announcement vote) AsServer StIdle m () +leiosNotifyServerPeerLookahead incr next = + PeerLookaheadFixedSender responder start + where + responder :: Sender (LeiosNotify point announcement vote) AsServer VariableSender StBusy StIdle m + responder = SenderEffect $ next <&> \msg -> SenderYield ReflServerAgency msg SenderDone + + -- StIdle with nothing outstanding: receive the first request (or done). A + -- plain 'Await' is required here; 'AwaitLookahead' defers the StBusy->StIdle + -- send, so it is only usable once we hold a request (i.e. are at StBusy). + start :: + Peer (LeiosNotify point announcement vote) AsServer (Lookahead Z (FixedSender StBusy StIdle)) StIdle m () + start = + Await ReflClientAgency $ \case + MsgDone -> Done ReflNobodyAgency () + MsgLeiosNotificationRequestNext -> + Effect $ do + incr >>= \case + ExcessiveRequests -> throwIO MkExnLeiosNotifyExcessiveRequests + NotExcessiveRequests -> pure () + pure $ busy Zero + + -- StBusy with @n@ deferred sends outstanding: hand this reply off to the + -- responder and look ahead to the next request (or done) in one step. + busy :: forall n. + Nat n -> + Peer (LeiosNotify point announcement vote) AsServer (Lookahead n (FixedSender StBusy StIdle)) StBusy m () + busy n = + AwaitLookahead ReflClientAgency TheSender $ \case + MsgDone -> drain (Succ n) + MsgLeiosNotificationRequestNext -> + Effect $ do + incr >>= \case + ExcessiveRequests -> throwIO MkExnLeiosNotifyExcessiveRequests + NotExcessiveRequests -> pure () + pure $ busy (Succ n) + + -- on termination, flush the sends we've handed off, then Done. + drain :: forall n. + Nat n -> + Peer (LeiosNotify point announcement vote) AsServer (Lookahead n (FixedSender StBusy StIdle)) StDone m () + drain = \case + Zero -> Done ReflNobodyAgency () + Succ j -> FlushSender Nothing (drain j) + +data ExnLeiosNotifyExcessiveRequests = MkExnLeiosNotifyExcessiveRequests + deriving Show + +instance Exception ExnLeiosNotifyExcessiveRequests + +----- + +-- | Run a lookahead (fixed-sender) peer with the given channel via the given codec. +-- +-- The lookahead dual of 'runPipelinedPeerWithLimits': the peer receives ahead +-- and its sends are performed by a parallel thread, hence the 'MonadAsync' +-- constraint. +-- +-- TODO: upstream this to ouroboros-network +runLookaheadFixedSenderPeerWithLimits + :: forall ps (st :: ps) pr failure bytes m a. + ( MonadAsync m + , MonadEvaluate m + , MonadFork m + , MonadMask m + , MonadTimer m + , MonadThrow (STM m) + , ShowProxy ps + , forall (st' :: ps) stok. stok ~ StateToken st' => Show stok + , BearerBytes bytes + , NFData a + , NFData failure + , Show failure + ) + => Tracer m (TraceSendRecv ps) + -> Codec ps failure m bytes + -> ProtocolSizeLimits ps bytes + -> ProtocolTimeLimits ps + -> Channel m bytes + -> PeerLookaheadFixedSender ps pr st m a + -> m (a, Maybe bytes) +runLookaheadFixedSenderPeerWithLimits tracer codec slimits tlimits channel peer = + withTimeoutSerial $ \timeoutFn -> + let driver = driverWithLimits tracer timeoutFn codec slimits tlimits channel + in runLookaheadFixedSenderPeerWithDriver driver peer diff --git a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoTypes.hs b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoTypes.hs index 80dda3fe37..ac825f2d88 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoTypes.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/LeiosDemoTypes.hs @@ -72,6 +72,8 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS import qualified Data.ByteString.Base16 as BS16 import qualified Data.ByteString.Char8 as BS8 +import qualified Data.ByteString.Short as SBS +import LeiosDemoLogic.Announcements.ElBimap (ElId (..)) import Data.Fixed (Pico) import qualified Data.Foldable as F import Data.Function (on) @@ -90,6 +92,7 @@ import qualified Data.Set.NonEmpty as NESet import Data.String (fromString) import Data.Vector.Strict (Vector) import qualified Data.Vector.Strict as V +import Data.Time.Clock (NominalDiffTime) import Data.Word (Word16, Word32, Word64) import Debug.Trace (trace) import GHC.Generics (Generic) @@ -780,17 +783,26 @@ class HasLeiosVoting blk where -- * Tracing messageLeiosNotifyToObject :: + -- | Extracts the announced EB point and body size from the relayed RB + -- header, so an announcement's diffusion can be correlated with its EB. + (announcement -> Maybe (LeiosPoint, BytesSize)) -> Message (LeiosNotify LeiosPoint announcement LeiosVote) st st' -> Aeson.Object -messageLeiosNotifyToObject = \case +messageLeiosNotifyToObject announcedEb = \case MsgLeiosNotificationRequestNext -> mconcat [ "kind" .= Aeson.String "MsgLeiosNotificationRequestNext" ] - MsgLeiosBlockAnnouncement{} -> - mconcat - [ "kind" .= Aeson.String "MsgLeiosBlockAnnouncement" - ] + MsgLeiosBlockAnnouncement announcement -> + mconcat $ + "kind" .= Aeson.String "MsgLeiosBlockAnnouncement" + : case announcedEb announcement of + Nothing -> [] + Just (MkLeiosPoint ebSlot ebHash, ebBodySize) -> + [ "ebSlot" .= ebSlot + , "ebHash" .= prettyEbHash ebHash + , "ebBodySize" .= ebBodySize + ] MsgLeiosBlockOffer (MkLeiosPoint ebSlot ebHash) ebBytesSize -> mconcat [ "kind" .= Aeson.String "MsgLeiosBlockOffer" @@ -885,6 +897,41 @@ data TraceLeiosKernel | TraceLeiosDb TraceLeiosDb | -- | A forged RB both certifies an EB and announce a new one TraceLeiosCertifiedAndAnnounced {atSlot :: SlotNo, rbHash :: RbHash} + | -- | The node accepted a new EB announcement, deduplicated across all peers + -- and its own block forging (see 'AnnouncementSource'). + TraceLeiosAnnouncementAccepted + !AnnouncementSource + !AnnouncementEquivocation + !AnnouncementFields + !(Maybe NominalDiffTime) + -- ^ How late the announcement was — seconds from the election slot's + -- wall-clock onset to when this node counted it — when known (relayed + -- announcements carry it; a locally-forged one does not). + +-- | The data of a relayed EB announcement, shared by 'TraceLeiosPeerAnnouncement' +-- and 'TraceLeiosAnnouncementAccepted'. A separate record so its selectors are +-- total rather than partial over the trace sum types. +data AnnouncementFields = MkAnnouncementFields + { announcementElection :: !ElId + , announcementEbHash :: !EbHash + , announcementEbBodySize :: !BytesSize + } + deriving (Eq, Show) + +-- | Whether the accepted announcement equivocates: a second, distinct header +-- announcing an election that a prior header already announced. (The two +-- headers can even announce the same EB hash and size and still equivocate, +-- since it is the header, not the EB, that carries the election.) The flag +-- spares a log consumer from statefully correlating the two announcements. +data AnnouncementEquivocation + = NoEquivocation + | Equivocation + deriving (Eq, Show) + +-- | Whether the node accepted an EB announcement it forged itself or one +-- relayed by an upstream peer. +data AnnouncementSource = ForgedLocally | ReceivedFromPeer + deriving (Eq, Show) -- | Reasons 'runLeiosVoting' may decline to cast a vote after acquiring an -- EB closure. See 'TraceLeiosNotVoted'. @@ -996,6 +1043,34 @@ traceLeiosKernelToObject = \case , "slotNo" .= slotNo , "rbHash" .= prettyRbHash rbHash ] + TraceLeiosAnnouncementAccepted announcementSource equivocation acc mbAge -> + mconcat $ + [ "kind" .= Aeson.String "LeiosAnnouncementAccepted" + , "source" .= announcementSourceText announcementSource + , announcementFieldsToObject acc + , announcementEquivocationToObject equivocation + ] + ++ foldMap (\age -> ["announcementAgeSeconds" .= (realToFrac age :: Double)]) mbAge + +announcementFieldsToObject :: AnnouncementFields -> Aeson.Object +announcementFieldsToObject + (MkAnnouncementFields (MkElId (SlotNo electionSlot) poolId) ebHash ebBodySize) = + mconcat + [ "electionSlot" .= electionSlot + , "electionPool" .= BS8.unpack (BS16.encode (SBS.fromShort poolId)) + , "ebHash" .= prettyEbHash ebHash + , "ebBodySize" .= ebBodySize + ] + +announcementEquivocationToObject :: AnnouncementEquivocation -> Aeson.Object +announcementEquivocationToObject = \case + NoEquivocation -> "equivocation" .= False + Equivocation -> "equivocation" .= True + +announcementSourceText :: AnnouncementSource -> Aeson.Value +announcementSourceText = \case + ForgedLocally -> Aeson.String "forgedLocally" + ReceivedFromPeer -> Aeson.String "receivedFromPeer" notVotedReasonText :: LeiosNotVotedReason -> Aeson.Value notVotedReasonText = \case @@ -1006,12 +1081,20 @@ notVotedReasonText = \case data TraceLeiosPeer = MkTraceLeiosPeer String | TraceLeiosPeerDbException LeiosDbException + | -- | This upstream peer relayed a valid, newly-counted EB announcement. + TraceLeiosPeerAnnouncement !AnnouncementEquivocation !AnnouncementFields deriving Show traceLeiosPeerToObject :: TraceLeiosPeer -> Aeson.Object traceLeiosPeerToObject = \case MkTraceLeiosPeer s -> fromString "msg" .= Aeson.String (fromString s) TraceLeiosPeerDbException e -> jsonLeiosDbException e + TraceLeiosPeerAnnouncement equivocation acc -> + mconcat + [ fromString "kind" .= Aeson.String "LeiosPeerAnnouncement" + , announcementFieldsToObject acc + , announcementEquivocationToObject equivocation + ] -- * Protocol parameters diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HardFork/Combinator/Protocol.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HardFork/Combinator/Protocol.hs index 5f7127c5d2..e97541c9d5 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HardFork/Combinator/Protocol.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HardFork/Combinator/Protocol.hs @@ -28,6 +28,10 @@ module Ouroboros.Consensus.HardFork.Combinator.Protocol -- * Type family instances , Ticked (..) + + -- * For Leios code, maybe only temporary + , chainDepStateInfo + , injectValidationErr ) where import Control.Monad.Except diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HeaderValidation.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HeaderValidation.hs index ee1b64a1c7..260428e8a4 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HeaderValidation.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/HeaderValidation.hs @@ -21,6 +21,7 @@ module Ouroboros.Consensus.HeaderValidation ( revalidateHeader , validateHeader + , validateHeaderProtocol -- * Annotated tips , AnnTip (..) @@ -514,8 +515,25 @@ validateHeader cfg ledgerView hdr st = do ledgerView (untickedHeaderStateTip st) hdr + withExcept HeaderProtocolError $ + validateHeaderProtocol cfg hdr st + +-- | The protocol-level portion of 'validateHeader' (KES\/VRF\/operational +-- certificate, and any header rules the ledger layers on top), i.e. +-- 'validateHeader' /without/ the envelope check. +-- +-- Leios EB-announcement relay uses this directly: an announced RB header is +-- out-of-order with respect to the local chain, so the envelope check +-- (previous hash and block number) does not apply — but the rest of header +-- validation still must. +validateHeaderProtocol :: + (BlockSupportsProtocol blk, HasAnnTip blk) => + TopLevelConfig blk -> + Header blk -> + Ticked (HeaderState blk) -> + Except (ValidationErr (BlockProtocol blk)) (HeaderState blk) +validateHeaderProtocol cfg hdr st = do chainDepState' <- - withExcept HeaderProtocolError $ updateChainDepState (configConsensus cfg) (validateView (configBlock cfg) hdr) diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Forker.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Forker.hs index 1c24e9c7a0..7556cd1ebc 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Forker.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Forker.hs @@ -64,7 +64,8 @@ module Ouroboros.Consensus.Storage.LedgerDB.Forker ) where import Control.Monad.Except - ( runExcept + ( Except + , runExcept ) import Data.Bifunctor (first) import Data.Functor ((<&>)) @@ -76,6 +77,7 @@ import qualified Data.Set as Set import Data.Word import GHC.Generics import LeiosDemoDb (LeiosDbConnection) +import LeiosDemoLogic.Announcements.ElBimap (ElId) import LeiosDemoTypes ( BytesSize , EbHash @@ -98,7 +100,14 @@ import Ouroboros.Consensus.Ledger.Tables.Utils , prependDiffs , trackingToDiffs ) -import Ouroboros.Consensus.Protocol.Abstract (ChainDepState) +import Ouroboros.Consensus.Protocol.Abstract + ( ChainDepState + , ConsensusConfig + , ConsensusProtocol + , ValidateView + , ValidationErr + , updateChainDepState + ) import Ouroboros.Consensus.Storage.ChainDB.Impl.BlockCache import qualified Ouroboros.Consensus.Storage.ChainDB.Impl.BlockCache as BlockCache import Ouroboros.Consensus.Util.CallStack @@ -786,6 +795,33 @@ class ResolveLeiosBlock blk where headerLeiosAnnouncement :: Header blk -> Maybe (LeiosPoint, BytesSize) headerLeiosAnnouncement _ = Nothing + -- | The election id (slot + block issuer) of this header. + -- + -- TODO no default; every block type should be able to implement + -- this, even though only Dijkstra+ would ever currently call it + headerElId :: Header blk -> ElId + headerElId _ = error "TODO headerElId stub" + + -- | Protocol-level validation of a relayed announcement's header, against the + -- ticked chain-dep state at the announced slot. + -- + -- Mirrors 'updateChainDepState' (its strict behaviour is the default), but is + -- run out-of-context against a possibly-lagging immutable tip, so eras that + -- relay announcements override it to skip the checks that such lag spuriously + -- trips — currently just the OCERT counter's upper bound (see + -- 'WhetherToUpperBoundOCERT'). Unlike a full header validation it also omits + -- the RB-header-specific checks (body size etc.) that a bare EB announcement + -- would not carry. Any error it returns is therefore a genuine rejection. + validateAnnouncementChainDepState :: + ConsensusProtocol (BlockProtocol blk) => + ConsensusConfig (BlockProtocol blk) -> + ValidateView (BlockProtocol blk) -> + SlotNo -> + Ticked (ChainDepState (BlockProtocol blk)) -> + Except (ValidationErr (BlockProtocol blk)) () + validateAnnouncementChainDepState cfg vv slot tcs = + () <$ updateChainDepState cfg vv slot tcs + -- | The EB most recent announcement in the 'HeaderState', if any. 'Nothing' -- for headers in eras that don't carry Leios announcements. protocolStateLeiosAnnouncement :: diff --git a/ouroboros-consensus/test/consensus-test/Main.hs b/ouroboros-consensus/test/consensus-test/Main.hs index 8b4ce5a428..9c23fa3ba5 100644 --- a/ouroboros-consensus/test/consensus-test/Main.hs +++ b/ouroboros-consensus/test/consensus-test/Main.hs @@ -25,6 +25,7 @@ import qualified Test.Consensus.Util.Pred (tests) import qualified Test.Consensus.Util.Versioned (tests) import qualified Test.LeiosDemoDb (tests) import qualified Test.LeiosDemoLogic (tests) +import qualified Test.LeiosDemoLogic.Announcements (tests) import qualified Test.LeiosDemoTypes (tests) import qualified Test.LeiosUtils.CallTrace (tests) import qualified Test.LeiosVoteState (tests) @@ -82,6 +83,7 @@ tests = [ Test.LeiosDemoTypes.tests , Test.LeiosDemoDb.tests , Test.LeiosDemoLogic.tests + , Test.LeiosDemoLogic.Announcements.tests , Test.LeiosVoteState.tests , Test.LeiosUtils.CallTrace.tests ] diff --git a/ouroboros-consensus/test/consensus-test/Test/LeiosDemoLogic/Announcements.hs b/ouroboros-consensus/test/consensus-test/Test/LeiosDemoLogic/Announcements.hs new file mode 100644 index 0000000000..ab35317ccf --- /dev/null +++ b/ouroboros-consensus/test/consensus-test/Test/LeiosDemoLogic/Announcements.hs @@ -0,0 +1,395 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Tests for the block-agnostic announcement logic in +-- 'LeiosDemoLogic.Announcements' and 'LeiosDemoLogic.Announcements.ElBimap': +-- per-peer equivocation counting ('extendLive'\/'onAnnouncement'), the +-- node-wide dedup + relay ('onAnnouncementCentral'), pruning, and the +-- bidirectional election map. +-- +-- Everything here is exercised through a trivial mock announcement type +-- ('Anc'), so the tests depend on none of the block\/ledger machinery — which +-- is the point of keeping that logic polymorphic. +-- +-- The third announcement module, 'LeiosDemoLogic.Announcements.Validate', is +-- deliberately out of scope for this specific test suite: +-- 'validateAnnouncementHeader' needs a concrete block (@LedgerSupportsProtocol@ +-- + @ResolveLeiosBlock@, plus forecasting and header-protocol validation), so +-- it is exercised by the proto-devnet rather than here. +module Test.LeiosDemoLogic.Announcements (tests) where + +import Cardano.Slotting.Slot (SlotNo (..)) +import Control.Concurrent.Class.MonadSTM.Strict.TVar + ( StrictTVar + , newTVarIO + , readTVarIO + ) +import Control.Monad.Except (runExceptT) +import Control.Tracer (nullTracer) +import qualified Data.ByteString.Short as SBS +import Data.IORef +import Data.List (nub) +import qualified Data.Map.Strict as Map +import qualified Data.Set as Set +import Data.Set.NonEmpty (NESet) +import qualified Data.Set.NonEmpty as NESet +import Data.Void (Void) +import Data.Word (Word64, Word8) +import LeiosDemoLogic.Announcements +import LeiosDemoLogic.Announcements.ElBimap +import Test.Tasty (TestTree, adjustOption, testGroup) +import Test.Tasty.HUnit (Assertion, assertFailure, testCase, (@?=)) +import Test.Tasty.QuickCheck + ( Gen + , Property + , QuickCheckTests (..) + , chooseInt + , forAll + , listOf + , oneof + , testProperty + , (===) + ) + +tests :: TestTree +tests = + -- These properties are cheap; run 1000x the usual repetitions. + adjustOption (\(QuickCheckTests n) -> QuickCheckTests (n * 1000)) $ + testGroup + "Announcements" + [ extendLiveTests + , onAnnouncementTests + , centralTests + , pruneTests + , elBimapTests + ] + +{------------------------------------------------------------------------------- + Mock announcement type +-------------------------------------------------------------------------------} + +-- | A mock announcement: its election plus a tag distinguishing announcements +-- for the same election. Two 'Anc's with the same election but different tags +-- are an equivocation; identical 'Anc's are a repeat. +data Anc = Anc !ElId !Int + deriving (Eq, Show) + +ancEl :: Anc -> ElId +ancEl (Anc e _) = e + +ancTag :: Anc -> Int +ancTag (Anc _ i) = i + +-- | An election at the given slot for the given (one-byte) pool id. +mkEl :: Word64 -> Word8 -> ElId +mkEl slot poolId = MkElId (SlotNo slot) (SBS.pack [poolId]) + +elSlot :: ElId -> Word64 +elSlot (MkElId (SlotNo s) _) = s + +{------------------------------------------------------------------------------- + extendLive +-------------------------------------------------------------------------------} + +-- | A comparable summary of an 'extendLive' outcome. 'St' lists the tags of the +-- election's stored announcements (in order). +data Ext = ErrR | ErrT | St [Int] + deriving (Eq, Show) + +step :: PeerState Anc -> Anc -> Either (ErrAnnouncement Void) (ElState Anc, PeerState Anc) +step st a = extendLive (ancEl a) a st + +classifyExt :: Either (ErrAnnouncement Void) (ElState Anc, PeerState Anc) -> Ext +classifyExt = \case + Left ErrRepeat -> ErrR + Left ErrThird -> ErrT + Left ErrTooOld -> error "unreachable: extendLive never yields ErrTooOld" + Right (elSt, _) -> St (elStateTags elSt) + +elStateTags :: ElState Anc -> [Int] +elStateTags elSt = + ancTag (firstAnnouncement elSt) : maybe [] (pure . ancTag) (secondAnnouncement elSt) + +-- | Fold announcements into a 'PeerState', keeping successful extensions and +-- silently dropping rejected ones. +buildState :: [Anc] -> PeerState Anc +buildState = foldl (\st a -> either (const st) snd (step st a)) emptyPeerState + +el0 :: ElId +el0 = mkEl 10 0 + +extendLiveTests :: TestTree +extendLiveTests = + testGroup + "extendLive" + [ testCase "first announcement is accepted" $ + classifyExt (step emptyPeerState (Anc el0 0)) @?= St [0] + , testCase "identical repeat is ErrRepeat" $ + classifyExt (step (buildState [Anc el0 0]) (Anc el0 0)) @?= ErrR + , testCase "second distinct announcement equivocates (kept as two)" $ + classifyExt (step (buildState [Anc el0 0]) (Anc el0 1)) @?= St [0, 1] + , testCase "third announcement is ErrThird" $ + classifyExt (step (buildState [Anc el0 0, Anc el0 1]) (Anc el0 2)) @?= ErrT + , testCase "repeat of one of two is still ErrThird" $ + classifyExt (step (buildState [Anc el0 0, Anc el0 1]) (Anc el0 0)) @?= ErrT + , testCase "distinct elections are independent" $ + Set.fromList (map elSlot (Map.keys (live (buildState [Anc (mkEl 10 0) 0, Anc (mkEl 20 0) 0])))) + @?= Set.fromList [10, 20] + , testProperty "stored announcements are the first <=2 distinct tags, in order" $ + prop_extendLiveModel + ] + +-- | Folding a random announcement stream leaves, for each election, exactly the +-- first at-most-two distinct tags in order of first appearance. +prop_extendLiveModel :: Property +prop_extendLiveModel = forAll genStream $ \ops -> + let actual = Map.map elStateTags (live (buildState [Anc e t | (e, t) <- ops])) + model = Map.map (take 2 . nub) (Map.fromListWith (\new old -> old ++ new) [(e, [t]) | (e, t) <- ops]) + in actual === model + +genElId :: Gen ElId +genElId = mkEl <$> (fromIntegral <$> chooseInt (1, 3)) <*> (fromIntegral <$> chooseInt (0, 1)) + +genStream :: Gen [(ElId, Int)] +genStream = listOf ((,) <$> genElId <*> chooseInt (0, 3)) + +{------------------------------------------------------------------------------- + onAnnouncement (per-peer wrapper: count, then validate, then process) +-------------------------------------------------------------------------------} + +onAnnouncementTests :: TestTree +onAnnouncementTests = + testGroup + "onAnnouncement" + [ testCase "invalid announcement surfaces as ErrInvalid" test_oaInvalid + , testCase "too-old announcement (Left Nothing) surfaces as ErrTooOld" test_oaTooOld + , testCase "valid announcement runs process and returns new state" test_oaProcess + , testCase "repeat is rejected without running process" test_oaRepeat + ] + +test_oaInvalid :: Assertion +test_oaInvalid = do + res <- + runExceptT $ + onAnnouncement nullTracer ancEl (\_ -> pure (Left (Just "bad")) :: IO (Either (Maybe String) ())) (\_ _ -> pure ()) emptyPeerState (Anc el0 0) + case res of + Left (ErrInvalid inv) -> inv @?= "bad" + _ -> assertFailure "expected ErrInvalid" + +test_oaTooOld :: Assertion +test_oaTooOld = do + res <- + runExceptT $ + onAnnouncement nullTracer ancEl (\_ -> pure (Left Nothing) :: IO (Either (Maybe String) ())) (\_ _ -> pure ()) emptyPeerState (Anc el0 0) + case res of + Left ErrTooOld -> pure () + _ -> assertFailure "expected ErrTooOld" + +test_oaProcess :: Assertion +test_oaProcess = do + ref <- newIORef [] + res <- + runExceptT $ + onAnnouncement nullTracer ancEl (\_ -> pure (Right ())) (\a () -> modifyIORef' ref (a :)) emptyPeerState (Anc el0 0) + readIORef ref >>= (@?= [Anc el0 0]) + case res of + Right _ -> pure () + Left _ -> assertFailure "expected success" + +test_oaRepeat :: Assertion +test_oaRepeat = do + ref <- newIORef (0 :: Int) + let validate = \_ -> pure (Right ()) + process = \_ _ -> modifyIORef' ref (+ 1) :: IO () + go st = runExceptT $ onAnnouncement nullTracer ancEl validate process st (Anc el0 0) + Right st1 <- go emptyPeerState + res2 <- go st1 + case res2 of + Left ErrRepeat -> pure () + _ -> assertFailure "expected ErrRepeat" + readIORef ref >>= (@?= 1) -- process ran once (the first, not the repeat) + +{------------------------------------------------------------------------------- + onAnnouncementCentral (node-wide dedup + relay) +-------------------------------------------------------------------------------} + +-- | A downstream peer's queue: a credit counter and a FIFO of relayed +-- announcements, plus readers for both. +data TestQueue = TestQueue + { tqView :: QueueAnnouncementView IO Anc + , tqRead :: IO [Anc] + , tqCredits :: IO Int + } + +newQueue :: Int -> IO TestQueue +newQueue credits = do + free <- newTVarIO credits :: IO (StrictTVar IO Int) + q <- newTVarIO [] :: IO (StrictTVar IO [Anc]) + pure + TestQueue + { tqView = MkQueueAnnouncementView free (\xs a -> xs ++ [a]) q + , tqRead = readTVarIO q + , tqCredits = readTVarIO free + } + +-- | Central-state helper: relay from a given source with a given 'ShouldRelay'. +central :: + Maybe Int -> + ShouldRelay -> + Anc -> + CentralState IO Int Anc -> + IO (CentralState IO Int Anc) +central src rel anc st = + onAnnouncementCentral nullTracer ancEl (\_ -> pure ()) st src rel Nothing anc + +centralTests :: TestTree +centralTests = + testGroup + "onAnnouncementCentral" + [ testCase "new announcement is relayed, credit spent, peer gated" test_relay + , testCase "duplicate announcement is a no-op" test_dedup + , testCase "DoNotRelay skips the queue and subsequent DoRelay can't retcon that" test_noRelay + , testCase "no relay when the peer has no credits" test_noCredit + , testCase "equivocation goes only to peers already sent the first" test_equivocation + , testCase "publishLocally fires once per new announcement" test_publish + ] + +test_relay :: Assertion +test_relay = do + tq <- newQueue 5 + let st0 = insertPeerCentral 1 (tqView tq) emptyCentralState + st1 <- central (Just 2) DoRelay (Anc el0 0) st0 + tqRead tq >>= (@?= [Anc el0 0]) + tqCredits tq >>= (@?= 4) + Set.toList (lookupElBimapL el0 (gate st1)) @?= [1] + +test_dedup :: Assertion +test_dedup = do + tq <- newQueue 5 + let st0 = insertPeerCentral 1 (tqView tq) emptyCentralState + st1 <- central (Just 2) DoRelay (Anc el0 0) st0 + _ <- central (Just 2) DoRelay (Anc el0 0) st1 -- same announcement again + tqRead tq >>= (@?= [Anc el0 0]) -- not relayed twice + tqCredits tq >>= (@?= 4) -- credit spent only once + +test_noRelay :: Assertion +test_noRelay = do + tq <- newQueue 5 + let st0 = insertPeerCentral 1 (tqView tq) emptyCentralState + st1 <- central Nothing DoNotRelay (Anc el0 0) st0 -- e.g. self-forged, too old to relay + tqRead tq >>= (@?= []) -- nothing sent + _st2 <- central (Just 2) DoRelay (Anc el0 0) st1 -- same announcement, now DoRelay + tqRead tq >>= (@?= []) -- still nothing: DoNotRelay had recorded it in selfPeer + +test_noCredit :: Assertion +test_noCredit = do + tq <- newQueue 0 + let st0 = insertPeerCentral 1 (tqView tq) emptyCentralState + st1 <- central (Just 2) DoRelay (Anc el0 0) st0 + tqRead tq >>= (@?= []) -- dropped for lack of credits + Set.toList (lookupElBimapL el0 (gate st1)) @?= [] -- and not gated + +test_equivocation :: Assertion +test_equivocation = do + tqA <- newQueue 5 + tqB <- newQueue 5 + let stA = insertPeerCentral 1 (tqView tqA) emptyCentralState + st1 <- central (Just 9) DoRelay (Anc el0 0) stA -- peer 1 gets the first + let stAB = insertPeerCentral 2 (tqView tqB) st1 -- peer 2 joins afterwards + _ <- central (Just 9) DoRelay (Anc el0 1) stAB -- the equivocating second + tqRead tqA >>= (@?= [Anc el0 0, Anc el0 1]) -- peer 1: first + equivocation proof + tqRead tqB >>= (@?= []) -- peer 2: never saw the first, so not the proof + +test_publish :: Assertion +test_publish = do + ref <- newIORef (0 :: Int) + tq <- newQueue 5 + let st0 = insertPeerCentral 'A' (tqView tq) emptyCentralState + pub = \_ -> modifyIORef' ref (+ 1) + st1 <- onAnnouncementCentral nullTracer ancEl pub st0 (Just 'B') DoRelay Nothing (Anc el0 0) + _ <- onAnnouncementCentral nullTracer ancEl pub st1 (Just 'B') DoRelay Nothing (Anc el0 0) -- duplicate + readIORef ref >>= (@?= 1) + +{------------------------------------------------------------------------------- + pruning +-------------------------------------------------------------------------------} + +pruneTests :: TestTree +pruneTests = + testGroup + "prunePeerState" + [ testProperty "keeps exactly the elections at or above the immutable tip" prop_prune + ] + +-- | 'prunePeerState' keeps precisely the elections whose slot is at or above the +-- immutable tip, leaving each such election's announcements untouched. +prop_prune :: Property +prop_prune = forAll genStream $ \ops -> forAll genSlot $ \s -> + let st = buildState [Anc e t | (e, t) <- ops] + tagsOf ps = Map.map elStateTags (live ps) + in tagsOf (prunePeerState (SlotNo s) st) + === Map.filterWithKey (\e _ -> elSlot e >= s) (tagsOf st) + +{------------------------------------------------------------------------------- + ElBimap +-------------------------------------------------------------------------------} + +elBimapTests :: TestTree +elBimapTests = + testGroup + "ElBimap" + [ testProperty "forward half is a plain Map ElId (NESet a)" prop_bimapForwardModel + , testProperty "inverse half is a plain Map a (NESet ElId)" prop_bimapInverseModel + ] + +-- | An immutable-tip slot spanning the range of generated election slots (so +-- prune boundaries are exercised). Used by 'prop_prune'. +genSlot :: Gen Word64 +genSlot = fromIntegral <$> chooseInt (0, 4) + +-- | The forward half's semantics, implemented directly on a plain 'Map'. +applyFwd :: Map.Map ElId (NESet Int) -> BOp -> Map.Map ElId (NESet Int) +applyFwd m = \case + Ins l r -> Map.insertWith NESet.union l (NESet.singleton r) m + DelL l -> Map.delete l m + DelR r -> Map.mapMaybe (NESet.nonEmptySet . NESet.delete r) m + Prune s -> Map.filterWithKey (\l _ -> elSlot l >= s) m + +-- | The inverse half's semantics, implemented directly on a plain 'Map'. +applyInv :: Map.Map Int (NESet ElId) -> BOp -> Map.Map Int (NESet ElId) +applyInv m = \case + Ins l r -> Map.insertWith NESet.union r (NESet.singleton l) m + DelL l -> Map.mapMaybe (NESet.nonEmptySet . NESet.delete l) m + DelR r -> Map.delete r m + Prune s -> Map.mapMaybe (NESet.nonEmptySet . NESet.filter (\l -> elSlot l >= s)) m + +-- | The forward half is indistinguishable from a plain @Map ElId (NESet a)@ +-- maintained directly, under any mix of inserts, deletes, and prunes. +prop_bimapForwardModel :: Property +prop_bimapForwardModel = forAll (listOf genBOp) $ \ops -> + forwardHalf (foldl applyBOp emptyElBimap ops) === foldl applyFwd Map.empty ops + +-- | The inverse half is indistinguishable from a plain @Map a (NESet ElId)@ +-- maintained directly, under any mix of inserts, deletes, and prunes. +prop_bimapInverseModel :: Property +prop_bimapInverseModel = forAll (listOf genBOp) $ \ops -> + inverseHalf (foldl applyBOp emptyElBimap ops) === foldl applyInv Map.empty ops + +data BOp = Ins ElId Int | DelL ElId | DelR Int | Prune Word64 + deriving Show + +applyBOp :: ElBimap Int -> BOp -> ElBimap Int +applyBOp bm = \case + Ins l r -> insertElBimap l r bm + DelL l -> deleteElBimapL l bm + DelR r -> deleteElBimapR r bm + Prune s -> pruneElBimap (SlotNo s) bm + +genBOp :: Gen BOp +genBOp = + oneof + [ Ins <$> genElId <*> chooseInt (0, 3) + , DelL <$> genElId + , DelR <$> chooseInt (0, 3) + , Prune . fromIntegral <$> chooseInt (0, 4) + ]