-- | GossipSub mesh management: JOIN, LEAVE, GRAFT/PRUNE, message forwarding.
--
-- The router manages the mesh overlay and handles inbound/outbound
-- RPC messages. For testability, peer communication is injectable
-- via gsSendRPC on GossipSubRouter.
module LibP2P.Protocol.GossipSub.Router
  ( -- * Construction
    newRouter
    -- * Peer management
  , addPeer
  , removePeer
  , setPeerIP
  , setSignedPeerRecord
    -- * Topic subscription
  , join
  , leave
    -- * Publishing
  , publish
    -- * Topic validation
  , registerValidator
  , unregisterValidator
    -- * Inbound RPC handling
  , handleRPC
    -- * Control message handlers
  , handleGraft
  , handlePrune
  , handleIHave
  , handleIWant
  , handleSubscriptions
    -- * Message forwarding
  , forwardMessage
    -- * Scoring
  , peerScore
    -- * Peer exchange
  , selectPXPeers
    -- * Version-gated PRUNE construction
  , buildPrune
  ) where

import Prelude
import Control.Exception (throwIO)
import Control.Monad (unless, when)
import Control.Concurrent.STM
import Data.ByteString (ByteString)
import qualified Data.Map.Strict as Map
import qualified Data.Set as Set
import Data.Time (UTCTime, addUTCTime, diffUTCTime)
import Data.Word (Word64)
import Crypto.Random (getRandomBytes)
import List.Shuffle (sampleIO)
import LibP2P.Crypto.PeerId (PeerId, peerIdBytes)
import LibP2P.Crypto.PeerRecord (PeerRecord (..), openPeerRecordEnvelope)
import LibP2P.Crypto.Key (KeyPair (..), sign)
import LibP2P.Crypto.Protobuf (encodePublicKey)
import LibP2P.Protocol.GossipSub.Types
import LibP2P.Protocol.GossipSub.MessageCache (newMessageCache, cachePut, cacheGet)
import LibP2P.Protocol.GossipSub.Score
  ( computeScore
  , addP7Penalty
  , recordMeshFailure
  , recordInvalidMessage
  , recordFirstDelivery
  , recordMeshDelivery
  , markPeerInMesh
  , unmarkPeerInMesh
  )
import LibP2P.Protocol.GossipSub.Validation (validateMessage, signingBytes)

-- | Create a new GossipSub router with empty state.
newRouter :: GossipSubParams
          -> PeerId
          -> (PeerId -> RPC -> IO ())   -- ^ RPC sender
          -> IO UTCTime                 -- ^ Time source
          -> IO GossipSubRouter
newRouter :: GossipSubParams
-> PeerId
-> (PeerId -> RPC -> IO ())
-> IO UTCTime
-> IO GossipSubRouter
newRouter GossipSubParams
params PeerId
localPid PeerId -> RPC -> IO ()
sendRPC IO UTCTime
getTime = do
  subs     <- Set Topic -> IO (TVar (Set Topic))
forall a. a -> IO (TVar a)
newTVarIO Set Topic
forall a. Set a
Set.empty
  mesh     <- newTVarIO Map.empty
  fanout   <- newTVarIO Map.empty
  fanoutPub <- newTVarIO Map.empty
  peers    <- newTVarIO Map.empty
  seen     <- newTVarIO Map.empty
  backoff  <- newTVarIO Map.empty
  ipCount  <- newTVarIO Map.empty
  mcache   <- newTVarIO (newMessageCache (paramMcacheLen params) (paramMcacheGossip params))
  hbCount  <- newTVarIO 0
  onMsg    <- newTVarIO (\Topic
_ PubSubMessage
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  validators <- newTVarIO Map.empty
  promises <- newTVarIO Map.empty
  ihaveCounts <- newTVarIO Map.empty
  iaskedCounts <- newTVarIO Map.empty
  iwantServed <- newTVarIO Map.empty
  onPX     <- newTVarIO (\Topic
_ [PeerExchangeInfo]
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  signedRecords <- newTVarIO Map.empty
  pure GossipSubRouter
    { gsParams         = params
    , gsLocalPeerId    = localPid
    , gsSubscriptions  = subs
    , gsMesh           = mesh
    , gsFanout         = fanout
    , gsFanoutPub      = fanoutPub
    , gsPeers          = peers
    , gsSeen           = seen
    , gsBackoff        = backoff
    , gsScoreParams    = defaultPeerScoreParams
    , gsThresholds     = defaultScoreThresholds
    , gsIPPeerCount    = ipCount
    , gsIWantPromises  = promises
    , gsIHaveCounts    = ihaveCounts
    , gsIAskedCounts   = iaskedCounts
    , gsIWantServed    = iwantServed
    , gsMessageCache   = mcache
    , gsHeartbeatCount = hbCount
    , gsSendRPC        = sendRPC
    , gsGetTime        = getTime
    , gsOnMessage      = onMsg
    , gsValidators     = validators
    , gsOnPeerExchange = onPX
    , gsSignedPeerRecords = signedRecords
    }

-- Topic validation

-- | Attach an application validator to a topic. Messages the validator
-- rejects are dropped without propagation and count against the sender's
-- P4 score; ignored messages are dropped without any penalty
-- (gossipsub-v1.1.md extended validators).
registerValidator :: GossipSubRouter -> Topic -> TopicValidator -> IO ()
registerValidator :: GossipSubRouter -> Topic -> TopicValidator -> IO ()
registerValidator GossipSubRouter
router Topic
topic TopicValidator
v = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map Topic TopicValidator)
-> (Map Topic TopicValidator -> Map Topic TopicValidator) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map Topic TopicValidator)
gsValidators GossipSubRouter
router) (Topic
-> TopicValidator
-> Map Topic TopicValidator
-> Map Topic TopicValidator
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Topic
topic TopicValidator
v)

-- | Remove a topic's validator.
unregisterValidator :: GossipSubRouter -> Topic -> IO ()
unregisterValidator :: GossipSubRouter -> Topic -> IO ()
unregisterValidator GossipSubRouter
router Topic
topic = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map Topic TopicValidator)
-> (Map Topic TopicValidator -> Map Topic TopicValidator) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map Topic TopicValidator)
gsValidators GossipSubRouter
router) (Topic -> Map Topic TopicValidator -> Map Topic TopicValidator
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete Topic
topic)

-- Peer management

-- | Register a connected peer. If the peer already exists, preserves
-- accumulated state (topics, scores) to avoid overwriting subscriptions.
addPeer :: GossipSubRouter -> PeerId -> PeerProtocol -> Bool -> UTCTime -> IO ()
addPeer :: GossipSubRouter
-> PeerId -> PeerProtocol -> Bool -> UTCTime -> IO ()
addPeer GossipSubRouter
router PeerId
pid PeerProtocol
proto Bool
isOutbound UTCTime
now = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$ \Map PeerId PeerState
m ->
    case PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId PeerState
m of
      Just PeerState
_existing -> Map PeerId PeerState
m  -- Peer already registered, keep existing state
      Maybe PeerState
Nothing -> PeerId -> PeerState -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
pid PeerState
        { psProtocol :: PeerProtocol
psProtocol        = PeerProtocol
proto
        , psTopics :: Set Topic
psTopics          = Set Topic
forall a. Set a
Set.empty
        , psIsOutbound :: Bool
psIsOutbound      = Bool
isOutbound
        , psConnectedAt :: UTCTime
psConnectedAt     = UTCTime
now
        , psTopicState :: Map Topic TopicPeerState
psTopicState      = Map Topic TopicPeerState
forall k a. Map k a
Map.empty
        , psBehaviorPenalty :: Double
psBehaviorPenalty = Double
0
        , psIPAddress :: Maybe ByteString
psIPAddress       = Maybe ByteString
forall a. Maybe a
Nothing
        , psCachedScore :: Double
psCachedScore     = Double
0
        } Map PeerId PeerState
m

-- | Remove a disconnected peer and clean up mesh/fanout membership,
-- IP colocation tracking (P6) and outstanding IWANT promises.
removePeer :: GossipSubRouter -> PeerId -> IO ()
removePeer :: GossipSubRouter -> PeerId -> IO ()
removePeer GossipSubRouter
router PeerId
pid = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  peers <- TVar (Map PeerId PeerState) -> STM (Map PeerId PeerState)
forall a. TVar a -> STM a
readTVar (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  case Map.lookup pid peers >>= psIPAddress of
    Just ByteString
ip -> TVar (Map ByteString (Set PeerId))
-> (Map ByteString (Set PeerId) -> Map ByteString (Set PeerId))
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map ByteString (Set PeerId))
gsIPPeerCount GossipSubRouter
router) (ByteString
-> PeerId
-> Map ByteString (Set PeerId)
-> Map ByteString (Set PeerId)
removeIPMember ByteString
ip PeerId
pid)
    Maybe ByteString
Nothing -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  modifyTVar' (gsPeers router) (Map.delete pid)
  modifyTVar' (gsMesh router) (Map.map (Set.delete pid))
  modifyTVar' (gsSignedPeerRecords router) (Map.delete pid)
  modifyTVar' (gsFanout router) (Map.map (Set.delete pid))
  modifyTVar' (gsIWantPromises router) $
    Map.filterWithKey (\(PeerId
p, ByteString
_) UTCTime
_ -> PeerId
p PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerId
pid)

-- | Record a peer's IP address for P6 (IP colocation) scoring.
-- No-op for unknown peers; replaces any previously recorded address.
setPeerIP :: GossipSubRouter -> PeerId -> ByteString -> IO ()
setPeerIP :: GossipSubRouter -> PeerId -> ByteString -> IO ()
setPeerIP GossipSubRouter
router PeerId
pid ByteString
ip = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  peers <- TVar (Map PeerId PeerState) -> STM (Map PeerId PeerState)
forall a. TVar a -> STM a
readTVar (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  case Map.lookup pid peers of
    Maybe PeerState
Nothing -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    Just PeerState
ps -> do
      case PeerState -> Maybe ByteString
psIPAddress PeerState
ps of
        Just ByteString
oldIp | ByteString
oldIp ByteString -> ByteString -> Bool
forall a. Eq a => a -> a -> Bool
/= ByteString
ip ->
          TVar (Map ByteString (Set PeerId))
-> (Map ByteString (Set PeerId) -> Map ByteString (Set PeerId))
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map ByteString (Set PeerId))
gsIPPeerCount GossipSubRouter
router) (ByteString
-> PeerId
-> Map ByteString (Set PeerId)
-> Map ByteString (Set PeerId)
removeIPMember ByteString
oldIp PeerId
pid)
        Maybe ByteString
_ -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$
        PeerId -> PeerState -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
pid PeerState
ps { psIPAddress = Just ip }
      TVar (Map ByteString (Set PeerId))
-> (Map ByteString (Set PeerId) -> Map ByteString (Set PeerId))
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map ByteString (Set PeerId))
gsIPPeerCount GossipSubRouter
router) ((Map ByteString (Set PeerId) -> Map ByteString (Set PeerId))
 -> STM ())
-> (Map ByteString (Set PeerId) -> Map ByteString (Set PeerId))
-> STM ()
forall a b. (a -> b) -> a -> b
$
        (Set PeerId -> Set PeerId -> Set PeerId)
-> ByteString
-> Set PeerId
-> Map ByteString (Set PeerId)
-> Map ByteString (Set PeerId)
forall k a. Ord k => (a -> a -> a) -> k -> a -> Map k a -> Map k a
Map.insertWith Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.union ByteString
ip (PeerId -> Set PeerId
forall a. a -> Set a
Set.singleton PeerId
pid)

-- | Drop a peer from an IP's membership set, deleting empty sets.
removeIPMember :: ByteString -> PeerId
               -> Map.Map ByteString (Set.Set PeerId)
               -> Map.Map ByteString (Set.Set PeerId)
removeIPMember :: ByteString
-> PeerId
-> Map ByteString (Set PeerId)
-> Map ByteString (Set PeerId)
removeIPMember ByteString
ip PeerId
pid = (Set PeerId -> Maybe (Set PeerId))
-> ByteString
-> Map ByteString (Set PeerId)
-> Map ByteString (Set PeerId)
forall k a. Ord k => (a -> Maybe a) -> k -> Map k a -> Map k a
Map.update
  (\Set PeerId
s -> let s' :: Set PeerId
s' = PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => a -> Set a -> Set a
Set.delete PeerId
pid Set PeerId
s
         in if Set PeerId -> Bool
forall a. Set a -> Bool
Set.null Set PeerId
s' then Maybe (Set PeerId)
forall a. Maybe a
Nothing else Set PeerId -> Maybe (Set PeerId)
forall a. a -> Maybe a
Just Set PeerId
s') ByteString
ip

-- Direct peers (gossipsub-v1.1.md explicit peering agreements)

-- | True when the peer is a configured direct peer.
isDirectPeer :: GossipSubRouter -> PeerId -> Bool
isDirectPeer :: GossipSubRouter -> PeerId -> Bool
isDirectPeer GossipSubRouter
router PeerId
pid = PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid (GossipSubParams -> Set PeerId
paramDirectPeers (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router))

-- | Known direct peers subscribed to the topic. They always receive
-- published and forwarded messages although they are never in the mesh.
directTopicPeers :: GossipSubRouter -> Map.Map PeerId PeerState -> Topic
                 -> Set.Set PeerId
directTopicPeers :: GossipSubRouter -> Map PeerId PeerState -> Topic -> Set PeerId
directTopicPeers GossipSubRouter
router Map PeerId PeerState
peers Topic
topic =
  (PeerId -> Bool) -> Set PeerId -> Set PeerId
forall a. (a -> Bool) -> Set a -> Set a
Set.filter (\PeerId
pid -> case PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId PeerState
peers of
    Just PeerState
ps -> Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
    Maybe PeerState
Nothing -> Bool
False) (GossipSubParams -> Set PeerId
paramDirectPeers (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router))

-- FloodSub peers (gossipsub-v1.0.md "Compatibility with FloodSub")

-- | True when the peer negotiated /floodsub/1.0.0. Floodsub peers have
-- no mesh and understand no gossipsub control messages: they are flooded
-- every message for topics they subscribe to and must never be grafted,
-- pruned or gossiped to.
isFloodSubPeer :: Map.Map PeerId PeerState -> PeerId -> Bool
isFloodSubPeer :: Map PeerId PeerState -> PeerId -> Bool
isFloodSubPeer Map PeerId PeerState
peers PeerId
pid = case PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId PeerState
peers of
  Just PeerState
ps -> PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
== PeerProtocol
FloodSubPeer
  Maybe PeerState
Nothing -> Bool
False

-- | Known floodsub peers subscribed to the topic. They receive every
-- published and forwarded message for it (flooding).
floodSubTopicPeers :: Map.Map PeerId PeerState -> Topic -> Set.Set PeerId
floodSubTopicPeers :: Map PeerId PeerState -> Topic -> Set PeerId
floodSubTopicPeers Map PeerId PeerState
peers Topic
topic =
  (Set PeerId -> PeerId -> PeerState -> Set PeerId)
-> Set PeerId -> Map PeerId PeerState -> Set PeerId
forall a k b. (a -> k -> b -> a) -> a -> Map k b -> a
Map.foldlWithKey' (\Set PeerId
acc PeerId
pid PeerState
ps ->
    if PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
== PeerProtocol
FloodSubPeer Bool -> Bool -> Bool
&& Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
    then PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => a -> Set a -> Set a
Set.insert PeerId
pid Set PeerId
acc
    else Set PeerId
acc) Set PeerId
forall a. Set a
Set.empty Map PeerId PeerState
peers

-- Protocol version gating

-- | True when the peer negotiated /meshsub/1.1.0. Unknown peers are
-- treated as v1.1 (the preferred protocol).
peerSupportsV11 :: Map.Map PeerId PeerState -> PeerId -> Bool
peerSupportsV11 :: Map PeerId PeerState -> PeerId -> Bool
peerSupportsV11 Map PeerId PeerState
peers PeerId
pid = case PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId PeerState
peers of
  Just PeerState
ps -> PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
== PeerProtocol
GossipSubPeer
  Maybe PeerState
Nothing -> Bool
True

-- | Build a PRUNE for a peer, gated on its negotiated protocol version:
-- /meshsub/1.0.0 peers receive a bare PRUNE — PX records and the backoff
-- field are v1.1 control extensions (#157).
buildPrune :: GossipSubRouter -> PeerId -> Topic
           -> Bool     -- ^ Include PX records (only for v1.1 peers in good standing)
           -> Word64   -- ^ Backoff seconds (only encoded for v1.1 peers)
           -> IO Prune
buildPrune :: GossipSubRouter -> PeerId -> Topic -> Bool -> Word64 -> IO Prune
buildPrune GossipSubRouter
router PeerId
pid Topic
topic Bool
withPX Word64
backoffSecs = do
  peers <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  if peerSupportsV11 peers pid
    then do
      px <- if withPX then selectPXPeers router topic pid else pure []
      pure (Prune topic px (Just backoffSecs))
    else pure (Prune topic [] Nothing)

-- Topic subscription

-- | Subscribe to a topic (JOIN): announce, fanout→mesh transition, fill to D, GRAFT.
join :: GossipSubRouter -> Topic -> IO ()
join :: GossipSubRouter -> Topic -> IO ()
join GossipSubRouter
router Topic
topic = do
  -- 1. Record the subscription. This must happen regardless of how many
  -- peers currently know the topic: the subscription set (not mesh key
  -- presence) is what GRAFT acceptance and hello-packet announcements
  -- consult (gossipsub-v1.0.md JOIN/GRAFT; issue #155).
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Set Topic) -> (Set Topic -> Set Topic) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Set Topic)
gsSubscriptions GossipSubRouter
router) (Topic -> Set Topic -> Set Topic
forall a. Ord a => a -> Set a -> Set a
Set.insert Topic
topic)

  -- 2. Announce subscription to all known peers
  peers <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  let allPeerIds = Map PeerId PeerState -> [PeerId]
forall k a. Map k a -> [k]
Map.keys Map PeerId PeerState
peers
      subRPC = RPC
emptyRPC { rpcSubscriptions = [SubOpts True topic] }
  mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
subRPC) allPeerIds

  -- 3. Check fanout and transition to mesh
  (fanoutPeers, topicPeers) <- atomically $ do
    fo <- readTVar (gsFanout router)
    let foPeers = Set PeerId -> Topic -> Map Topic (Set PeerId) -> Set PeerId
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Set PeerId
forall a. Set a
Set.empty Topic
topic Map Topic (Set PeerId)
fo
    -- Move fanout peers to mesh
    unless (Set.null foPeers) $ do
      modifyTVar' (gsMesh router) (Map.insert topic foPeers)
      modifyTVar' (gsFanout router) (Map.delete topic)
      modifyTVar' (gsFanoutPub router) (Map.delete topic)
    -- Get current mesh and all eligible peers
    meshNow <- readTVar (gsMesh router)
    let currentMesh = Set PeerId -> Topic -> Map Topic (Set PeerId) -> Set PeerId
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Set PeerId
forall a. Set a
Set.empty Topic
topic Map Topic (Set PeerId)
meshNow
    peerMap <- readTVar (gsPeers router)
    -- Direct peers are never mesh candidates (gossipsub-v1.1.md
    -- explicit peering agreements); neither are floodsub peers, which
    -- have no mesh and would not understand the GRAFT (#157)
    let eligible = (Set PeerId -> PeerId -> PeerState -> Set PeerId)
-> Set PeerId -> Map PeerId PeerState -> Set PeerId
forall a k b. (a -> k -> b -> a) -> a -> Map k b -> a
Map.foldlWithKey' (\Set PeerId
acc PeerId
pid PeerState
ps ->
          if Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
             Bool -> Bool -> Bool
&& Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
currentMesh)
             Bool -> Bool -> Bool
&& PeerId
pid PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= GossipSubRouter -> PeerId
gsLocalPeerId GossipSubRouter
router
             Bool -> Bool -> Bool
&& Bool -> Bool
not (GossipSubRouter -> PeerId -> Bool
isDirectPeer GossipSubRouter
router PeerId
pid)
             Bool -> Bool -> Bool
&& PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerProtocol
FloodSubPeer
          then PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => a -> Set a -> Set a
Set.insert PeerId
pid Set PeerId
acc
          else Set PeerId
acc) Set PeerId
forall a. Set a
Set.empty Map PeerId PeerState
peerMap
    pure (foPeers, eligible)

  -- 4. Fill mesh to D if needed
  currentMesh <- atomically $ do
    m <- readTVar (gsMesh router)
    pure (Map.findWithDefault Set.empty topic m)
  let needed = GossipSubParams -> Int
paramD (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router) Int -> Int -> Int
forall a. Num a => a -> a -> a
- Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
currentMesh
  newPeers <- if needed > 0 && not (Set.null topicPeers)
    then do
      selected <- sampleIO (min needed (Set.size topicPeers)) (Set.toList topicPeers)
      let newSet = [PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
selected
      atomically $ modifyTVar' (gsMesh router) $
        Map.insertWith Set.union topic newSet
      pure newSet
    else pure Set.empty

  -- 5. Send GRAFT to all new mesh peers, including former fanout peers.
  -- fanoutPeers was captured before the mesh insert; currentMesh already
  -- contains the promoted peers, so it must not be subtracted here (#155).
  let allNewMeshPeers = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set PeerId
fanoutPeers Set PeerId
newPeers
  -- Start the P1 mesh clock for every peer entering the mesh (#156)
  now <- gsGetTime router
  atomically $ modifyTVar' (gsPeers router) $ \Map PeerId PeerState
pm ->
    (Map PeerId PeerState -> PeerId -> Map PeerId PeerState)
-> Map PeerId PeerState -> Set PeerId -> Map PeerId PeerState
forall a b. (a -> b -> a) -> a -> Set b -> a
Set.foldl' (\Map PeerId PeerState
m PeerId
pid -> (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (Topic -> UTCTime -> PeerState -> PeerState
markPeerInMesh Topic
topic UTCTime
now) PeerId
pid Map PeerId PeerState
m)
      Map PeerId PeerState
pm Set PeerId
allNewMeshPeers
  mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid (Topic -> RPC
graftRPC Topic
topic)) (Set.toList allNewMeshPeers)

-- | Unsubscribe from a topic (LEAVE): announce, PRUNE with backoff, delete mesh.
leave :: GossipSubRouter -> Topic -> IO ()
leave :: GossipSubRouter -> Topic -> IO ()
leave GossipSubRouter
router Topic
topic = do
  -- 1. Drop the subscription
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Set Topic) -> (Set Topic -> Set Topic) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Set Topic)
gsSubscriptions GossipSubRouter
router) (Topic -> Set Topic -> Set Topic
forall a. Ord a => a -> Set a -> Set a
Set.delete Topic
topic)

  -- 2. Announce unsubscription to all known peers
  peers <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  let allPeerIds = Map PeerId PeerState -> [PeerId]
forall k a. Map k a -> [k]
Map.keys Map PeerId PeerState
peers
      unsubRPC = RPC
emptyRPC { rpcSubscriptions = [SubOpts False topic] }
  mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
unsubRPC) allPeerIds

  -- 3. Send PRUNE with unsubscribe backoff and peer exchange to mesh
  -- peers, then delete (gossipsub-v1.1.md PRUNE peer exchange: help the
  -- pruned peer re-form its mesh without a discovery service)
  meshPeers <- atomically $ do
    m <- readTVar (gsMesh router)
    let mp = Set PeerId -> Topic -> Map Topic (Set PeerId) -> Set PeerId
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Set PeerId
forall a. Set a
Set.empty Topic
topic Map Topic (Set PeerId)
m
    modifyTVar' (gsMesh router) (Map.delete topic)
    pure mp
  let backoffSecs = NominalDiffTime -> Word64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (GossipSubParams -> NominalDiffTime
paramUnsubBackoff (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)) :: Word64
  mapM_ (\PeerId
pid -> do
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$
            (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (Topic -> PeerState -> PeerState
unmarkPeerInMesh Topic
topic) PeerId
pid
          prn <- GossipSubRouter -> PeerId -> Topic -> Bool -> Word64 -> IO Prune
buildPrune GossipSubRouter
router PeerId
pid Topic
topic Bool
True Word64
backoffSecs
          gsSendRPC router pid (pruneRPC prn))
    (Set.toList meshPeers)

-- Publishing

-- | Publish a message to a topic.
-- In StrictSign mode, signs the message and populates from/seqno/signature/key.
-- With FloodPublish=True (default), sends to ALL topic peers above PublishThreshold.
-- Otherwise, sends via mesh (or fanout if not subscribed).
publish :: GossipSubRouter -> Topic -> ByteString -> Maybe KeyPair -> IO ()
publish :: GossipSubRouter -> Topic -> ByteString -> Maybe KeyPair -> IO ()
publish GossipSubRouter
router Topic
topic ByteString
payload Maybe KeyPair
mKeyPair = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router

  -- Build message (with signing if StrictSign)
  msg <- case paramSignaturePolicy (gsParams router) of
    SignaturePolicy
StrictSign -> case Maybe KeyPair
mKeyPair of
      Maybe KeyPair
Nothing -> GossipSubError -> IO PubSubMessage
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> GossipSubError
SigningFailed String
"StrictSign publish requires a key pair")
      Just KeyPair
kp -> GossipSubRouter
-> Topic -> ByteString -> KeyPair -> IO PubSubMessage
mkSignedMessage GossipSubRouter
router Topic
topic ByteString
payload KeyPair
kp
    SignaturePolicy
StrictNoSign -> PubSubMessage -> IO PubSubMessage
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (PubSubMessage -> IO PubSubMessage)
-> PubSubMessage -> IO PubSubMessage
forall a b. (a -> b) -> a -> b
$ Topic -> ByteString -> PubSubMessage
mkUnsignedMessage Topic
topic ByteString
payload

  let msgId = GossipSubParams -> PubSubMessage -> ByteString
paramMessageIdFn (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router) PubSubMessage
msg

  -- Mark as seen and cache for IWANT/IHAVE: gossipsub-v1.0.md answers
  -- IWANT from the mcache, so our own messages must be cached too (#155).
  atomically $ do
    modifyTVar' (gsSeen router) (Map.insert msgId now)
    modifyTVar' (gsMessageCache router) (cachePut msgId msg)

  -- Build RPC with published message
  let pubRPC = RPC
emptyRPC { rpcPublish = [msg] }

  if paramFloodPublish (gsParams router)
    then do
      -- Flood publish: send to all topic peers scoring at or above the
      -- publish threshold (gossipsub-v1.1.md flood publishing). Direct
      -- peers always receive our messages regardless of score.
      peers <- readTVarIO (gsPeers router)
      ipMap <- readTVarIO (gsIPPeerCount router)
      let threshold = ScoreThresholds -> Double
stPublishThreshold (GossipSubRouter -> ScoreThresholds
gsThresholds GossipSubRouter
router)
          targets = ([PeerId] -> PeerId -> PeerState -> [PeerId])
-> [PeerId] -> Map PeerId PeerState -> [PeerId]
forall a k b. (a -> k -> b -> a) -> a -> Map k b -> a
Map.foldlWithKey' (\[PeerId]
acc PeerId
pid PeerState
ps ->
            if Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
               Bool -> Bool -> Bool
&& PeerId
pid PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= GossipSubRouter -> PeerId
gsLocalPeerId GossipSubRouter
router
               Bool -> Bool -> Bool
&& (GossipSubRouter -> PeerId -> Bool
isDirectPeer GossipSubRouter
router PeerId
pid
                   Bool -> Bool -> Bool
|| PeerScoreParams
-> PeerId
-> PeerState
-> Map ByteString (Set PeerId)
-> UTCTime
-> Double
computeScore (GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router) PeerId
pid PeerState
ps Map ByteString (Set PeerId)
ipMap UTCTime
now Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
>= Double
threshold)
            then PeerId
pid PeerId -> [PeerId] -> [PeerId]
forall a. a -> [a] -> [a]
: [PeerId]
acc
            else [PeerId]
acc) [] Map PeerId PeerState
peers
      mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
pubRPC) targets
    else do
      -- Mesh-based publish; subscribed direct and floodsub peers are
      -- always included although they are never mesh or fanout members
      -- (floodsub peers are flooded every message for their topics, #157)
      peers <- readTVarIO (gsPeers router)
      let direct = GossipSubRouter -> Map PeerId PeerState -> Topic -> Set PeerId
directTopicPeers GossipSubRouter
router Map PeerId PeerState
peers Topic
topic
          flood  = Map PeerId PeerState -> Topic -> Set PeerId
floodSubTopicPeers Map PeerId PeerState
peers Topic
topic
      meshPeers <- atomically $ do
        m <- readTVar (gsMesh router)
        pure (Map.findWithDefault Set.empty topic m)
      if not (Set.null meshPeers)
        then mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
pubRPC)
               (Set.toList (Set.unions [meshPeers, direct, flood]))
        else do
          -- Fanout: use existing or create new (direct and floodsub
          -- peers excluded from selection but always sent to)
          foPeers <- atomically $ do
            fo <- readTVar (gsFanout router)
            pure (Map.findWithDefault Set.empty topic fo)
          targets <- if Set.null foPeers
            then do
              let eligible = ([PeerId] -> PeerId -> PeerState -> [PeerId])
-> [PeerId] -> Map PeerId PeerState -> [PeerId]
forall a k b. (a -> k -> b -> a) -> a -> Map k b -> a
Map.foldlWithKey' (\[PeerId]
acc PeerId
pid PeerState
ps ->
                    if Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
                       Bool -> Bool -> Bool
&& PeerId
pid PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= GossipSubRouter -> PeerId
gsLocalPeerId GossipSubRouter
router
                       Bool -> Bool -> Bool
&& Bool -> Bool
not (GossipSubRouter -> PeerId -> Bool
isDirectPeer GossipSubRouter
router PeerId
pid)
                       Bool -> Bool -> Bool
&& PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerProtocol
FloodSubPeer
                    then PeerId
pid PeerId -> [PeerId] -> [PeerId]
forall a. a -> [a] -> [a]
: [PeerId]
acc
                    else [PeerId]
acc) [] Map PeerId PeerState
peers
              selected <- sampleIO (min (paramD (gsParams router)) (length eligible)) eligible
              let selectedSet = [PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
selected
              atomically $ do
                modifyTVar' (gsFanout router) (Map.insert topic selectedSet)
                modifyTVar' (gsFanoutPub router) (Map.insert topic now)
              pure selectedSet
            else do
              atomically $ modifyTVar' (gsFanoutPub router) (Map.insert topic now)
              pure foPeers
          mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
pubRPC)
            (Set.toList (Set.unions [targets, direct, flood]))

  -- Deliver to local application
  onMsg <- readTVarIO (gsOnMessage router)
  onMsg topic msg

-- Inbound RPC handling

-- | Handle an inbound RPC from a peer.
--
-- Graylisted peers (score below 'stGraylistThreshold') have their RPCs
-- ignored entirely (gossipsub-v1.1.md graylist). Direct peers are exempt:
-- explicit peering agreements exchange messages unconditionally.
handleRPC :: GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC :: GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC GossipSubRouter
router PeerId
sender RPC
rpc = do
  score <- GossipSubRouter -> PeerId -> IO Double
peerScore GossipSubRouter
router PeerId
sender
  if score < stGraylistThreshold (gsThresholds router)
       && not (isDirectPeer router sender)
    then pure ()
    else handleRPC' router sender rpc

handleRPC' :: GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC' :: GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC' GossipSubRouter
router PeerId
sender RPC
rpc = do
  -- Process subscriptions
  GossipSubRouter -> PeerId -> [SubOpts] -> IO ()
handleSubscriptions GossipSubRouter
router PeerId
sender (RPC -> [SubOpts]
rpcSubscriptions RPC
rpc)

  -- Process published messages
  (PubSubMessage -> IO ()) -> [PubSubMessage] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
handlePublishedMessage GossipSubRouter
router PeerId
sender) (RPC -> [PubSubMessage]
rpcPublish RPC
rpc)

  -- Process control messages
  case RPC -> Maybe ControlMessage
rpcControl RPC
rpc of
    Maybe ControlMessage
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    Just ControlMessage
ctrl -> do
      GossipSubRouter -> PeerId -> [IHave] -> IO ()
handleIHave GossipSubRouter
router PeerId
sender (ControlMessage -> [IHave]
ctrlIHave ControlMessage
ctrl)
      GossipSubRouter -> PeerId -> [IWant] -> IO ()
handleIWant GossipSubRouter
router PeerId
sender (ControlMessage -> [IWant]
ctrlIWant ControlMessage
ctrl)
      GossipSubRouter -> PeerId -> [Graft] -> IO ()
handleGraft GossipSubRouter
router PeerId
sender (ControlMessage -> [Graft]
ctrlGraft ControlMessage
ctrl)
      GossipSubRouter -> PeerId -> [Prune] -> IO ()
handlePrune GossipSubRouter
router PeerId
sender (ControlMessage -> [Prune]
ctrlPrune ControlMessage
ctrl)

-- | Process a published message: verify, deduplicate, validate, forward, deliver.
--
-- Signature verification runs before deduplication so that an invalid message
-- is never cached, forwarded or delivered, and never poisons the seen cache for
-- the genuine message with the same ID.
handlePublishedMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
handlePublishedMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
handlePublishedMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg =
  case SignaturePolicy -> PubSubMessage -> Either ValidationError ()
validateMessage (GossipSubParams -> SignaturePolicy
paramSignaturePolicy (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)) PubSubMessage
msg of
    Left ValidationError
_err -> GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
rejectMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg
    Right ()  -> do
      let msgId :: ByteString
msgId = GossipSubParams -> PubSubMessage -> ByteString
paramMessageIdFn (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router) PubSubMessage
msg
          topic :: Topic
topic = PubSubMessage -> Topic
msgTopic PubSubMessage
msg
      now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router

      -- Deduplicate, keeping the first-seen time for the P3 near-first window
      mFirstSeen <- atomically $ do
        s <- readTVar (gsSeen router)
        case Map.lookup msgId s of
          Just UTCTime
firstSeen -> Maybe UTCTime -> STM (Maybe UTCTime)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just UTCTime
firstSeen)
          Maybe UTCTime
Nothing -> do
            TVar (Map ByteString UTCTime) -> Map ByteString UTCTime -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (GossipSubRouter -> TVar (Map ByteString UTCTime)
gsSeen GossipSubRouter
router) (ByteString
-> UTCTime -> Map ByteString UTCTime -> Map ByteString UTCTime
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert ByteString
msgId UTCTime
now Map ByteString UTCTime
s)
            Maybe UTCTime -> STM (Maybe UTCTime)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe UTCTime
forall a. Maybe a
Nothing

      -- A signature-valid delivery fulfils any outstanding IWANT promise
      -- for this message ID (P7 promise tracking, gossipsub-v1.1.md)
      atomically $ modifyTVar' (gsIWantPromises router) $
        Map.filterWithKey (\(PeerId
_, ByteString
mid) UTCTime
_ -> ByteString
mid ByteString -> ByteString -> Bool
forall a. Eq a => a -> a -> Bool
/= ByteString
msgId)

      case mFirstSeen of
        Just UTCTime
firstSeen ->
          -- Duplicate: count as a mesh delivery (P3) when the sender is
          -- in our mesh and delivered within the near-first window
          GossipSubRouter
-> PeerId -> Topic -> Maybe (UTCTime, UTCTime) -> IO ()
creditMeshDelivery GossipSubRouter
router PeerId
sender Topic
topic ((UTCTime, UTCTime) -> Maybe (UTCTime, UTCTime)
forall a. a -> Maybe a
Just (UTCTime
firstSeen, UTCTime
now))
        Maybe UTCTime
Nothing -> do
          outcome <- GossipSubRouter -> TopicValidator
runTopicValidator GossipSubRouter
router PeerId
sender PubSubMessage
msg
          case outcome of
            -- Provably invalid: drop and penalise the source (P4)
            ValidationResult
ValidationReject -> GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
rejectMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg
            -- Undecidable: drop without penalty (extended validators,
            -- gossipsub-v1.1.md — Ignore must not affect the score)
            ValidationResult
ValidationIgnore -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            ValidationResult
ValidationAccept -> do
              -- First valid delivery: P2, plus P3 for mesh senders (#156)
              GossipSubRouter -> PeerId -> Topic -> IO ()
creditFirstDelivery GossipSubRouter
router PeerId
sender Topic
topic
              GossipSubRouter
-> PeerId -> Topic -> Maybe (UTCTime, UTCTime) -> IO ()
creditMeshDelivery GossipSubRouter
router PeerId
sender Topic
topic Maybe (UTCTime, UTCTime)
forall a. Maybe a
Nothing

              -- Cache the message for IWANT responses
              STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar MessageCache -> (MessageCache -> MessageCache) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar MessageCache
gsMessageCache GossipSubRouter
router) ((MessageCache -> MessageCache) -> STM ())
-> (MessageCache -> MessageCache) -> STM ()
forall a b. (a -> b) -> a -> b
$
                ByteString -> PubSubMessage -> MessageCache -> MessageCache
cachePut ByteString
msgId PubSubMessage
msg

              -- Forward to mesh peers (excluding sender)
              GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
forwardMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg

              -- Deliver to application
              onMsg <- TVar (Topic -> PubSubMessage -> IO ())
-> IO (Topic -> PubSubMessage -> IO ())
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Topic -> PubSubMessage -> IO ())
gsOnMessage GossipSubRouter
router)
              onMsg (msgTopic msg) msg

-- | Record a P2 first-message delivery for the sender.
creditFirstDelivery :: GossipSubRouter -> PeerId -> Topic -> IO ()
creditFirstDelivery :: GossipSubRouter -> PeerId -> Topic -> IO ()
creditFirstDelivery GossipSubRouter
router PeerId
sender Topic
topic = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$ (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust PeerState -> PeerState
bump PeerId
sender
  where
    tsp :: TopicScoreParams
tsp = TopicScoreParams
-> Topic -> Map Topic TopicScoreParams -> TopicScoreParams
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicScoreParams
defaultTopicScoreParams Topic
topic
      (PeerScoreParams -> Map Topic TopicScoreParams
pspTopicParams (GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router))
    bump :: PeerState -> PeerState
bump PeerState
ps =
      let tps :: TopicPeerState
tps = TopicPeerState
-> Topic -> Map Topic TopicPeerState -> TopicPeerState
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicPeerState
defaultTopicPeerState Topic
topic (PeerState -> Map Topic TopicPeerState
psTopicState PeerState
ps)
      in PeerState
ps { psTopicState =
                Map.insert topic (recordFirstDelivery tsp tps) (psTopicState ps) }

-- | Record a P3 mesh delivery for a sender in our mesh. For duplicates,
-- the delivery only counts inside the near-first window after the first
-- sighting (gossipsub-v1.1.md mesh message delivery rate).
creditMeshDelivery :: GossipSubRouter -> PeerId -> Topic
                   -> Maybe (UTCTime, UTCTime) -> IO ()
creditMeshDelivery :: GossipSubRouter
-> PeerId -> Topic -> Maybe (UTCTime, UTCTime) -> IO ()
creditMeshDelivery GossipSubRouter
router PeerId
sender Topic
topic Maybe (UTCTime, UTCTime)
mWindow = do
  meshMap <- TVar (Map Topic (Set PeerId)) -> IO (Map Topic (Set PeerId))
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map Topic (Set PeerId))
gsMesh GossipSubRouter
router)
  let inMesh = PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
sender (Set PeerId -> Topic -> Map Topic (Set PeerId) -> Set PeerId
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Set PeerId
forall a. Set a
Set.empty Topic
topic Map Topic (Set PeerId)
meshMap)
      tsp = TopicScoreParams
-> Topic -> Map Topic TopicScoreParams -> TopicScoreParams
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicScoreParams
defaultTopicScoreParams Topic
topic
        (PeerScoreParams -> Map Topic TopicScoreParams
pspTopicParams (GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router))
      withinWindow = case Maybe (UTCTime, UTCTime)
mWindow of
        Maybe (UTCTime, UTCTime)
Nothing -> Bool
True
        Just (UTCTime
firstSeen, UTCTime
now) ->
          UTCTime -> UTCTime -> NominalDiffTime
diffUTCTime UTCTime
now UTCTime
firstSeen NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
<= TopicScoreParams -> NominalDiffTime
tspMeshMessageDeliveryWindow TopicScoreParams
tsp
  when (inMesh && withinWindow) $ atomically $
    modifyTVar' (gsPeers router) $ Map.adjust
      (\PeerState
ps ->
        let tps :: TopicPeerState
tps = TopicPeerState
-> Topic -> Map Topic TopicPeerState -> TopicPeerState
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicPeerState
defaultTopicPeerState Topic
topic (PeerState -> Map Topic TopicPeerState
psTopicState PeerState
ps)
        in PeerState
ps { psTopicState =
                  Map.insert topic (recordMeshDelivery tsp tps) (psTopicState ps) })
      sender

-- | Run the topic validator, if one is registered. No validator means accept.
runTopicValidator :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ValidationResult
runTopicValidator :: GossipSubRouter -> TopicValidator
runTopicValidator GossipSubRouter
router PeerId
sender PubSubMessage
msg = do
  validators <- TVar (Map Topic TopicValidator) -> IO (Map Topic TopicValidator)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map Topic TopicValidator)
gsValidators GossipSubRouter
router)
  case Map.lookup (msgTopic msg) validators of
    Maybe TopicValidator
Nothing -> ValidationResult -> IO ValidationResult
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ValidationResult
ValidationAccept
    Just TopicValidator
v  -> TopicValidator
v PeerId
sender PubSubMessage
msg

-- | Drop a message and charge the propagation source a P4 invalid delivery.
rejectMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
rejectMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
rejectMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$ (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust PeerState -> PeerState
bumpInvalid PeerId
sender
  where
    topic :: Topic
topic = PubSubMessage -> Topic
msgTopic PubSubMessage
msg
    bumpInvalid :: PeerState -> PeerState
bumpInvalid PeerState
ps =
      let tps :: TopicPeerState
tps = TopicPeerState
-> Topic -> Map Topic TopicPeerState -> TopicPeerState
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicPeerState
defaultTopicPeerState Topic
topic (PeerState -> Map Topic TopicPeerState
psTopicState PeerState
ps)
      in PeerState
ps { psTopicState = Map.insert topic (recordInvalidMessage tps) (psTopicState ps) }

-- Control message handlers

-- | Handle GRAFT: accept if subscribed, non-negative score, and no backoff.
handleGraft :: GossipSubRouter -> PeerId -> [Graft] -> IO ()
handleGraft :: GossipSubRouter -> PeerId -> [Graft] -> IO ()
handleGraft GossipSubRouter
router PeerId
sender [Graft]
grafts = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  pruneResponses <- mapM (handleOneGraft router sender now) grafts
  let prunes = [[Prune]] -> [Prune]
forall (t :: * -> *) a. Foldable t => t [a] -> [a]
concat [[Prune]]
pruneResponses
  unless (null prunes) $
    gsSendRPC router sender emptyRPC
      { rpcControl = Just emptyControlMessage { ctrlPrune = prunes } }

-- | Handle a single GRAFT request.
handleOneGraft :: GossipSubRouter -> PeerId -> UTCTime -> Graft -> IO [Prune]
handleOneGraft :: GossipSubRouter -> PeerId -> UTCTime -> Graft -> IO [Prune]
handleOneGraft GossipSubRouter
router PeerId
sender UTCTime
now (Graft Topic
topic) = do
  -- Check the subscription set, not mesh key presence: a topic joined
  -- with no peers has no mesh entry but its GRAFTs must be accepted
  -- (gossipsub-v1.0.md GRAFT handling; issue #155).
  subs <- TVar (Set Topic) -> IO (Set Topic)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Set Topic)
gsSubscriptions GossipSubRouter
router)
  peersMap <- readTVarIO (gsPeers router)
  let subscribed = Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic Set Topic
subs

  if isFloodSubPeer peersMap sender
    then
      -- Floodsub peers have no mesh and understand no control messages:
      -- never graft them and never answer with PRUNE (#157)
      pure []
    else if not subscribed
    then
      -- gossipsub-v1.1.md GRAFT flood protection: GRAFTs for unknown
      -- topics are ignored — replying with PRUNE (the v1.0 behaviour)
      -- lets an attacker elicit traffic with spam GRAFTs (#157).
      pure []
    else if isDirectPeer router sender
      then do
        -- gossipsub-v1.1.md explicit peering: direct peers must never be
        -- mesh members — answer with PRUNE and do not graft
        prn <- buildPrune router sender topic False
          (round (paramPruneBackoff (gsParams router)))
        pure [prn]
    else do
      -- Check backoff
      backoffMap <- readTVarIO (gsBackoff router)
      let inBackoff = case (PeerId, Topic) -> Map (PeerId, Topic) UTCTime -> Maybe UTCTime
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup (PeerId
sender, Topic
topic) Map (PeerId, Topic) UTCTime
backoffMap of
            Maybe UTCTime
Nothing -> Bool
False
            Just UTCTime
expires -> UTCTime
now UTCTime -> UTCTime -> Bool
forall a. Ord a => a -> a -> Bool
< UTCTime
expires

      score <- peerScore router sender

      -- Any rejection PRUNEs with a fresh backoff and never includes
      -- peer exchange (no PX for misbehaving or negative-score peers)
      let backoffSecs = NominalDiffTime -> Word64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (GossipSubParams -> NominalDiffTime
paramPruneBackoff (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)) :: Word64
          rejectWithBackoff = do
            STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map (PeerId, Topic) UTCTime)
-> (Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map (PeerId, Topic) UTCTime)
gsBackoff GossipSubRouter
router) ((Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
 -> STM ())
-> (Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
-> STM ()
forall a b. (a -> b) -> a -> b
$
              (PeerId, Topic)
-> UTCTime
-> Map (PeerId, Topic) UTCTime
-> Map (PeerId, Topic) UTCTime
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert (PeerId
sender, Topic
topic)
                (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime (GossipSubParams -> NominalDiffTime
paramPruneBackoff (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)) UTCTime
now)
            prn <- GossipSubRouter -> PeerId -> Topic -> Bool -> Word64 -> IO Prune
buildPrune GossipSubRouter
router PeerId
sender Topic
topic Bool
False Word64
backoffSecs
            pure [prn]

      if inBackoff
        then do
          -- GRAFT flood protection: re-GRAFTing inside the backoff window
          -- is a protocol violation — penalise (P7) and prune with backoff
          atomically $ modifyTVar' (gsPeers router) $
            Map.adjust addP7Penalty sender
          rejectWithBackoff
        else if score < 0
          then rejectWithBackoff
          else do
            -- Accept: add sender to mesh and start its P1 mesh clock
            atomically $ do
              modifyTVar' (gsMesh router) $
                Map.insertWith Set.union topic (Set.singleton sender)
              modifyTVar' (gsPeers router) $
                Map.adjust (markPeerInMesh topic now) sender
            pure []

-- | Handle PRUNE: remove from mesh and start backoff.
handlePrune :: GossipSubRouter -> PeerId -> [Prune] -> IO ()
handlePrune :: GossipSubRouter -> PeerId -> [Prune] -> IO ()
handlePrune GossipSubRouter
router PeerId
sender [Prune]
prunes = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  mapM_ (handleOnePrune router sender now) prunes

handleOnePrune :: GossipSubRouter -> PeerId -> UTCTime -> Prune -> IO ()
handleOnePrune :: GossipSubRouter -> PeerId -> UTCTime -> Prune -> IO ()
handleOnePrune GossipSubRouter
router PeerId
sender UTCTime
now Prune
prune = do
  let topic :: Topic
topic = Prune -> Topic
pruneTopic Prune
prune
  -- Record P3b mesh failure: snapshot delivery deficit before removing
  let scoreParams :: PeerScoreParams
scoreParams = GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router
  case Topic -> Map Topic TopicScoreParams -> Maybe TopicScoreParams
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Topic
topic (PeerScoreParams -> Map Topic TopicScoreParams
pspTopicParams PeerScoreParams
scoreParams) of
    Just TopicScoreParams
tsp -> STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$
      (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (\PeerState
ps ->
        let topicSt :: TopicPeerState
topicSt = TopicPeerState
-> Topic -> Map Topic TopicPeerState -> TopicPeerState
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault TopicPeerState
defaultTopicPeerState Topic
topic (PeerState -> Map Topic TopicPeerState
psTopicState PeerState
ps)
            topicSt' :: TopicPeerState
topicSt' = TopicScoreParams -> TopicPeerState -> TopicPeerState
recordMeshFailure TopicScoreParams
tsp TopicPeerState
topicSt
        in PeerState
ps { psTopicState = Map.insert topic topicSt' (psTopicState ps) }
      ) PeerId
sender
    Maybe TopicScoreParams
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  -- Remove sender from mesh and stop its P1 mesh clock
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    TVar (Map Topic (Set PeerId))
-> (Map Topic (Set PeerId) -> Map Topic (Set PeerId)) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map Topic (Set PeerId))
gsMesh GossipSubRouter
router) ((Map Topic (Set PeerId) -> Map Topic (Set PeerId)) -> STM ())
-> (Map Topic (Set PeerId) -> Map Topic (Set PeerId)) -> STM ()
forall a b. (a -> b) -> a -> b
$
      (Set PeerId -> Set PeerId)
-> Topic -> Map Topic (Set PeerId) -> Map Topic (Set PeerId)
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => a -> Set a -> Set a
Set.delete PeerId
sender) Topic
topic
    TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$
      (PeerState -> PeerState)
-> PeerId -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (Topic -> PeerState -> PeerState
unmarkPeerInMesh Topic
topic) PeerId
sender
  -- Start backoff timer
  let backoffDuration :: NominalDiffTime
backoffDuration = case Prune -> Maybe Word64
pruneBackoff Prune
prune of
        Just Word64
secs -> Word64 -> NominalDiffTime
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word64
secs
        Maybe Word64
Nothing   -> GossipSubParams -> NominalDiffTime
paramPruneBackoff (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
      expires :: UTCTime
expires = NominalDiffTime -> UTCTime -> UTCTime
addUTCTime NominalDiffTime
backoffDuration UTCTime
now
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map (PeerId, Topic) UTCTime)
-> (Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map (PeerId, Topic) UTCTime)
gsBackoff GossipSubRouter
router) ((Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
 -> STM ())
-> (Map (PeerId, Topic) UTCTime -> Map (PeerId, Topic) UTCTime)
-> STM ()
forall a b. (a -> b) -> a -> b
$
    (PeerId, Topic)
-> UTCTime
-> Map (PeerId, Topic) UTCTime
-> Map (PeerId, Topic) UTCTime
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert (PeerId
sender, Topic
topic) UTCTime
expires
  -- Honour peer exchange, but only from peers whose score clears the
  -- PX acceptance threshold (gossipsub-v1.1.md: PX from low-scoring
  -- peers is an eclipse-attack vector)
  -- Any attached signed peer record must verify (RFC 0003 envelope
  -- opens and names the advertised peer) — a record that fails drops
  -- its whole PX entry, matching go-libp2p's pxConnect. Entries without
  -- a record are passed through unchanged.
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([PeerExchangeInfo] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null (Prune -> [PeerExchangeInfo]
prunePeers Prune
prune)) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    score <- GossipSubRouter -> PeerId -> IO Double
peerScore GossipSubRouter
router PeerId
sender
    when (score >= stAcceptPXThreshold (gsThresholds router)) $ do
      let verified = (PeerExchangeInfo -> Bool)
-> [PeerExchangeInfo] -> [PeerExchangeInfo]
forall a. (a -> Bool) -> [a] -> [a]
filter PeerExchangeInfo -> Bool
validPXRecord (Prune -> [PeerExchangeInfo]
prunePeers Prune
prune)
      unless (null verified) $ do
        onPX <- readTVarIO (gsOnPeerExchange router)
        onPX topic verified

-- | A PX entry is acceptable if it carries no signed record, or a
-- record whose envelope verifies and whose subject is the advertised
-- peer id.
validPXRecord :: PeerExchangeInfo -> Bool
validPXRecord :: PeerExchangeInfo -> Bool
validPXRecord PeerExchangeInfo
pxi = case PeerExchangeInfo -> Maybe ByteString
pxSignedPeerRecord PeerExchangeInfo
pxi of
  Maybe ByteString
Nothing -> Bool
True
  Just ByteString
envBytes -> case ByteString -> Either String (SignedEnvelope, PeerRecord)
openPeerRecordEnvelope ByteString
envBytes of
    Right (SignedEnvelope
_, PeerRecord
record) -> PeerRecord -> ByteString
prPeerId PeerRecord
record ByteString -> ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== PeerExchangeInfo -> ByteString
pxPeerId PeerExchangeInfo
pxi
    Left String
_ -> Bool
False

-- | Handle IHAVE: request unseen messages via IWANT.
--
-- Gossip from peers below the gossip threshold is ignored
-- (gossipsub-v1.1.md gossip threshold). One advertised-and-requested
-- message ID is tracked as an IWANT promise: if the peer never delivers
-- it before the follow-up deadline, it is a P7 behavioural violation.
--
-- IHAVE flood protection (#157): at most 'paramMaxIHaveMessages' IHAVE
-- batches are accepted per peer per heartbeat, and at most
-- 'paramMaxIHaveLength' message ids are requested from a peer per
-- heartbeat (go-libp2p defaults 10 and 5000).
handleIHave :: GossipSubRouter -> PeerId -> [IHave] -> IO ()
handleIHave :: GossipSubRouter -> PeerId -> [IHave] -> IO ()
handleIHave GossipSubRouter
router PeerId
sender [IHave]
ihaves = do
  score <- GossipSubRouter -> PeerId -> IO Double
peerScore GossipSubRouter
router PeerId
sender
  peersMap <- readTVarIO (gsPeers router)
  -- Floodsub peers never receive control messages, so an (unexpected)
  -- IHAVE from one must not be answered with IWANT (#157)
  unless (score < stGossipThreshold (gsThresholds router)
          || isFloodSubPeer peersMap sender
          || null ihaves) $ do
    let params = GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router
    withinBudget <- atomically $ do
      counts <- readTVar (gsIHaveCounts router)
      let n = Int -> PeerId -> Map PeerId Int -> Int
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Int
0 PeerId
sender Map PeerId Int
counts Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1
      writeTVar (gsIHaveCounts router) (Map.insert sender n counts)
      pure (n <= paramMaxIHaveMessages params)
    when withinBudget $ do
      seenMap <- readTVarIO (gsSeen router)
      let unseen = (IHave -> [ByteString]) -> [IHave] -> [ByteString]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap (\(IHave Topic
_ [ByteString]
mids) ->
            (ByteString -> Bool) -> [ByteString] -> [ByteString]
forall a. (a -> Bool) -> [a] -> [a]
filter (\ByteString
mid -> Bool -> Bool
not (ByteString -> Map ByteString UTCTime -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.member ByteString
mid Map ByteString UTCTime
seenMap)) [ByteString]
mids) [IHave]
ihaves
      toAsk <- atomically $ do
        asked <- readTVar (gsIAskedCounts router)
        let a = Int -> PeerId -> Map PeerId Int -> Int
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Int
0 PeerId
sender Map PeerId Int
asked
            budget = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 (GossipSubParams -> Int
paramMaxIHaveLength GossipSubParams
params Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
a)
            capped = Int -> [ByteString] -> [ByteString]
forall a. Int -> [a] -> [a]
take Int
budget [ByteString]
unseen
        writeTVar (gsIAskedCounts router)
          (Map.insert sender (a + length capped) asked)
        pure capped
      unless (null toAsk) $ do
        now <- gsGetTime router
        promised <- sampleIO 1 toAsk
        let deadline = NominalDiffTime -> UTCTime -> UTCTime
addUTCTime (GossipSubParams -> NominalDiffTime
paramIWantFollowupTime GossipSubParams
params) UTCTime
now
        atomically $ modifyTVar' (gsIWantPromises router) $ \Map (PeerId, ByteString) UTCTime
m ->
          (ByteString
 -> Map (PeerId, ByteString) UTCTime
 -> Map (PeerId, ByteString) UTCTime)
-> Map (PeerId, ByteString) UTCTime
-> [ByteString]
-> Map (PeerId, ByteString) UTCTime
forall a b. (a -> b -> b) -> b -> [a] -> b
forall (t :: * -> *) a b.
Foldable t =>
(a -> b -> b) -> b -> t a -> b
foldr (\ByteString
mid -> (PeerId, ByteString)
-> UTCTime
-> Map (PeerId, ByteString) UTCTime
-> Map (PeerId, ByteString) UTCTime
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert (PeerId
sender, ByteString
mid) UTCTime
deadline) Map (PeerId, ByteString) UTCTime
m [ByteString]
promised
        gsSendRPC router sender emptyRPC
          { rpcControl = Just emptyControlMessage { ctrlIWant = [IWant toAsk] } }

-- | Handle IWANT: respond with cached messages from the message cache.
-- Requests from peers below the gossip threshold are ignored, and at
-- most 'paramMaxIHaveLength' messages are served per peer per heartbeat
-- (IWANT flood protection, #157).
handleIWant :: GossipSubRouter -> PeerId -> [IWant] -> IO ()
handleIWant :: GossipSubRouter -> PeerId -> [IWant] -> IO ()
handleIWant GossipSubRouter
router PeerId
sender [IWant]
iwants = do
  score <- GossipSubRouter -> PeerId -> IO Double
peerScore GossipSubRouter
router PeerId
sender
  unless (score < stGossipThreshold (gsThresholds router)) $ do
    cache <- readTVarIO (gsMessageCache router)
    let requestedIds = (IWant -> [ByteString]) -> [IWant] -> [ByteString]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap IWant -> [ByteString]
iwantMessageIds [IWant]
iwants
        found = [ PubSubMessage
msg | ByteString
mid <- [ByteString]
requestedIds
                       , Just PubSubMessage
msg <- [ByteString -> MessageCache -> Maybe PubSubMessage
cacheGet ByteString
mid MessageCache
cache] ]
    toServe <- atomically $ do
      served <- readTVar (gsIWantServed router)
      let s = Int -> PeerId -> Map PeerId Int -> Int
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Int
0 PeerId
sender Map PeerId Int
served
          budget = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 (GossipSubParams -> Int
paramMaxIHaveLength (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router) Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
s)
          capped = Int -> [PubSubMessage] -> [PubSubMessage]
forall a. Int -> [a] -> [a]
take Int
budget [PubSubMessage]
found
      writeTVar (gsIWantServed router)
        (Map.insert sender (s + length capped) served)
      pure capped
    unless (null toServe) $
      gsSendRPC router sender emptyRPC { rpcPublish = toServe }

-- | Handle subscription changes from a peer.
handleSubscriptions :: GossipSubRouter -> PeerId -> [SubOpts] -> IO ()
handleSubscriptions :: GossipSubRouter -> PeerId -> [SubOpts] -> IO ()
handleSubscriptions GossipSubRouter
router PeerId
sender [SubOpts]
subs = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map PeerId PeerState)
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router) ((Map PeerId PeerState -> Map PeerId PeerState) -> STM ())
-> (Map PeerId PeerState -> Map PeerId PeerState) -> STM ()
forall a b. (a -> b) -> a -> b
$ \Map PeerId PeerState
peerMap ->
    case PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
sender Map PeerId PeerState
peerMap of
      Maybe PeerState
Nothing -> Map PeerId PeerState
peerMap  -- Unknown peer, ignore
      Just PeerState
ps ->
        let topics' :: Set Topic
topics' = (Set Topic -> SubOpts -> Set Topic)
-> Set Topic -> [SubOpts] -> Set Topic
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl (\Set Topic
ts SubOpts
sub ->
              if SubOpts -> Bool
subSubscribe SubOpts
sub
                then Topic -> Set Topic -> Set Topic
forall a. Ord a => a -> Set a -> Set a
Set.insert (SubOpts -> Topic
subTopicId SubOpts
sub) Set Topic
ts
                else Topic -> Set Topic -> Set Topic
forall a. Ord a => a -> Set a -> Set a
Set.delete (SubOpts -> Topic
subTopicId SubOpts
sub) Set Topic
ts
              ) (PeerState -> Set Topic
psTopics PeerState
ps) [SubOpts]
subs
        in PeerId -> PeerState -> Map PeerId PeerState -> Map PeerId PeerState
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
sender PeerState
ps { psTopics = topics' } Map PeerId PeerState
peerMap

-- Message forwarding

-- | Forward a message to mesh peers for its topic, excluding the sender.
-- Subscribed direct peers always receive the message even though they
-- are never mesh members (gossipsub-v1.1.md explicit peering), and so do
-- subscribed floodsub peers, which are flooded before the mesh
-- (gossipsub-v1.0.md: forward "to every peer in peers.floodsub[topic]").
forwardMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
forwardMessage :: GossipSubRouter -> PeerId -> PubSubMessage -> IO ()
forwardMessage GossipSubRouter
router PeerId
sender PubSubMessage
msg = do
  let topic :: Topic
topic = PubSubMessage -> Topic
msgTopic PubSubMessage
msg
  meshPeers <- STM (Set PeerId) -> IO (Set PeerId)
forall a. STM a -> IO a
atomically (STM (Set PeerId) -> IO (Set PeerId))
-> STM (Set PeerId) -> IO (Set PeerId)
forall a b. (a -> b) -> a -> b
$ do
    m <- TVar (Map Topic (Set PeerId)) -> STM (Map Topic (Set PeerId))
forall a. TVar a -> STM a
readTVar (GossipSubRouter -> TVar (Map Topic (Set PeerId))
gsMesh GossipSubRouter
router)
    pure (Map.findWithDefault Set.empty topic m)
  peers <- readTVarIO (gsPeers router)
  let direct = GossipSubRouter -> Map PeerId PeerState -> Topic -> Set PeerId
directTopicPeers GossipSubRouter
router Map PeerId PeerState
peers Topic
topic
      flood  = Map PeerId PeerState -> Topic -> Set PeerId
floodSubTopicPeers Map PeerId PeerState
peers Topic
topic
      targets = PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => a -> Set a -> Set a
Set.delete PeerId
sender ([Set PeerId] -> Set PeerId
forall (f :: * -> *) a. (Foldable f, Ord a) => f (Set a) -> Set a
Set.unions [Set PeerId
meshPeers, Set PeerId
direct, Set PeerId
flood])
      fwdRPC = RPC
emptyRPC { rpcPublish = [msg] }
  mapM_ (\PeerId
pid -> GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
fwdRPC) (Set.toList targets)

-- Scoring

-- | Compute peer score using Score.computeScore (P1-P7).
peerScore :: GossipSubRouter -> PeerId -> IO Double
peerScore :: GossipSubRouter -> PeerId -> IO Double
peerScore GossipSubRouter
router PeerId
pid = do
  peers <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
  now <- gsGetTime router
  ipMap <- readTVarIO (gsIPPeerCount router)
  case Map.lookup pid peers of
    Maybe PeerState
Nothing -> Double -> IO Double
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Double
0
    Just PeerState
ps -> Double -> IO Double
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Double -> IO Double) -> Double -> IO Double
forall a b. (a -> b) -> a -> b
$ PeerScoreParams
-> PeerId
-> PeerState
-> Map ByteString (Set PeerId)
-> UTCTime
-> Double
computeScore (GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router) PeerId
pid PeerState
ps Map ByteString (Set PeerId)
ipMap UTCTime
now

-- Peer exchange

-- | Store the encoded signed peer record (RFC 0003 envelope bytes) for
-- a peer, typically obtained via identify. It is attached to PX entries
-- advertising that peer in outgoing PRUNEs (gossipsub-v1.1.md). Cleared
-- by 'removePeer'.
setSignedPeerRecord :: GossipSubRouter -> PeerId -> ByteString -> IO ()
setSignedPeerRecord :: GossipSubRouter -> PeerId -> ByteString -> IO ()
setSignedPeerRecord GossipSubRouter
router PeerId
pid ByteString
recBytes = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
  TVar (Map PeerId ByteString)
-> (Map PeerId ByteString -> Map PeerId ByteString) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map PeerId ByteString)
gsSignedPeerRecords GossipSubRouter
router) (PeerId
-> ByteString -> Map PeerId ByteString -> Map PeerId ByteString
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
pid ByteString
recBytes)

-- | Select peer-exchange records for a PRUNE: up to 'paramPrunePeers'
-- random peers subscribed to the topic with non-negative score,
-- excluding the pruned peer itself (gossipsub-v1.1.md peer exchange).
-- Peers whose signed peer record we hold get it attached.
selectPXPeers :: GossipSubRouter -> Topic -> PeerId -> IO [PeerExchangeInfo]
selectPXPeers :: GossipSubRouter -> Topic -> PeerId -> IO [PeerExchangeInfo]
selectPXPeers GossipSubRouter
router Topic
topic PeerId
excluded = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  peers <- readTVarIO (gsPeers router)
  ipMap <- readTVarIO (gsIPPeerCount router)
  signedRecords <- readTVarIO (gsSignedPeerRecords router)
  let candidates =
        [ PeerId
pid | (PeerId
pid, PeerState
ps) <- Map PeerId PeerState -> [(PeerId, PeerState)]
forall k a. Map k a -> [(k, a)]
Map.toList Map PeerId PeerState
peers
              , PeerId
pid PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerId
excluded
              , PeerId
pid PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= GossipSubRouter -> PeerId
gsLocalPeerId GossipSubRouter
router
              , Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
              , PeerScoreParams
-> PeerId
-> PeerState
-> Map ByteString (Set PeerId)
-> UTCTime
-> Double
computeScore (GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router) PeerId
pid PeerState
ps Map ByteString (Set PeerId)
ipMap UTCTime
now Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
>= Double
0 ]
  chosen <- sampleIO
    (min (paramPrunePeers (gsParams router)) (length candidates)) candidates
  pure [ PeerExchangeInfo (peerIdBytes pid) (Map.lookup pid signedRecords)
       | pid <- chosen ]

-- Helper: construct a GRAFT RPC
graftRPC :: Topic -> RPC
graftRPC :: Topic -> RPC
graftRPC Topic
topic = RPC
emptyRPC
  { rpcControl = Just emptyControlMessage { ctrlGraft = [Graft topic] } }

-- Helper: wrap a PRUNE in an RPC
pruneRPC :: Prune -> RPC
pruneRPC :: Prune -> RPC
pruneRPC Prune
prn = RPC
emptyRPC
  { rpcControl = Just emptyControlMessage { ctrlPrune = [prn] } }

-- Helpers: message construction

mkUnsignedMessage :: Topic -> ByteString -> PubSubMessage
mkUnsignedMessage :: Topic -> ByteString -> PubSubMessage
mkUnsignedMessage Topic
topic ByteString
payload = PubSubMessage
  { msgFrom :: Maybe ByteString
msgFrom      = Maybe ByteString
forall a. Maybe a
Nothing
  , msgData :: ByteString
msgData      = ByteString
payload
  , msgSeqNo :: Maybe ByteString
msgSeqNo     = Maybe ByteString
forall a. Maybe a
Nothing
  , msgTopic :: Topic
msgTopic     = Topic
topic
  , msgSignature :: Maybe ByteString
msgSignature = Maybe ByteString
forall a. Maybe a
Nothing
  , msgKey :: Maybe ByteString
msgKey       = Maybe ByteString
forall a. Maybe a
Nothing
  }

mkSignedMessage :: GossipSubRouter -> Topic -> ByteString -> KeyPair -> IO PubSubMessage
mkSignedMessage :: GossipSubRouter
-> Topic -> ByteString -> KeyPair -> IO PubSubMessage
mkSignedMessage GossipSubRouter
router Topic
topic ByteString
payload KeyPair
kp = do
  seqno <- Int -> IO ByteString
forall byteArray. ByteArray byteArray => Int -> IO byteArray
forall (m :: * -> *) byteArray.
(MonadRandom m, ByteArray byteArray) =>
Int -> m byteArray
getRandomBytes Int
8 :: IO ByteString
  let from = PeerId -> ByteString
peerIdBytes (GossipSubRouter -> PeerId
gsLocalPeerId GossipSubRouter
router)
      pubKeyBytes = PublicKey -> ByteString
encodePublicKey (KeyPair -> PublicKey
kpPublic KeyPair
kp)
      -- Build unsigned message for signing
      unsigned = PubSubMessage
        { msgFrom :: Maybe ByteString
msgFrom      = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
from
        , msgData :: ByteString
msgData      = ByteString
payload
        , msgSeqNo :: Maybe ByteString
msgSeqNo     = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
seqno
        , msgTopic :: Topic
msgTopic     = Topic
topic
        , msgSignature :: Maybe ByteString
msgSignature = Maybe ByteString
forall a. Maybe a
Nothing
        , msgKey :: Maybe ByteString
msgKey       = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
pubKeyBytes
        }
  case sign (kpPrivate kp) (signingBytes unsigned) of
    -- Publishing unsigned under StrictSign would emit a message every
    -- compliant receiver must drop, so fail loudly instead.
    Left String
err  -> GossipSubError -> IO PubSubMessage
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> GossipSubError
SigningFailed String
err)
    Right ByteString
sig -> PubSubMessage -> IO PubSubMessage
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure PubSubMessage
unsigned { msgSignature = Just sig }