Replace toEngine channel with a dedicated subscription API

chains/linearizable_vm.go

@@ -5,6 +5,7 @@ package chains
 
 import (
 	"context"
+	"sync"
 
 	"github.com/ava-labs/avalanchego/database"
 	"github.com/ava-labs/avalanchego/ids"
@@ -28,26 +29,38 @@ type initializeOnLinearizeVM struct {
 	vmToInitialize common.VM
 	vmToLinearize  *linearizeOnInitializeVM
 
-	ctx          *snow.Context
-	db           database.Database
-	genesisBytes []byte
-	upgradeBytes []byte
-	configBytes  []byte
-	toEngine     chan<- common.Message
-	fxs          []*common.Fx
-	appSender    common.AppSender
+	ctx              *snow.Context
+	db               database.Database
+	genesisBytes     []byte
+	upgradeBytes     []byte
+	configBytes      []byte
+	fxs              []*common.Fx
+	appSender        common.AppSender
+	waitForLinearize chan struct{}
+	linearizeOnce    sync.Once
+}
+
+func (vm *initializeOnLinearizeVM) WaitForEvent(ctx context.Context) (common.Message, error) {
+	select {
+	case <-vm.waitForLinearize:
+		return vm.vmToInitialize.WaitForEvent(ctx)
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	}
 }
 
 func (vm *initializeOnLinearizeVM) Linearize(ctx context.Context, stopVertexID ids.ID) error {
 	vm.vmToLinearize.stopVertexID = stopVertexID
+	defer vm.linearizeOnce.Do(func() {
+		close(vm.waitForLinearize)
+	})
 	return vm.vmToInitialize.Initialize(
 		ctx,
 		vm.ctx,
 		vm.db,
 		vm.genesisBytes,
 		vm.upgradeBytes,
 		vm.configBytes,
-		vm.toEngine,
 		vm.fxs,
 		vm.appSender,
 	)
@@ -74,9 +87,8 @@ func (vm *linearizeOnInitializeVM) Initialize(
 	_ []byte,
 	_ []byte,
 	_ []byte,
-	toEngine chan<- common.Message,
 	_ []*common.Fx,
 	_ common.AppSender,
 ) error {
-	return vm.Linearize(ctx, vm.stopVertexID, toEngine)
+	return vm.Linearize(ctx, vm.stopVertexID)
 }

chains/manager.go

@@ -727,10 +727,6 @@ func (m *manager) createAvalancheChain(
 		},
 	)
 
-	// The channel through which a VM may send messages to the consensus engine
-	// VM uses this channel to notify engine that a block is ready to be made
-	msgChan := make(chan common.Message, defaultChannelSize)
-
 	// The only difference between using avalancheMessageSender and
 	// snowmanMessageSender here is where the metrics will be placed. Because we
 	// end up using this sender after the linearization, we pass in
@@ -742,7 +738,6 @@ func (m *manager) createAvalancheChain(
 		genesisData,
 		chainConfig.Upgrade,
 		chainConfig.Config,
-		msgChan,
 		fxs,
 		snowmanMessageSender,
 	)
@@ -811,19 +806,25 @@ func (m *manager) createAvalancheChain(
 		vmWrappingProposerVM = tracedvm.NewBlockVM(vmWrappingProposerVM, "proposervm", m.Tracer)
 	}
 
+	cn := &block.ChangeNotifier{
+		ChainVM: vmWrappingProposerVM,
+	}
+
+	vmWrappingProposerVM = cn
+
 	// Note: linearizableVM is the VM that the Avalanche engines should be
 	// using.
 	linearizableVM := &initializeOnLinearizeVM{
-		DAGVM:          dagVM,
-		vmToInitialize: vmWrappingProposerVM,
-		vmToLinearize:  untracedVMWrappedInsideProposerVM,
+		waitForLinearize: make(chan struct{}),
+		DAGVM:            dagVM,
+		vmToInitialize:   vmWrappingProposerVM,
+		vmToLinearize:    untracedVMWrappedInsideProposerVM,
 
 		ctx:          ctx.Context,
 		db:           vmDB,
 		genesisBytes: genesisData,
 		upgradeBytes: chainConfig.Upgrade,
 		configBytes:  chainConfig.Config,
-		toEngine:     msgChan,
 		fxs:          fxs,
 		appSender:    snowmanMessageSender,
 	}
@@ -885,8 +886,9 @@ func (m *manager) createAvalancheChain(
 	// Asynchronously passes messages from the network to the consensus engine
 	h, err := handler.New(
 		ctx,
+		cn,
+		linearizableVM.WaitForEvent,
 		vdrs,
-		msgChan,
 		m.FrontierPollFrequency,
 		m.ConsensusAppConcurrency,
 		m.ResourceTracker,
@@ -1202,9 +1204,10 @@ func (m *manager) createSnowmanChain(
 		vm = tracedvm.NewBlockVM(vm, "proposervm", m.Tracer)
 	}
 
-	// The channel through which a VM may send messages to the consensus engine
-	// VM uses this channel to notify engine that a block is ready to be made
-	msgChan := make(chan common.Message, defaultChannelSize)
+	cn := &block.ChangeNotifier{
+		ChainVM: vm,
+	}
+	vm = cn
 
 	if err := vm.Initialize(
 		context.TODO(),
@@ -1213,7 +1216,6 @@ func (m *manager) createSnowmanChain(
 		genesisData,
 		chainConfig.Upgrade,
 		chainConfig.Config,
-		msgChan,
 		fxs,
 		messageSender,
 	); err != nil {
@@ -1277,8 +1279,9 @@ func (m *manager) createSnowmanChain(
 	// Asynchronously passes messages from the network to the consensus engine
 	h, err := handler.New(
 		ctx,
+		cn,
+		vm.WaitForEvent,
 		vdrs,
-		msgChan,
 		m.FrontierPollFrequency,
 		m.ConsensusAppConcurrency,
 		m.ResourceTracker,

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/DataDog/zstd v1.5.2
 	github.com/StephenButtolph/canoto v0.17.1
 	github.com/antithesishq/antithesis-sdk-go v0.3.8
-	github.com/ava-labs/coreth v0.15.2-rc.0.0.20250620163936-0058a7ec6d9d
+	github.com/ava-labs/coreth v0.15.3-0.20250707191145-055af77d543b
 	github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60
 	github.com/ava-labs/libevm v1.13.14-0.3.0.rc.1
 	github.com/btcsuite/btcd/btcutil v1.1.3

go.sum

@@ -66,8 +66,8 @@ github.com/antithesishq/antithesis-sdk-go v0.3.8/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
-github.com/ava-labs/coreth v0.15.2-rc.0.0.20250620163936-0058a7ec6d9d h1:nZA7nEhlf2uWM1doZz4WFiMkZVZ0iH+c3IebokIO1sk=
-github.com/ava-labs/coreth v0.15.2-rc.0.0.20250620163936-0058a7ec6d9d/go.mod h1:Mk5g2ZI9lEbxA67qikOtUrq71Yr389B34S+Ddsa51K4=
+github.com/ava-labs/coreth v0.15.3-0.20250707191145-055af77d543b h1:+Dmp2AiyY+A9eTyDFkElYtM3ZxJe8bSxAjTAePVoHB8=
+github.com/ava-labs/coreth v0.15.3-0.20250707191145-055af77d543b/go.mod h1:uRMiwnKMtGac3Oftq9/tULLjtgjChe5zoRjQhvhePaw=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60 h1:EL66gtXOAwR/4KYBjOV03LTWgkEXvLePribLlJNu4g0=
 github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60/go.mod h1:/7qKobTfbzBu7eSTVaXMTr56yTYk4j2Px6/8G+idxHo=
 github.com/ava-labs/libevm v1.13.14-0.3.0.rc.1 h1:vBMYo+Iazw0rGTr+cwjkBdh5eadLPlv4ywI4lKye3CA=

node/node.go

@@ -80,7 +80,7 @@ import (
 	databasefactory "github.com/ava-labs/avalanchego/database/factory"
 	avmconfig "github.com/ava-labs/avalanchego/vms/avm/config"
 	platformconfig "github.com/ava-labs/avalanchego/vms/platformvm/config"
-	coreth "github.com/ava-labs/coreth/plugin/evm"
+	coreth "github.com/ava-labs/coreth/plugin/factory"
 )
 
 const (