-- | GossipSub heartbeat procedure (specs/pubsub/gossipsub).
--
-- The heartbeat runs periodically and performs:
-- 1. Mesh maintenance: prune negative-score, fill undersubscribed, trim oversubscribed
-- 2. Fanout maintenance: expire old, fill undersubscribed
-- 3. Gossip emission: send IHAVE to non-mesh peers, rotate cache
-- 4. Score decay: decay all counters for all peers
-- 5. Seen cache cleanup: remove expired entries
-- 6. Heartbeat counter increment (for opportunistic graft timing)
module LibP2P.Protocol.GossipSub.Heartbeat
  ( heartbeatOnce
  , runHeartbeat
  ) where

import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (Async, async)
import Control.Concurrent.STM
import Control.Monad (forM_, unless, when)
import Data.List (sortOn)
import qualified Data.Map.Strict as Map
import qualified Data.Set as Set
import Data.Ord (Down (..))
import Data.Time (UTCTime, addUTCTime, diffUTCTime)
import Data.Word (Word64)
import List.Shuffle (sampleIO)
import LibP2P.Crypto.PeerId (PeerId)
import LibP2P.Protocol.GossipSub.Types
import LibP2P.Protocol.GossipSub.MessageCache (cacheGetGossipIds, cacheShift)
import LibP2P.Protocol.GossipSub.Router (buildPrune)
import LibP2P.Protocol.GossipSub.Score
  ( computeScore
  , decayPeerCounters
  , addP7Penalty
  , refreshMeshTime
  , markPeerInMesh
  , unmarkPeerInMesh
  )

-- | Run a single heartbeat cycle. Exported for testing.
heartbeatOnce :: GossipSubRouter -> IO ()
heartbeatOnce :: GossipSubRouter -> IO ()
heartbeatOnce GossipSubRouter
router = do
  GossipSubRouter -> IO ()
meshMaintenance GossipSubRouter
router
  GossipSubRouter -> IO ()
fanoutMaintenance GossipSubRouter
router
  GossipSubRouter -> IO ()
emitGossip GossipSubRouter
router
  GossipSubRouter -> IO ()
expireIWantPromises GossipSubRouter
router
  GossipSubRouter -> IO ()
decayAllScores GossipSubRouter
router
  GossipSubRouter -> IO ()
cleanSeenCache GossipSubRouter
router
  GossipSubRouter -> IO ()
resetGossipBudgets GossipSubRouter
router
  -- Increment heartbeat counter
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar Int -> (Int -> Int) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar Int
gsHeartbeatCount GossipSubRouter
router) (Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)

-- | Start the heartbeat background thread.
runHeartbeat :: GossipSubRouter -> IO (Async ())
runHeartbeat :: GossipSubRouter -> IO (Async ())
runHeartbeat GossipSubRouter
router = IO () -> IO (Async ())
forall a. IO a -> IO (Async a)
async (IO () -> IO (Async ())) -> IO () -> IO (Async ())
forall a b. (a -> b) -> a -> b
$ GossipSubRouter -> IO ()
heartbeatLoop GossipSubRouter
router

heartbeatLoop :: GossipSubRouter -> IO ()
heartbeatLoop :: GossipSubRouter -> IO ()
heartbeatLoop GossipSubRouter
router = do
  let intervalUs :: Int
intervalUs = NominalDiffTime -> Int
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (GossipSubParams -> NominalDiffTime
paramHeartbeatInterval (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router) NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Num a => a -> a -> a
* NominalDiffTime
1000000) :: Int
  Int -> IO ()
threadDelay Int
intervalUs
  GossipSubRouter -> IO ()
heartbeatOnce GossipSubRouter
router
  GossipSubRouter -> IO ()
heartbeatLoop GossipSubRouter
router

-- Mesh maintenance

meshMaintenance :: GossipSubRouter -> IO ()
meshMaintenance :: GossipSubRouter -> IO ()
meshMaintenance GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  meshMap <- readTVarIO (gsMesh router)
  subs <- readTVarIO (gsSubscriptions router)
  -- Maintain every subscribed topic, not just topics with a mesh entry:
  -- a topic joined with no peers must be filled once peers appear (#155).
  let topics = Set Topic -> Set Topic -> Set Topic
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set Topic
subs (Map Topic (Set PeerId) -> Set Topic
forall k a. Map k a -> Set k
Map.keysSet Map Topic (Set PeerId)
meshMap)
  forM_ (Set.toList topics) $ \Topic
topic -> do
    let meshPeers :: Set PeerId
meshPeers = 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
    -- Step 1: Remove negative-score peers
    remaining <- GossipSubRouter
-> Topic -> Set PeerId -> UTCTime -> IO (Set PeerId)
pruneNegativeScore GossipSubRouter
router Topic
topic Set PeerId
meshPeers UTCTime
now
    -- Step 2: Fill if undersubscribed (< D_lo)
    filled <- fillUndersubscribed router topic remaining now
    -- Step 3: Trim if oversubscribed (> D_hi)
    trimOversubscribed router topic filled

-- | Remove peers with negative score from mesh, send PRUNE. Direct
-- peers are exempt: explicit peering agreements are never scored out
-- (gossipsub-v1.1.md; they should never be mesh members to begin with).
pruneNegativeScore :: GossipSubRouter -> Topic -> Set.Set PeerId -> UTCTime -> IO (Set.Set PeerId)
pruneNegativeScore :: GossipSubRouter
-> Topic -> Set PeerId -> UTCTime -> IO (Set PeerId)
pruneNegativeScore GossipSubRouter
router Topic
topic Set PeerId
meshPeers UTCTime
now = do
  let scoreParams :: PeerScoreParams
scoreParams = GossipSubRouter -> PeerScoreParams
gsScoreParams GossipSubRouter
router
      direct :: Set PeerId
direct = GossipSubParams -> Set PeerId
paramDirectPeers (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
  ipMap <- TVar (Map ByteString (Set PeerId))
-> IO (Map ByteString (Set PeerId))
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map ByteString (Set PeerId))
gsIPPeerCount GossipSubRouter
router)
  peers <- readTVarIO (gsPeers router)
  let negatives = (PeerId -> Bool) -> Set PeerId -> Set PeerId
forall a. (a -> Bool) -> Set a -> Set a
Set.filter (\PeerId
pid ->
        Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
direct) Bool -> Bool -> Bool
&&
        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
          Maybe PeerState
Nothing -> Bool
False
          Just PeerState
ps -> PeerScoreParams
-> PeerId
-> PeerState
-> Map ByteString (Set PeerId)
-> UTCTime
-> Double
computeScore PeerScoreParams
scoreParams PeerId
pid PeerState
ps Map ByteString (Set PeerId)
ipMap UTCTime
now Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
0
        ) Set PeerId
meshPeers
  -- Send PRUNE to negative-score peers (no PX for negative-score peers)
  forM_ (Set.toList negatives) $ \PeerId
pid -> do
    let backoffSecs :: Word64
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
    prn <- GossipSubRouter -> PeerId -> Topic -> Bool -> Word64 -> IO Prune
buildPrune GossipSubRouter
router PeerId
pid Topic
topic Bool
False Word64
backoffSecs
    gsSendRPC router pid emptyRPC
      { rpcControl = Just emptyControlMessage { ctrlPrune = [prn] } }
    -- Start backoff and stop the P1 mesh clock
    atomically $ do
      modifyTVar' (gsBackoff router) $
        Map.insert (pid, topic) (addUTCTime (paramPruneBackoff (gsParams router)) now)
      modifyTVar' (gsPeers router) $
        Map.adjust (unmarkPeerInMesh topic) pid
  -- Update mesh
  let remaining = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.difference Set PeerId
meshPeers Set PeerId
negatives
  atomically $ modifyTVar' (gsMesh router) $
    Map.insert topic remaining
  pure remaining

-- | Fill mesh if below D_lo with eligible peers (non-negative score, no backoff).
fillUndersubscribed :: GossipSubRouter -> Topic -> Set.Set PeerId -> UTCTime -> IO (Set.Set PeerId)
fillUndersubscribed :: GossipSubRouter
-> Topic -> Set PeerId -> UTCTime -> IO (Set PeerId)
fillUndersubscribed GossipSubRouter
router Topic
topic Set PeerId
meshPeers UTCTime
now = do
  let params :: GossipSubParams
params = GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router
      dlo :: Int
dlo = GossipSubParams -> Int
paramDlo GossipSubParams
params
      d :: Int
d   = GossipSubParams -> Int
paramD GossipSubParams
params
  if Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
meshPeers Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
dlo
    then Set PeerId -> IO (Set PeerId)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Set PeerId
meshPeers
    else do
      -- Find eligible peers: subscribed to topic, not in mesh, not in backoff, score >= 0
      peersMap <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
      backoffMap <- readTVarIO (gsBackoff router)
      ipMap <- readTVarIO (gsIPPeerCount router)
      let direct = GossipSubParams -> Set PeerId
paramDirectPeers GossipSubParams
params
          eligible = [ PeerId
pid | (PeerId
pid, PeerState
ps) <- Map PeerId PeerState -> [(PeerId, PeerState)]
forall k a. Map k a -> [(k, a)]
Map.toList Map PeerId PeerState
peersMap
                           , Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
                           , Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
meshPeers)
                           , Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
direct)  -- never graft direct peers
                           , PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerProtocol
FloodSubPeer  -- floodsub peers have no mesh (#157)
                           , Bool -> Bool
not (Map (PeerId, Topic) UTCTime -> PeerId -> Topic -> UTCTime -> Bool
isInBackoff Map (PeerId, Topic) UTCTime
backoffMap PeerId
pid Topic
topic UTCTime
now)
                           , 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
                           ]
      let needed = Int
d Int -> Int -> Int
forall a. Num a => a -> a -> a
- Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
meshPeers
      selected <- sampleIO (min needed (length eligible)) eligible
      -- Send GRAFT and add to mesh
      forM_ selected $ \PeerId
pid ->
        GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
emptyRPC
          { rpcControl = Just emptyControlMessage { ctrlGraft = [Graft topic] } }
      let newMesh = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set PeerId
meshPeers ([PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
selected)
      atomically $ do
        modifyTVar' (gsMesh router) $ Map.insert topic newMesh
        -- Start the P1 mesh clock for the newly grafted peers
        modifyTVar' (gsPeers router) $ \Map PeerId PeerState
pm ->
          (Map PeerId PeerState -> PeerId -> Map PeerId PeerState)
-> Map PeerId PeerState -> [PeerId] -> Map PeerId PeerState
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
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 [PeerId]
selected
      pure newMesh

-- | Trim mesh if above D_hi down to D peers, send PRUNE with PX.
--
-- Per gossipsub-v1.1.md mesh maintenance: keep the best D_score peers by
-- score, select the rest at random, under the constraint that at least
-- D_out of the kept peers are outbound connections.
trimOversubscribed :: GossipSubRouter -> Topic -> Set.Set PeerId -> IO ()
trimOversubscribed :: GossipSubRouter -> Topic -> Set PeerId -> IO ()
trimOversubscribed GossipSubRouter
router Topic
topic Set PeerId
meshPeers = do
  let params :: GossipSubParams
params = GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router
      dhi :: Int
dhi    = GossipSubParams -> Int
paramDhi GossipSubParams
params
      d :: Int
d      = GossipSubParams -> Int
paramD GossipSubParams
params
      dscore :: Int
dscore = GossipSubParams -> Int
paramDscore GossipSubParams
params
      dout :: Int
dout   = GossipSubParams -> Int
paramDout GossipSubParams
params
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
meshPeers Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
dhi) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
    peersMap <- readTVarIO (gsPeers router)
    ipMap <- readTVarIO (gsIPPeerCount router)
    let scoreOf 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
peersMap of
          Maybe PeerState
Nothing -> Double
0
          Just 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
        isOutbound PeerId
pid = Bool -> (PeerState -> Bool) -> Maybe PeerState -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False PeerState -> Bool
psIsOutbound (PeerId -> Map PeerId PeerState -> Maybe PeerState
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId PeerState
peersMap)
        ranked = (PeerId -> Down Double) -> [PeerId] -> [PeerId]
forall b a. Ord b => (a -> b) -> [a] -> [a]
sortOn (Double -> Down Double
forall a. a -> Down a
Down (Double -> Down Double)
-> (PeerId -> Double) -> PeerId -> Down Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PeerId -> Double
scoreOf) (Set PeerId -> [PeerId]
forall a. Set a -> [a]
Set.toList Set PeerId
meshPeers)
        (best, rest) = splitAt (min dscore d) ranked
    restKept <- sampleIO (max 0 (d - length best)) rest
    -- keptList is ordered best-first: the score-retained peers, then the
    -- random selection. D_out swaps drop from the tail first.
    let keptList = [PeerId]
best [PeerId] -> [PeerId] -> [PeerId]
forall a. [a] -> [a] -> [a]
++ [PeerId]
restKept
        kept0 = [PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
keptList
        outDeficit = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 (Int
dout Int -> Int -> Int
forall a. Num a => a -> a -> a
- [PeerId] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ((PeerId -> Bool) -> [PeerId] -> [PeerId]
forall a. (a -> Bool) -> [a] -> [a]
filter PeerId -> Bool
isOutbound [PeerId]
keptList))
        swapIn = Int -> [PeerId] -> [PeerId]
forall a. Int -> [a] -> [a]
take Int
outDeficit
          ((PeerId -> Bool) -> [PeerId] -> [PeerId]
forall a. (a -> Bool) -> [a] -> [a]
filter (\PeerId
p -> PeerId -> Bool
isOutbound PeerId
p Bool -> Bool -> Bool
&& Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
p Set PeerId
kept0)) [PeerId]
ranked)
        swapOut = Int -> [PeerId] -> [PeerId]
forall a. Int -> [a] -> [a]
take ([PeerId] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [PeerId]
swapIn)
          ((PeerId -> Bool) -> [PeerId] -> [PeerId]
forall a. (a -> Bool) -> [a] -> [a]
filter (Bool -> Bool
not (Bool -> Bool) -> (PeerId -> Bool) -> PeerId -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PeerId -> Bool
isOutbound) ([PeerId] -> [PeerId]
forall a. [a] -> [a]
reverse [PeerId]
keptList))
        keptSet = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.union
          (Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.difference Set PeerId
kept0 ([PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
swapOut))
          ([PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
swapIn)
        toRemove = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.difference Set PeerId
meshPeers Set PeerId
keptSet
    -- Send PRUNE with peer exchange to removed peers
    forM_ (Set.toList toRemove) $ \PeerId
pid -> do
      let backoffSecs :: Word64
backoffSecs = NominalDiffTime -> Word64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (GossipSubParams -> NominalDiffTime
paramPruneBackoff GossipSubParams
params) :: Word64
      prn <- GossipSubRouter -> PeerId -> Topic -> Bool -> Word64 -> IO Prune
buildPrune GossipSubRouter
router PeerId
pid Topic
topic Bool
True Word64
backoffSecs
      gsSendRPC router pid emptyRPC
        { rpcControl = Just emptyControlMessage { ctrlPrune = [prn] } }
      atomically $ do
        modifyTVar' (gsBackoff router) $
          Map.insert (pid, topic) (addUTCTime (paramPruneBackoff params) now)
        modifyTVar' (gsPeers router) $
          Map.adjust (unmarkPeerInMesh topic) pid
    -- Update mesh
    atomically $ modifyTVar' (gsMesh router) $
      Map.insert topic keptSet

-- Fanout maintenance

fanoutMaintenance :: GossipSubRouter -> IO ()
fanoutMaintenance :: GossipSubRouter -> IO ()
fanoutMaintenance GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  let ttl = GossipSubParams -> NominalDiffTime
paramFanoutTTL (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
  fanoutMap <- readTVarIO (gsFanout router)
  fanoutPubMap <- readTVarIO (gsFanoutPub router)
  forM_ (Map.toList fanoutMap) $ \(Topic
topic, Set PeerId
fanoutPeers) -> do
    let lastPub :: UTCTime
lastPub = UTCTime -> Topic -> Map Topic UTCTime -> UTCTime
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault UTCTime
now Topic
topic Map Topic UTCTime
fanoutPubMap
    if UTCTime -> UTCTime -> NominalDiffTime
diffUTCTime UTCTime
now UTCTime
lastPub NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
> NominalDiffTime
ttl
      then -- Expire fanout entry
        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))
gsFanout GossipSubRouter
router) (Topic -> Map Topic (Set PeerId) -> Map Topic (Set PeerId)
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete Topic
topic)
          TVar (Map Topic UTCTime)
-> (Map Topic UTCTime -> Map Topic UTCTime) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubRouter -> TVar (Map Topic UTCTime)
gsFanoutPub GossipSubRouter
router) (Topic -> Map Topic UTCTime -> Map Topic UTCTime
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete Topic
topic)
      else do
        -- Fill if below D
        let d :: Int
d = GossipSubParams -> Int
paramD (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
        Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
fanoutPeers Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
d) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          peersMap <- TVar (Map PeerId PeerState) -> IO (Map PeerId PeerState)
forall a. TVar a -> IO a
readTVarIO (GossipSubRouter -> TVar (Map PeerId PeerState)
gsPeers GossipSubRouter
router)
          let direct = GossipSubParams -> Set PeerId
paramDirectPeers (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
              eligible = [ PeerId
pid | (PeerId
pid, PeerState
ps) <- Map PeerId PeerState -> [(PeerId, PeerState)]
forall k a. Map k a -> [(k, a)]
Map.toList Map PeerId PeerState
peersMap
                               , Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
                               , Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
fanoutPeers)
                               , Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
direct)
                               , PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerProtocol
FloodSubPeer  -- flooded, never fanout members (#157)
                               ]
          let needed = Int
d Int -> Int -> Int
forall a. Num a => a -> a -> a
- Set PeerId -> Int
forall a. Set a -> Int
Set.size Set PeerId
fanoutPeers
          selected <- sampleIO (min needed (length eligible)) eligible
          let newFanout = Set PeerId -> Set PeerId -> Set PeerId
forall a. Ord a => Set a -> Set a -> Set a
Set.union Set PeerId
fanoutPeers ([PeerId] -> Set PeerId
forall a. Ord a => [a] -> Set a
Set.fromList [PeerId]
selected)
          atomically $ modifyTVar' (gsFanout router) $
            Map.insert topic newFanout

-- Gossip emission

emitGossip :: GossipSubRouter -> IO ()
emitGossip :: GossipSubRouter -> IO ()
emitGossip GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  meshMap <- readTVarIO (gsMesh router)
  fanoutMap <- readTVarIO (gsFanout router)
  subs <- readTVarIO (gsSubscriptions router)
  cache <- readTVarIO (gsMessageCache router)
  peersMap <- readTVarIO (gsPeers router)
  ipMap <- readTVarIO (gsIPPeerCount router)
  let params = GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router
      -- gossipsub-v1.0.md heartbeat: gossip covers "each topic in
      -- mesh+fanout", so topics we publish to without subscribing
      -- also emit IHAVE (#155).
      topics = [Set Topic] -> Set Topic
forall (f :: * -> *) a. (Foldable f, Ord a) => f (Set a) -> Set a
Set.unions
        [ Set Topic
subs, Map Topic (Set PeerId) -> Set Topic
forall k a. Map k a -> Set k
Map.keysSet Map Topic (Set PeerId)
meshMap, Map Topic (Set PeerId) -> Set Topic
forall k a. Map k a -> Set k
Map.keysSet Map Topic (Set PeerId)
fanoutMap ]

  -- For each topic in mesh+fanout, send IHAVE to non-mesh peers.
  -- The advertised id list is capped at paramMaxIHaveLength (#157).
  forM_ (Set.toList topics) $ \Topic
topic -> do
    let gossipIds :: [ByteString]
gossipIds = Int -> [ByteString] -> [ByteString]
forall a. Int -> [a] -> [a]
take (GossipSubParams -> Int
paramMaxIHaveLength GossipSubParams
params) (Topic -> MessageCache -> [ByteString]
cacheGetGossipIds Topic
topic MessageCache
cache)
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([ByteString] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [ByteString]
gossipIds) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      let meshPeers :: Set PeerId
meshPeers = 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
          -- Eligible: subscribed to topic, not in mesh, and scoring at or
          -- above the gossip threshold — no gossip is emitted towards
          -- peers below it (gossipsub-v1.1.md gossip threshold). Floodsub
          -- peers never receive control messages, IHAVE included (#157).
          gossipThreshold :: Double
gossipThreshold = ScoreThresholds -> Double
stGossipThreshold (GossipSubRouter -> ScoreThresholds
gsThresholds GossipSubRouter
router)
          nonMeshPeers :: [PeerId]
nonMeshPeers = [ PeerId
pid | (PeerId
pid, PeerState
ps) <- Map PeerId PeerState -> [(PeerId, PeerState)]
forall k a. Map k a -> [(k, a)]
Map.toList Map PeerId PeerState
peersMap
                               , Topic -> Set Topic -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Topic
topic (PeerState -> Set Topic
psTopics PeerState
ps)
                               , Bool -> Bool
not (PeerId -> Set PeerId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member PeerId
pid Set PeerId
meshPeers)
                               , PeerState -> PeerProtocol
psProtocol PeerState
ps PeerProtocol -> PeerProtocol -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerProtocol
FloodSubPeer
                               , 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
gossipThreshold
                               ]
          -- Select max(D_lazy, |eligible| * gossipFactor) targets
          dlazy :: Int
dlazy = GossipSubParams -> Int
paramDlazy GossipSubParams
params
          factor :: Double
factor = GossipSubParams -> Double
paramGossipFactor GossipSubParams
params
          targetCount :: Int
targetCount = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
dlazy (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (Double
factor Double -> Double -> Double
forall a. Num a => a -> a -> a
* Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral ([PeerId] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [PeerId]
nonMeshPeers)))
      targets <- Int -> [PeerId] -> IO [PeerId]
forall (m :: * -> *) a. MonadIO m => Int -> [a] -> m [a]
sampleIO (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
targetCount ([PeerId] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [PeerId]
nonMeshPeers)) [PeerId]
nonMeshPeers
      forM_ targets $ \PeerId
pid ->
        GossipSubRouter -> PeerId -> RPC -> IO ()
gsSendRPC GossipSubRouter
router PeerId
pid RPC
emptyRPC
          { rpcControl = Just emptyControlMessage
              { ctrlIHave = [IHave topic gossipIds] }
          }

  -- Rotate cache
  atomically $ modifyTVar' (gsMessageCache router) cacheShift

-- IWANT promise expiry (P7)

-- | Penalise peers whose IWANT promises expired without delivery
-- (gossipsub-v1.1.md: a peer that advertises via IHAVE but never sends
-- the requested message commits a behavioural violation).
expireIWantPromises :: GossipSubRouter -> IO ()
expireIWantPromises :: GossipSubRouter -> IO ()
expireIWantPromises GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  broken <- atomically $ do
    promises <- readTVar (gsIWantPromises router)
    let (expired, live) = Map.partition (<= now) promises
    writeTVar (gsIWantPromises router) live
    pure (map fst (Map.keys expired))
  atomically $ modifyTVar' (gsPeers router) $ \Map PeerId PeerState
pm ->
    (Map PeerId PeerState -> PeerId -> Map PeerId PeerState)
-> Map PeerId PeerState -> [PeerId] -> Map PeerId PeerState
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
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 PeerState -> PeerState
addP7Penalty PeerId
pid Map PeerId PeerState
m) Map PeerId PeerState
pm [PeerId]
broken

-- IHAVE/IWANT budget reset

-- | Reset the per-peer IHAVE/IWANT flood-protection budgets; the caps in
-- Router.handleIHave / Router.handleIWant apply per heartbeat (#157).
resetGossipBudgets :: GossipSubRouter -> IO ()
resetGossipBudgets :: GossipSubRouter -> IO ()
resetGossipBudgets GossipSubRouter
router = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  TVar (Map PeerId Int) -> Map PeerId Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (GossipSubRouter -> TVar (Map PeerId Int)
gsIHaveCounts GossipSubRouter
router) Map PeerId Int
forall k a. Map k a
Map.empty
  TVar (Map PeerId Int) -> Map PeerId Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (GossipSubRouter -> TVar (Map PeerId Int)
gsIAskedCounts GossipSubRouter
router) Map PeerId Int
forall k a. Map k a
Map.empty
  TVar (Map PeerId Int) -> Map PeerId Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (GossipSubRouter -> TVar (Map PeerId Int)
gsIWantServed GossipSubRouter
router) Map PeerId Int
forall k a. Map k a
Map.empty

-- Score decay

-- | Refresh accrued mesh time (P1 input) and decay all scoring counters.
decayAllScores :: GossipSubRouter -> IO ()
decayAllScores :: GossipSubRouter -> IO ()
decayAllScores GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  atomically $ modifyTVar' (gsPeers router) $
    Map.map (decayPeerCounters (gsScoreParams router) . refreshMeshTime now)

-- Seen cache cleanup

cleanSeenCache :: GossipSubRouter -> IO ()
cleanSeenCache :: GossipSubRouter -> IO ()
cleanSeenCache GossipSubRouter
router = do
  now <- GossipSubRouter -> IO UTCTime
gsGetTime GossipSubRouter
router
  let ttl = GossipSubParams -> NominalDiffTime
paramSeenTTL (GossipSubRouter -> GossipSubParams
gsParams GossipSubRouter
router)
  atomically $ modifyTVar' (gsSeen router) $
    Map.filter (\UTCTime
ts -> UTCTime -> UTCTime -> NominalDiffTime
diffUTCTime UTCTime
now UTCTime
ts NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
<= NominalDiffTime
ttl)

-- Helpers

isInBackoff :: Map.Map (PeerId, Topic) UTCTime -> PeerId -> Topic -> UTCTime -> Bool
isInBackoff :: Map (PeerId, Topic) UTCTime -> PeerId -> Topic -> UTCTime -> Bool
isInBackoff Map (PeerId, Topic) UTCTime
backoffMap PeerId
pid Topic
topic UTCTime
now =
  case (PeerId, Topic) -> Map (PeerId, Topic) UTCTime -> Maybe UTCTime
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup (PeerId
pid, 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