sync: coreth PR #963,981,1009: sync package extension and related items (#1679)

Signed-off-by: Jonathan Oppenheimer <[email protected]>
Signed-off-by: Jonathan Oppenheimer <[email protected]>
Co-authored-by: Austin Larson <[email protected]>
Co-authored-by: Ceyhun Onur <[email protected]>

Author: 3 people

consensus/dummy/consensus.go

@@ -34,10 +34,39 @@ type Mode struct {
 	ModeSkipCoinbase bool
 }
 
-type DummyEngine struct {
-	consensusMode      Mode
-	desiredDelayExcess *acp226.DelayExcess
-}
+type (
+	OnFinalizeAndAssembleCallbackType = func(
+		header *types.Header,
+		parent *types.Header,
+		state *state.StateDB,
+		txs []*types.Transaction,
+	) (
+		extraData []byte,
+		blockFeeContribution *big.Int,
+		extDataGasUsed *big.Int,
+		err error,
+	)
+
+	OnExtraStateChangeType = func(
+		block *types.Block,
+		parent *types.Header,
+		statedb *state.StateDB,
+	) (
+		blockFeeContribution *big.Int,
+		extDataGasUsed *big.Int,
+		err error,
+	)
+
+	ConsensusCallbacks struct {
+		OnFinalizeAndAssemble OnFinalizeAndAssembleCallbackType
+		OnExtraStateChange    OnExtraStateChangeType
+	}
+
+	DummyEngine struct {
+		consensusMode      Mode
+		desiredDelayExcess *acp226.DelayExcess
+	}
+)
 
 func NewDummyEngine(
 	mode Mode,

plugin/evm/block_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/ava-labs/subnet-evm/params"
 	"github.com/ava-labs/subnet-evm/params/extras"
+	"github.com/ava-labs/subnet-evm/plugin/evm/extension"
 	"github.com/ava-labs/subnet-evm/precompile/precompileconfig"
 )
 
@@ -26,8 +27,9 @@ func TestHandlePrecompileAccept(t *testing.T) {
 
 	db := rawdb.NewMemoryDatabase()
 	vm := &VM{
-		chaindb:     db,
-		chainConfig: params.TestChainConfig,
+		chaindb:         db,
+		chainConfig:     params.TestChainConfig,
+		extensionConfig: &extension.Config{},
 	}
 
 	precompileAddr := common.Address{0x05}

plugin/evm/config/config.go

@@ -236,7 +236,7 @@ func (d Duration) MarshalJSON() ([]byte, error) {
 }
 
 // validate returns an error if this is an invalid config.
-func (c *Config) validate(_ uint32) error {
+func (c *Config) validate(uint32) error {
 	if c.PopulateMissingTries != nil && (c.OfflinePruning || c.Pruning) {
 		return fmt.Errorf("cannot enable populate missing tries while offline pruning (enabled: %t)/pruning (enabled: %t) are enabled", c.OfflinePruning, c.Pruning)
 	}

plugin/evm/extension/config.go

@@ -4,13 +4,30 @@
 package extension
 
 import (
+	"context"
 	"errors"
 
+	"github.com/ava-labs/avalanchego/database"
+	"github.com/ava-labs/avalanchego/database/versiondb"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p"
+	"github.com/ava-labs/avalanchego/snow/consensus/snowman"
+	"github.com/ava-labs/avalanchego/snow/engine/snowman/block"
 	"github.com/ava-labs/avalanchego/utils/timer/mockable"
+	"github.com/ava-labs/libevm/common"
+	"github.com/ava-labs/libevm/core/types"
+	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/ava-labs/subnet-evm/consensus/dummy"
+	"github.com/ava-labs/subnet-evm/core"
+	"github.com/ava-labs/subnet-evm/params"
+	"github.com/ava-labs/subnet-evm/params/extras"
+	"github.com/ava-labs/subnet-evm/plugin/evm/config"
 	"github.com/ava-labs/subnet-evm/plugin/evm/message"
 	"github.com/ava-labs/subnet-evm/plugin/evm/sync"
 	"github.com/ava-labs/subnet-evm/sync/handlers"
+
+	avalanchecommon "github.com/ava-labs/avalanchego/snow/engine/common"
 )
 
 var (
@@ -20,6 +37,85 @@ var (
 	errNilClock               = errors.New("nil clock")
 )
 
+type ExtensibleVM interface {
+	// SetExtensionConfig sets the configuration for the VM extension
+	// Should be called before any other method and only once
+	SetExtensionConfig(config *Config) error
+	// NewClient returns a client to send messages with for the given protocol
+	NewClient(protocol uint64) *p2p.Client
+	// AddHandler registers a server handler for an application protocol
+	AddHandler(protocol uint64, handler p2p.Handler) error
+	// GetExtendedBlock returns the VMBlock for the given ID or an error if the block is not found
+	GetExtendedBlock(context.Context, ids.ID) (ExtendedBlock, error)
+	// LastAcceptedExtendedBlock returns the last accepted VM block
+	LastAcceptedExtendedBlock() ExtendedBlock
+	// ChainConfig returns the chain config for the VM
+	ChainConfig() *params.ChainConfig
+	// P2PValidators returns the validators for the network
+	P2PValidators() *p2p.Validators
+	// Blockchain returns the blockchain client
+	Blockchain() *core.BlockChain
+	// Config returns the configuration for the VM
+	Config() config.Config
+	// MetricRegistry returns the metric registry for the VM
+	MetricRegistry() *prometheus.Registry
+	// ReadLastAccepted returns the last accepted block hash and height
+	ReadLastAccepted() (common.Hash, uint64, error)
+	// VersionDB returns the versioned database for the VM
+	VersionDB() *versiondb.Database
+}
+
+// InnerVM is the interface that must be implemented by the VM
+// that's being wrapped by the extension
+type InnerVM interface {
+	ExtensibleVM
+	avalanchecommon.VM
+	block.ChainVM
+	block.BuildBlockWithContextChainVM
+	block.StateSyncableVM
+}
+
+// ExtendedBlock is a block that can be used by the extension
+type ExtendedBlock interface {
+	snowman.Block
+	GetEthBlock() *types.Block
+	GetBlockExtension() BlockExtension
+}
+
+type BlockExtender interface {
+	// NewBlockExtension is called when a new block is created
+	NewBlockExtension(b ExtendedBlock) (BlockExtension, error)
+}
+
+// BlockExtension allows the VM extension to handle block processing events.
+type BlockExtension interface {
+	// SyntacticVerify verifies the block syntactically
+	// it can be implemented to extend inner block verification
+	SyntacticVerify(rules extras.Rules) error
+	// SemanticVerify verifies the block semantically
+	// it can be implemented to extend inner block verification
+	SemanticVerify() error
+	// CleanupVerified is called when a block has passed SemanticVerify and SynctacticVerify,
+	// and should be cleaned up due to error or verification runs under non-write mode. This
+	// does not return an error because the block has already been verified.
+	CleanupVerified()
+	// Accept is called when a block is accepted by the block manager. Accept takes a
+	// database.Batch that contains the changes that were made to the database as a result
+	// of accepting the block. The changes in the batch should be flushed to the database in this method.
+	Accept(acceptedBatch database.Batch) error
+	// Reject is called when a block is rejected by the block manager
+	Reject() error
+}
+
+// BuilderMempool is a mempool that's used in the block builder
+type BuilderMempool interface {
+	// PendingLen returns the number of pending transactions
+	// that are waiting to be included in a block
+	PendingLen() int
+	// SubscribePendingTxs returns a channel that's signaled when there are pending transactions
+	SubscribePendingTxs() <-chan struct{}
+}
+
 // LeafRequestConfig is the configuration to handle leaf requests
 // in the network and syncer
 type LeafRequestConfig struct {
@@ -33,13 +129,29 @@ type LeafRequestConfig struct {
 
 // Config is the configuration for the VM extension
 type Config struct {
+	// ConsensusCallbacks is the consensus callbacks to use
+	// for the VM to be used in consensus engine.
+	// Callback functions can be nil.
+	ConsensusCallbacks dummy.ConsensusCallbacks
 	// SyncSummaryProvider is the sync summary provider to use
 	// for the VM to be used in syncer.
 	// It's required and should be non-nil
 	SyncSummaryProvider sync.SummaryProvider
+	// SyncExtender can extend the syncer to handle custom sync logic.
+	// It's optional and can be nil
+	SyncExtender sync.Extender
 	// SyncableParser is to parse summary messages from the network.
 	// It's required and should be non-nil
 	SyncableParser message.SyncableParser
+	// BlockExtender allows the VM extension to create an extension to handle block processing events.
+	// It's optional and can be nil
+	BlockExtender BlockExtender
+	// ExtraSyncLeafHandlerConfig is the extra configuration to handle leaf requests
+	// in the network and syncer. It's optional and can be nil
+	ExtraSyncLeafHandlerConfig *LeafRequestConfig
+	// ExtraMempool is the mempool to be used in the block builder.
+	// It's optional and can be nil
+	ExtraMempool BuilderMempool
 	// Clock is the clock to use for time related operations.
 	// It's optional and can be nil
 	Clock *mockable.Clock

plugin/evm/message/block_sync_summary_test.go

@@ -0,0 +1,43 @@
+// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package message
+
+import (
+	"encoding/base64"
+	"testing"
+
+	"github.com/ava-labs/avalanchego/snow/engine/snowman/block"
+	"github.com/ava-labs/libevm/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMarshalBlockSyncSummary(t *testing.T) {
+	blockSyncSummary, err := NewBlockSyncSummary(common.Hash{1}, 2, common.Hash{3})
+	require.NoError(t, err)
+
+	require.Equal(t, common.Hash{1}, blockSyncSummary.GetBlockHash())
+	require.Equal(t, uint64(2), blockSyncSummary.Height())
+	require.Equal(t, common.Hash{3}, blockSyncSummary.GetBlockRoot())
+
+	expectedBase64Bytes := "AAAAAAAAAAAAAgEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+	require.Equal(t, expectedBase64Bytes, base64.StdEncoding.EncodeToString(blockSyncSummary.Bytes()))
+
+	parser := NewBlockSyncSummaryParser()
+	called := false
+	acceptImplTest := func(Syncable) (block.StateSyncMode, error) {
+		called = true
+		return block.StateSyncSkipped, nil
+	}
+	s, err := parser.Parse(blockSyncSummary.Bytes(), acceptImplTest)
+	require.NoError(t, err)
+	require.Equal(t, blockSyncSummary.GetBlockHash(), s.GetBlockHash())
+	require.Equal(t, blockSyncSummary.Height(), s.Height())
+	require.Equal(t, blockSyncSummary.GetBlockRoot(), s.GetBlockRoot())
+	require.Equal(t, blockSyncSummary.Bytes(), s.Bytes())
+
+	mode, err := s.Accept(t.Context())
+	require.NoError(t, err)
+	require.Equal(t, block.StateSyncSkipped, mode)
+	require.True(t, called)
+}

plugin/evm/message/codec.go

@@ -22,13 +22,11 @@ func init() {
 	c := linearcodec.NewDefault()
 
 	// Skip registration to keep registeredTypes unchanged after legacy gossip deprecation
-	c.SkipRegistrations(1)
+	// Gossip types and sync summary type removed from codec
+	c.SkipRegistrations(2)
 
 	errs := wrappers.Errs{}
 	errs.Add(
-		// Types for state sync frontier consensus
-		c.RegisterType(BlockSyncSummary{}),
-
 		// state sync types
 		c.RegisterType(BlockRequest{}),
 		c.RegisterType(BlockResponse{}),

plugin/evm/vm.go

@@ -187,6 +187,7 @@ type VM struct {
 
 	config config.Config
 
+	chainID     *big.Int
 	genesisHash common.Hash
 	chainConfig *params.ChainConfig
 	ethConfig   ethconfig.Config
@@ -279,7 +280,7 @@ func (vm *VM) Initialize(
 	appSender commonEng.AppSender,
 ) error {
 	vm.ctx = chainCtx
-	vm.stateSyncDone = make(chan struct{})
+
 	cfg, deprecateMsg, err := config.GetConfig(configBytes, vm.ctx.NetworkID)
 	if err != nil {
 		return fmt.Errorf("failed to get config: %w", err)
@@ -341,14 +342,22 @@ func (vm *VM) Initialize(
 		return err
 	}
 
+	// vm.ChainConfig() should be available for wrapping VMs before vm.initializeChain()
+	vm.chainConfig = g.Config
+	vm.chainID = g.Config.ChainID
+
 	vm.ethConfig = ethconfig.NewDefaultConfig()
 	vm.ethConfig.Genesis = g
-	// NetworkID here is different than Avalanche's NetworkID.
-	// Avalanche's NetworkID represents the Avalanche network is running on
-	// like Fuji, Mainnet, Local, etc.
-	// The NetworkId here is kept same as ChainID to be compatible with
-	// Ethereum tooling.
-	vm.ethConfig.NetworkId = g.Config.ChainID.Uint64()
+	vm.ethConfig.NetworkId = vm.chainID.Uint64()
+	vm.genesisHash = vm.ethConfig.Genesis.ToBlock().Hash() // must create genesis hash before [vm.ReadLastAccepted]
+	lastAcceptedHash, lastAcceptedHeight, err := vm.ReadLastAccepted()
+	if err != nil {
+		return err
+	}
+	log.Info("read last accepted",
+		"hash", lastAcceptedHash,
+		"height", lastAcceptedHeight,
+	)
 
 	// Set minimum price for mining and default gas price oracle value to the min
 	// gas price to prevent so transactions and blocks all use the correct fees
@@ -438,20 +447,6 @@ func (vm *VM) Initialize(
 		vm.ethConfig.Miner.Etherbase = constants.BlackholeAddr
 	}
 
-	vm.chainConfig = g.Config
-
-	// create genesisHash after applying upgradeBytes in case
-	// upgradeBytes modifies genesis.
-	vm.genesisHash = vm.ethConfig.Genesis.ToBlock().Hash() // must create genesis hash before [vm.readLastAccepted]
-	lastAcceptedHash, lastAcceptedHeight, err := vm.readLastAccepted()
-	if err != nil {
-		return err
-	}
-	log.Info("read last accepted",
-		"hash", lastAcceptedHash,
-		"height", lastAcceptedHeight,
-	)
-
 	vm.networkCodec = message.Codec
 	vm.Network, err = network.NewNetwork(vm.ctx, appSender, vm.networkCodec, vm.config.MaxOutboundActiveRequests, vm.sdkMetrics)
 	if err != nil {
@@ -494,7 +489,6 @@ func (vm *VM) Initialize(
 	if err != nil {
 		return err
 	}
-
 	if err := vm.initializeChain(lastAcceptedHash, vm.ethConfig); err != nil {
 		return err
 	}
@@ -730,6 +724,7 @@ func (vm *VM) initializeStateSync(lastAcceptedHeight uint64) error {
 		MetadataDB:         vm.metadataDB,
 		Acceptor:           vm,
 		Parser:             vm.extensionConfig.SyncableParser,
+		Extender:           nil,
 	})
 
 	// If StateSync is disabled, clear any ongoing summary so that we will not attempt to resume
@@ -803,7 +798,6 @@ func (vm *VM) onBootstrapStarted() error {
 	// Ensure snapshots are initialized before bootstrapping (i.e., if state sync is skipped).
 	// Note calling this function has no effect if snapshots are already initialized.
 	vm.blockChain.InitializeSnapshots()
-
 	return nil
 }
 
@@ -1310,7 +1304,7 @@ func (vm *VM) startContinuousProfiler() {
 // last accepted block hash and height by reading directly from [vm.chaindb] instead of relying
 // on [chain].
 // Note: assumes [vm.chaindb] and [vm.genesisHash] have been initialized.
-func (vm *VM) readLastAccepted() (common.Hash, uint64, error) {
+func (vm *VM) ReadLastAccepted() (common.Hash, uint64, error) {
 	// Attempt to load last accepted block to determine if it is necessary to
 	// initialize state with the genesis block.
 	lastAcceptedBytes, lastAcceptedErr := vm.acceptedBlockDB.Get(lastAcceptedKey)