erigon-pulse/cl/phase1/network/gossip_manager.go

package network

import (
	"context"
	"runtime"

	"github.com/VictoriaMetrics/metrics"
	"github.com/ledgerwatch/erigon/cl/cltypes/solid"
	"github.com/ledgerwatch/erigon/cl/freezer"
	"github.com/ledgerwatch/erigon/cl/phase1/forkchoice"

	libcommon "github.com/ledgerwatch/erigon-lib/common"
	"github.com/ledgerwatch/erigon-lib/common/dbg"
	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentinel"
	"github.com/ledgerwatch/erigon-lib/types/ssz"
	"github.com/ledgerwatch/erigon/cl/clparams"
	"github.com/ledgerwatch/erigon/cl/cltypes"
	"github.com/ledgerwatch/erigon/cl/utils"
	"github.com/ledgerwatch/erigon/common"
	"github.com/ledgerwatch/log/v3"
)

// Gossip manager is sending all messages to fork choice or others
type GossipManager struct {
	ctx context.Context

	recorder   freezer.Freezer
	forkChoice *forkchoice.ForkChoiceStore
	sentinel   sentinel.SentinelClient
	// configs
	beaconConfig  *clparams.BeaconChainConfig
	genesisConfig *clparams.GenesisConfig
}

func NewGossipReceiver(ctx context.Context, s sentinel.SentinelClient, forkChoice *forkchoice.ForkChoiceStore,
	beaconConfig *clparams.BeaconChainConfig, genesisConfig *clparams.GenesisConfig, recorder freezer.Freezer) *GossipManager {
	return &GossipManager{
		sentinel:      s,
		forkChoice:    forkChoice,
		ctx:           ctx,
		beaconConfig:  beaconConfig,
		genesisConfig: genesisConfig,
		recorder:      recorder,
	}
}

func (g *GossipManager) onRecv(data *sentinel.GossipData, l log.Ctx) error {

	currentEpoch := utils.GetCurrentEpoch(g.genesisConfig.GenesisTime, g.beaconConfig.SecondsPerSlot, g.beaconConfig.SlotsPerEpoch)
	version := g.beaconConfig.GetCurrentStateVersion(currentEpoch)

	// Depending on the type of the received data, we create an instance of a specific type that implements the ObjectSSZ interface,
	// then attempts to deserialize the received data into it.
	// If the deserialization fails, an error is logged and the loop returns to the next iteration.
	// If the deserialization is successful, the object is set to the deserialized value and the loop returns to the next iteration.
	var object ssz.Unmarshaler
	switch data.Type {
	case sentinel.GossipType_BeaconBlockGossipType:
		object = &cltypes.SignedBeaconBlock{}
		if err := object.DecodeSSZ(common.CopyBytes(data.Data), int(version)); err != nil {
			g.sentinel.BanPeer(g.ctx, data.Peer)
			l["at"] = "decoding block"
			return err
		}
		block := object.(*cltypes.SignedBeaconBlock)
		l["slot"] = block.Block.Slot
		currentSlotByTime := utils.GetCurrentSlot(g.genesisConfig.GenesisTime, g.beaconConfig.SecondsPerSlot)
		maxGossipSlotThreshold := uint64(4)
		// Skip if slot is too far behind.
		if block.Block.Slot+maxGossipSlotThreshold < currentSlotByTime {
			return nil
		}
		if block.Block.Slot == currentSlotByTime {
			if _, err := g.sentinel.PublishGossip(g.ctx, data); err != nil {
				log.Debug("failed publish gossip", "err", err)
			}
		}

		count, err := g.sentinel.GetPeers(g.ctx, &sentinel.EmptyMessage{})
		if err != nil {
			l["at"] = "sentinel peer count"
			return err
		}
		var m runtime.MemStats
		dbg.ReadMemStats(&m)

		log.Debug("Received block via gossip",
			"peers", count.Amount,
			"slot", block.Block.Slot,
			"alloc/sys", libcommon.ByteCount(m.Alloc)+"/"+libcommon.ByteCount(m.Sys),
			"numGC", m.NumGC,
		)

		if err := freezer.PutObjectSSZIntoFreezer("signedBeaconBlock", "caplin_core", block.Block.Slot, block, g.recorder); err != nil {
			return err
		}

		peers := metrics.GetOrCreateGauge("caplin_peer_count", func() float64 {
			return float64(count.Amount)
		})

		peers.Get()

		if err := g.forkChoice.OnBlock(block, true, true); err != nil {
			// if we are within a quarter of an epoch within chain tip we ban it
			if currentSlotByTime < g.forkChoice.HighestSeen()+(g.beaconConfig.SlotsPerEpoch/4) {
				g.sentinel.BanPeer(g.ctx, data.Peer)
			}
			l["at"] = "block process"
			return err
		}
		block.Block.Body.Attestations.Range(func(idx int, a *solid.Attestation, total int) bool {
			if err = g.forkChoice.OnAttestation(a, true); err != nil {
				return false
			}
			return true
		})
		if err != nil {
			l["at"] = "attestation process"
			return err
		}
		// Now check the head
		headRoot, headSlot, err := g.forkChoice.GetHead()
		if err != nil {
			l["slot"] = block.Block.Slot
			l["at"] = "fetch head data"
			return err
		}
		// Do forkchoice if possible
		if g.forkChoice.Engine() != nil {
			finalizedCheckpoint := g.forkChoice.FinalizedCheckpoint()
			log.Info("Caplin is sending forkchoice")
			// Run forkchoice
			if err := g.forkChoice.Engine().ForkChoiceUpdate(
				g.forkChoice.GetEth1Hash(finalizedCheckpoint.BlockRoot()),
				g.forkChoice.GetEth1Hash(headRoot),
			); err != nil {
				log.Warn("Could not set forkchoice", "err", err)
				l["at"] = "sending forkchoice"
				return err
			}
		}
		// Log final result
		log.Debug("New gossip block imported",
			"slot", block.Block.Slot,
			"head", headSlot,
			"headRoot", headRoot,
		)
	case sentinel.GossipType_VoluntaryExitGossipType:
		object = &cltypes.SignedVoluntaryExit{}
		if err := object.DecodeSSZ(data.Data, int(version)); err != nil {
			g.sentinel.BanPeer(g.ctx, data.Peer)
			l["at"] = "decode exit"
			return err
		}
	case sentinel.GossipType_ProposerSlashingGossipType:
		object = &cltypes.ProposerSlashing{}
		if err := object.DecodeSSZ(data.Data, int(version)); err != nil {
			l["at"] = "decode proposer slash"
			g.sentinel.BanPeer(g.ctx, data.Peer)
			return err
		}
	case sentinel.GossipType_AttesterSlashingGossipType:
		object = &cltypes.AttesterSlashing{}
		if err := object.DecodeSSZ(data.Data, int(version)); err != nil {
			l["at"] = "decode attester slash"
			g.sentinel.BanPeer(g.ctx, data.Peer)
			return err
		}
		if err := g.forkChoice.OnAttesterSlashing(object.(*cltypes.AttesterSlashing)); err != nil {
			l["at"] = "on attester slash"
			return err
		}
	case sentinel.GossipType_AggregateAndProofGossipType:
		object = &cltypes.SignedAggregateAndProof{}
		if err := object.DecodeSSZ(data.Data, int(version)); err != nil {
			l["at"] = "decoding proof"
			g.sentinel.BanPeer(g.ctx, data.Peer)
			return err
		}
	}
	return nil
}

func (g *GossipManager) Start() {
	subscription, err := g.sentinel.SubscribeGossip(g.ctx, &sentinel.EmptyMessage{})
	if err != nil {
		return
	}

	l := log.Ctx{}
	for {
		data, err := subscription.Recv()
		if err != nil {
			log.Warn("[Beacon Gossip] Fatal error receiving gossip", "err", err)
			break
		}
		for k := range l {
			delete(l, k)
		}
		err = g.onRecv(data, l)
		if err != nil {
			l["err"] = err
			log.Debug("[Beacon Gossip] Recoverable Error", l)
		}
	}
}