shutter: add eon tracker (#13647)

- part 1 of #13384 (adding
eon based validation of decryption keys)
- part 2 will be to use the eon info to validate the keys
- takes a chunk of changes out of bigger branch
#13553

eth/backend.go

@@ -744,7 +744,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 		)
 		contractBackend := contracts.NewDirectBackend(ethApi)
 		secondaryTxnProvider := backend.txPool
-		backend.shutterPool = shutter.NewPool(logger, config.Shutter, secondaryTxnProvider, contractBackend)
+		backend.shutterPool = shutter.NewPool(logger, config.Shutter, secondaryTxnProvider, contractBackend, backend.stateDiffClient)
 		txnProvider = backend.shutterPool
 	}
 

txnprovider/shutter/block_listener.go

@@ -0,0 +1,75 @@
+// Copyright 2025 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package shutter
+
+import (
+	"context"
+
+	"github.com/erigontech/erigon-lib/event"
+	"github.com/erigontech/erigon-lib/gointerfaces/remoteproto"
+	"github.com/erigontech/erigon-lib/log/v3"
+)
+
+type BlockEvent struct {
+	BlockNum uint64
+	Unwind   bool
+}
+
+type BlockListener struct {
+	logger             log.Logger
+	stateChangesClient stateChangesClient
+	events             *event.Observers[BlockEvent]
+}
+
+func NewBlockListener(logger log.Logger, stateChangesClient stateChangesClient) BlockListener {
+	return BlockListener{
+		logger:             logger,
+		stateChangesClient: stateChangesClient,
+		events:             event.NewObservers[BlockEvent](),
+	}
+}
+
+func (bl BlockListener) RegisterObserver(o event.Observer[BlockEvent]) event.UnregisterFunc {
+	return bl.events.Register(o)
+}
+
+func (bl BlockListener) Run(ctx context.Context) error {
+	bl.logger.Info("running block listener")
+
+	sub, err := bl.stateChangesClient.StateChanges(ctx, &remoteproto.StateChangeRequest{})
+	if err != nil {
+		return err
+	}
+
+	// note the changes stream is ctx-aware so Recv should terminate with err if ctx gets done
+	var batch *remoteproto.StateChangeBatch
+	for batch, err = sub.Recv(); err != nil; batch, err = sub.Recv() {
+		if batch == nil || len(batch.ChangeBatch) == 0 {
+			continue
+		}
+
+		latestChange := batch.ChangeBatch[len(batch.ChangeBatch)-1]
+		blockEvent := BlockEvent{
+			BlockNum: latestChange.BlockHeight,
+			Unwind:   latestChange.Direction == remoteproto.Direction_UNWIND,
+		}
+
+		bl.events.NotifySync(blockEvent)
+	}
+
+	return err
+}

txnprovider/shutter/config.go

@@ -1,4 +1,4 @@
-// Copyright 2024 The Erigon Authors
+// Copyright 2025 The Erigon Authors
 // This file is part of Erigon.
 //
 // Erigon is free software: you can redistribute it and/or modify
@@ -23,17 +23,21 @@ import (
 	"github.com/multiformats/go-multiaddr"
 
 	"github.com/erigontech/erigon-lib/chain/networkname"
+	"github.com/erigontech/erigon/cl/clparams"
 )
 
 type Config struct {
 	P2pConfig
 	Enabled                          bool
 	InstanceId                       uint64
+	BeaconChainGenesisTimestamp      uint64
+	SecondsPerSlot                   uint64
 	SequencerContractAddress         string
 	ValidatorRegistryContractAddress string
 	KeyBroadcastContractAddress      string
 	KeyperSetManagerContractAddress  string
 	MaxNumKeysPerMessage             uint64
+	MaxRecentEons                    int
 }
 
 type P2pConfig struct {
@@ -76,11 +80,14 @@ var (
 	chiadoConfig = Config{
 		Enabled:                          true,
 		InstanceId:                       102_000,
+		BeaconChainGenesisTimestamp:      1665396300,
+		SecondsPerSlot:                   clparams.BeaconConfigs[clparams.ChiadoNetwork].SecondsPerSlot,
 		SequencerContractAddress:         "0x2aD8E2feB0ED5b2EC8e700edB725f120576994ed",
 		ValidatorRegistryContractAddress: "0xa9289A3Dd14FEBe10611119bE81E5d35eAaC3084",
 		KeyBroadcastContractAddress:      "0x9D31865BEffcE842FBd36CDA587aDDA8bef804B7",
 		KeyperSetManagerContractAddress:  "0xC4DE9FAf4ec882b33dA0162CBE628B0D8205D0c0",
 		MaxNumKeysPerMessage:             defaultMaxNumKeysPerMessage,
+		MaxRecentEons:                    defaultMaxRecentEons,
 		P2pConfig: P2pConfig{
 			ListenPort: defaultP2PListenPort,
 			BootstrapNodes: []string{
@@ -93,11 +100,14 @@ var (
 	gnosisConfig = Config{
 		Enabled:                          true,
 		InstanceId:                       1_000,
+		BeaconChainGenesisTimestamp:      1638993340,
+		SecondsPerSlot:                   clparams.BeaconConfigs[clparams.GnosisNetwork].SecondsPerSlot,
 		SequencerContractAddress:         "0xc5C4b277277A1A8401E0F039dfC49151bA64DC2E",
 		ValidatorRegistryContractAddress: "0xefCC23E71f6bA9B22C4D28F7588141d44496A6D6",
 		KeyBroadcastContractAddress:      "0x626dB87f9a9aC47070016A50e802dd5974341301",
 		KeyperSetManagerContractAddress:  "0x7C2337f9bFce19d8970661DA50dE8DD7d3D34abb",
 		MaxNumKeysPerMessage:             defaultMaxNumKeysPerMessage,
+		MaxRecentEons:                    defaultMaxRecentEons,
 		P2pConfig: P2pConfig{
 			ListenPort: defaultP2PListenPort,
 			BootstrapNodes: []string{
@@ -111,4 +121,5 @@ var (
 const (
 	defaultP2PListenPort        = 23_102
 	defaultMaxNumKeysPerMessage = 500
+	defaultMaxRecentEons        = 100
 )

txnprovider/shutter/decryption_keys_listener.go

@@ -41,16 +41,20 @@ const (
 )
 
 type DecryptionKeysListener struct {
-	logger    log.Logger
-	config    Config
-	observers *event.Observers[*proto.DecryptionKeys]
+	logger         log.Logger
+	config         Config
+	slotCalculator SlotCalculator
+	eonTracker     EonTracker
+	observers      *event.Observers[*proto.DecryptionKeys]
 }
 
-func NewDecryptionKeysListener(logger log.Logger, config Config) DecryptionKeysListener {
+func NewDecryptionKeysListener(logger log.Logger, config Config, sc SlotCalculator, et EonTracker) DecryptionKeysListener {
 	return DecryptionKeysListener{
-		logger:    logger,
-		config:    config,
-		observers: event.NewObservers[*proto.DecryptionKeys](),
+		logger:         logger,
+		config:         config,
+		slotCalculator: sc,
+		eonTracker:     et,
+		observers:      event.NewObservers[*proto.DecryptionKeys](),
 	}
 }
 
@@ -207,7 +211,7 @@ func (dkl DecryptionKeysListener) connectBootstrapNodes(ctx context.Context, hos
 }
 
 func (dkl DecryptionKeysListener) listenLoop(ctx context.Context, pubSub *pubsub.PubSub) error {
-	topicValidator := NewDecryptionKeysP2pValidatorEx(dkl.logger, dkl.config)
+	topicValidator := NewDecryptionKeysP2pValidatorEx(dkl.logger, dkl.config, dkl.slotCalculator, dkl.eonTracker)
 	err := pubSub.RegisterTopicValidator(DecryptionKeysTopic, topicValidator)
 	if err != nil {
 		return err

txnprovider/shutter/decryption_keys_validator.go

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with Erigon. If not, see <http://www.gnu.org/licenses/>.
 
+//go:build !abigen
+
 package shutter
 
 import (
@@ -33,19 +35,29 @@ var (
 	ErrInstanceIdMismatch     = errors.New("instance id mismatch")
 	ErrMissingGnosisExtraData = errors.New("missing gnosis extra data")
 	ErrSlotTooLarge           = errors.New("slot too large")
+	ErrSlotInThePast          = errors.New("slot in the past")
+	ErrSlotInTheFuture        = errors.New("slot in the future")
 	ErrTxPointerTooLarge      = errors.New("tx pointer too large")
 	ErrEonTooLarge            = errors.New("eon too large")
+	ErrCurrentEonUnavailable  = errors.New("current eon unavailable")
+	ErrEonInThePast           = errors.New("eon in the past")
+	ErrEonInTheFuture         = errors.New("eon in the future")
 	ErrEmptyKeys              = errors.New("empty keys")
 	ErrTooManyKeys            = errors.New("too many keys")
+	ErrIgnoreMsg              = errors.New("ignoring msg")
 )
 
 type DecryptionKeysValidator struct {
-	config Config
+	config         Config
+	slotCalculator SlotCalculator
+	eonTracker     EonTracker
 }
 
-func NewDecryptionKeysValidator(config Config) DecryptionKeysValidator {
+func NewDecryptionKeysValidator(config Config, sc SlotCalculator, et EonTracker) DecryptionKeysValidator {
 	return DecryptionKeysValidator{
-		config: config,
+		config:         config,
+		slotCalculator: sc,
+		eonTracker:     et,
 	}
 }
 
@@ -54,23 +66,54 @@ func (v DecryptionKeysValidator) Validate(msg *proto.DecryptionKeys) error {
 		return fmt.Errorf("%w: %d", ErrInstanceIdMismatch, msg.InstanceId)
 	}
 
+	if err := v.validateExtraData(msg); err != nil {
+		return err
+	}
+
+	if err := v.validateEonIndex(msg); err != nil {
+		return err
+	}
+
+	return v.validateKeys(msg)
+}
+
+func (v DecryptionKeysValidator) validateExtraData(msg *proto.DecryptionKeys) error {
 	gnosisExtraData := msg.GetGnosis()
 	if gnosisExtraData == nil {
 		return ErrMissingGnosisExtraData
 	}
 
-	if gnosisExtraData.Slot > math.MaxInt64 {
-		return fmt.Errorf("%w: %d", ErrSlotTooLarge, gnosisExtraData.Slot)
+	if err := v.validateSlot(msg); err != nil {
+		return err
 	}
 
 	if gnosisExtraData.TxPointer > math.MaxInt32 {
 		return fmt.Errorf("%w: %d", ErrTxPointerTooLarge, gnosisExtraData.TxPointer)
 	}
 
-	if msg.Eon > math.MaxInt64 {
-		return fmt.Errorf("%w: %d", ErrEonTooLarge, msg.Eon)
+	return nil
+}
+
+func (v DecryptionKeysValidator) validateSlot(msg *proto.DecryptionKeys) error {
+	msgSlot := msg.GetGnosis().Slot
+	if msgSlot > math.MaxInt64 {
+		return fmt.Errorf("%w: %d", ErrSlotTooLarge, msgSlot)
+	}
+
+	currentSlot := v.slotCalculator.CalcCurrentSlot()
+	if msgSlot < currentSlot {
+		return fmt.Errorf("%w: msgSlot=%d, currentSlot=%d", ErrSlotInThePast, msgSlot, currentSlot)
+	}
+
+	if msgSlot > currentSlot+1 {
+		// msgSlot can be either for current slot or for the next one depending on timing, but not further ahead than that
+		return fmt.Errorf("%w: msgSlot=%d, currentSlot=%d", ErrSlotInTheFuture, msgSlot, currentSlot)
 	}
 
+	return nil
+}
+
+func (v DecryptionKeysValidator) validateKeys(msg *proto.DecryptionKeys) error {
 	if len(msg.Keys) == 0 {
 		return ErrEmptyKeys
 	}
@@ -81,15 +124,41 @@ func (v DecryptionKeysValidator) Validate(msg *proto.DecryptionKeys) error {
 
 	//
 	// TODO when we add the shcrypto library and smart contract state accessors:
-	//      - add validation for signers and signatures based on keyper set and eon infos
+	//      - add validation for signers and signatures based on keyper set and currentEon infos
 	//      - add DecryptionKeys.Validate() equivalent which checks the Key unmarshalling into shcrypto.EpochSecretKey
 	//      - add validation forVerifyEpochSecretKey: check if we should be doing this validation
 	//
+
 	return nil
 }
 
-func NewDecryptionKeysP2pValidatorEx(logger log.Logger, config Config) pubsub.ValidatorEx {
-	dkv := NewDecryptionKeysValidator(config)
+func (v DecryptionKeysValidator) validateEonIndex(msg *proto.DecryptionKeys) error {
+	if msg.Eon > math.MaxInt64 {
+		return fmt.Errorf("%w: %d", ErrEonTooLarge, msg.Eon)
+	}
+
+	msgEonIndex := EonIndex(msg.Eon)
+	currentEon, ok := v.eonTracker.CurrentEon()
+	if !ok {
+		// we're still syncing and are behind - ignore msg, without penalizing peer
+		return fmt.Errorf("%w: %w", ErrIgnoreMsg, ErrCurrentEonUnavailable)
+	}
+
+	_, inRecent := v.eonTracker.RecentEon(msgEonIndex)
+	if msgEonIndex < currentEon.Index && !inRecent {
+		return fmt.Errorf("%w: msgEonIndex=%d, currentEonIndex=%d", ErrEonInThePast, msgEonIndex, currentEon.Index)
+	}
+
+	if msgEonIndex > currentEon.Index && !inRecent {
+		// we may be lagging behind - ignore msg, without penalizing peer
+		return fmt.Errorf("%w: %w: msgEonIndex=%d, currentEonIndex=%d", ErrIgnoreMsg, ErrEonInTheFuture, msgEonIndex, currentEon.Index)
+	}
+
+	return nil
+}
+
+func NewDecryptionKeysP2pValidatorEx(logger log.Logger, config Config, sc SlotCalculator, et EonTracker) pubsub.ValidatorEx {
+	dkv := NewDecryptionKeysValidator(config, sc, et)
 	return func(ctx context.Context, id peer.ID, msg *pubsub.Message) pubsub.ValidationResult {
 		if topic := msg.GetTopic(); topic != DecryptionKeysTopic {
 			logger.Debug("rejecting decryption keys msg due to topic mismatch", "topic", topic, "peer", id)
@@ -104,7 +173,12 @@ func NewDecryptionKeysP2pValidatorEx(logger log.Logger, config Config) pubsub.Va
 
 		err = dkv.Validate(decryptionKeys)
 		if err != nil {
-			logger.Debug("rejecting decryption keys msg due to data validation error", "err", err, "peer", id)
+			if errors.Is(err, ErrIgnoreMsg) {
+				logger.Debug("ignoring decryption keys msg due to", "err", err, "peer", id)
+				return pubsub.ValidationIgnore
+			}
+
+			logger.Debug("rejecting decryption keys msg due to", "err", err, "peer", id)
 			return pubsub.ValidationReject
 		}