sae: Introduce cross-chain tx gossip

vms/saevm/cchain/BUILD.bazel

@@ -17,22 +17,27 @@ go_library(
     name = "cchain",
     srcs = [
         "api.go",
+        "gossip.go",
         "hooks.go",
         "vm.go",
     ],
     importpath = "github.com/ava-labs/avalanchego/vms/saevm/cchain",
     deps = [
         "//api",
+        "//api/metrics",
         "//database",
         "//database/prefixdb",
         "//graft/coreth/core/extstate",
         "//graft/coreth/plugin/evm/customtypes",
         "//graft/evm/constants",
         "//graft/evm/utils/rpc",
         "//ids",
+        "//network/p2p",
+        "//network/p2p/gossip",
         "//snow",
         "//snow/engine/common",
         "//snow/engine/snowman/block",
+        "//utils/bloom",
         "//utils/constants",
         "//utils/formatting",
         "//utils/formatting/address",
@@ -71,6 +76,7 @@ go_test(
     name = "cchain_test",
     srcs = [
         "api_test.go",
+        "gossip_test.go",
         "hooks_test.go",
         "vm_test.go",
     ],
@@ -85,12 +91,14 @@ go_test(
         "//ids",
         "//snow",
         "//snow/engine/common",
-        "//snow/engine/enginetest",
         "//snow/engine/snowman/block",
         "//snow/snowtest",
+        "//snow/validators",
+        "//snow/validators/validatorstest",
         "//utils/crypto/secp256k1",
         "//utils/logging",
         "//utils/set",
+        "//version",
         "//vms/components/avax",
         "//vms/saevm/blocks",
         "//vms/saevm/cchain/tx",

vms/saevm/cchain/api.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/ava-labs/avalanchego/api"
 	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p/gossip"
 	"github.com/ava-labs/avalanchego/snow"
 	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/formatting"
@@ -32,9 +33,10 @@ import (
 // The type is unexported but its methods are exported because gorilla RPC
 // reflects on them to dispatch requests.
 type service struct {
-	ctx    *snow.Context
-	txpool *txpool.Txpool
-	state  *state.State
+	ctx          *snow.Context
+	txpool       *txpool.Txpool
+	pushGossiper *gossip.PushGossiper[*gossipTx]
+	state        *state.State
 
 	chainAlias  string
 	hrp         string
@@ -44,6 +46,7 @@ type service struct {
 func newService(
 	ctx *snow.Context,
 	pool *txpool.Txpool,
+	pushGossiper *gossip.PushGossiper[*gossipTx],
 	db *state.State,
 ) (*service, error) {
 	chainAlias, err := ctx.BCLookup.PrimaryAlias(ctx.ChainID)
@@ -58,9 +61,10 @@ func newService(
 	}
 
 	return &service{
-		ctx:    ctx,
-		txpool: pool,
-		state:  db,
+		ctx:          ctx,
+		txpool:       pool,
+		pushGossiper: pushGossiper,
+		state:        db,
 
 		chainAlias:  chainAlias,
 		hrp:         hrp,
@@ -220,11 +224,12 @@ func (s *service) IssueTx(_ *http.Request, a *api.FormattedTx, r *api.JSONTxID)
 		return fmt.Errorf("parsing transaction: %w", err)
 	}
 
-	if err := s.txpool.Add(t); err != nil {
+	if err := s.txpool.Add(t); err != nil && !errors.Is(err, txpool.ErrAlreadyKnown) {
 		return fmt.Errorf("%w: %w", errIssuingTx, err)
 	}
 
-	// TODO(StephenButtolph): Push gossip the tx.
+	// Even if already in the pool from a peer's gossip, push it to peers.
+	s.pushGossiper.Add(toGossipTx(t))
 
 	r.TxID = t.ID()
 	return nil

vms/saevm/cchain/gossip.go

@@ -0,0 +1,57 @@
+// Copyright (C) 2019, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package cchain
+
+import (
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p/gossip"
+	"github.com/ava-labs/avalanchego/vms/saevm/cchain/tx"
+	"github.com/ava-labs/avalanchego/vms/saevm/cchain/txpool"
+)
+
+var _ gossip.Gossipable = (*gossipTx)(nil)
+
+type gossipTx tx.Tx
+
+func toGossipTx(t *tx.Tx) *gossipTx {
+	return (*gossipTx)(t)
+}
+
+func (t *gossipTx) GossipID() ids.ID { return t.toTx().ID() }
+func (t *gossipTx) toTx() *tx.Tx     { return (*tx.Tx)(t) }
+
+var _ gossip.Marshaller[*gossipTx] = gossipMarshaller{}
+
+type gossipMarshaller struct{}
+
+func (gossipMarshaller) MarshalGossip(t *gossipTx) ([]byte, error) {
+	return t.toTx().Bytes()
+}
+
+func (gossipMarshaller) UnmarshalGossip(b []byte) (*gossipTx, error) {
+	t, err := tx.Parse(b)
+	return toGossipTx(t), err
+}
+
+var _ gossip.Set[*gossipTx] = (*gossipTxPool)(nil)
+
+type gossipTxPool struct {
+	*txpool.Txpool
+}
+
+func newGossipTxPool(pool *txpool.Txpool) *gossipTxPool {
+	return &gossipTxPool{Txpool: pool}
+}
+
+func (p *gossipTxPool) Add(t *gossipTx) error {
+	return p.Txpool.Add(t.toTx())
+}
+
+func (p *gossipTxPool) Iterate(f func(*gossipTx) bool) {
+	for t := range p.Txpool.Iter() {
+		if !f(toGossipTx(t)) {
+			return
+		}
+	}
+}

vms/saevm/cchain/gossip_test.go

@@ -0,0 +1,74 @@
+// Copyright (C) 2019, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package cchain
+
+import (
+	"testing"
+
+	"github.com/ava-labs/libevm/libevm/options"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/utils/set"
+	"github.com/ava-labs/avalanchego/vms/saevm/cchain/tx"
+	"github.com/ava-labs/avalanchego/vms/saevm/cchain/tx/txtest"
+	"github.com/ava-labs/avalanchego/vms/saevm/saetest"
+)
+
+// TestPushGossip verifies that a cross-chain transaction issued to an API node
+// is push-gossiped to a validator for block building.
+func TestPushGossip(t *testing.T) {
+	var (
+		sk        = txtest.NewKey(t)
+		withAlloc = options.Func[sutConfig](func(c *sutConfig) {
+			c.genesis.Alloc = saetest.MaxAllocFor(sk.EthAddress())
+		})
+		vdrID = ids.GenerateTestNodeID()
+		vdrs  = set.Of(vdrID)
+	)
+	apiCtx, api := newSUT(t, withAlloc, withValidators(vdrs))
+	vdrCtx, vdr := newSUT(t, withAlloc, withNodeID(vdrID), withValidators(vdrs))
+	saetest.Connect(t, api, vdr)
+
+	w := newWallet(sk, api.snowCtx, api.Client)
+	stx := w.newMinimalTx(t)
+	require.NoErrorf(t, api.IssueTx(apiCtx, stx), "%T.IssueTx()", api.Client)
+
+	blk := vdr.runConsensusLoop(vdrCtx, t)
+	if diff := cmp.Diff([]*tx.Tx{stx}, blockTxs(t, blk), txtest.CmpOpt()); diff != "" {
+		t.Errorf("%T built by validator after gossip (-want +got):\n%s", blk, diff)
+	}
+}
+
+// TestPullGossip verifies that a validator will share a cross-chain transaction
+// via pull gossip to another connected validator.
+//
+// The API node is only transitively connected to vdrB, so vdrB can only learn
+// about the transaction by pulling it from vdrA.
+func TestPullGossip(t *testing.T) {
+	var (
+		sk        = txtest.NewKey(t)
+		withAlloc = options.Func[sutConfig](func(c *sutConfig) {
+			c.genesis.Alloc = saetest.MaxAllocFor(sk.EthAddress())
+		})
+		vdrIDA = ids.GenerateTestNodeID()
+		vdrIDB = ids.GenerateTestNodeID()
+		vdrs   = set.Of(vdrIDA, vdrIDB)
+	)
+	apiCtx, api := newSUT(t, withAlloc, withValidators(vdrs))
+	_, vdrA := newSUT(t, withAlloc, withNodeID(vdrIDA), withValidators(vdrs))
+	vdrBCtx, vdrB := newSUT(t, withAlloc, withNodeID(vdrIDB), withValidators(vdrs))
+	saetest.Connect(t, api, vdrA)
+	saetest.Connect(t, vdrA, vdrB)
+
+	w := newWallet(sk, api.snowCtx, api.Client)
+	stx := w.newMinimalTx(t)
+	require.NoErrorf(t, api.IssueTx(apiCtx, stx), "%T.IssueTx()", api.Client)
+
+	blk := vdrB.runConsensusLoop(vdrBCtx, t)
+	if diff := cmp.Diff([]*tx.Tx{stx}, blockTxs(t, blk), txtest.CmpOpt()); diff != "" {
+		t.Errorf("%T built by vdrB after gossip (-want +got):\n%s", blk, diff)
+	}
+}

vms/saevm/cchain/vm.go

@@ -13,16 +13,22 @@ import (
 	"fmt"
 	"net/http"
 	"slices"
+	"sync"
+	"time"
 
 	"github.com/ava-labs/libevm/core"
 	"github.com/ava-labs/libevm/core/rawdb"
 	"github.com/ava-labs/libevm/core/txpool/legacypool"
 	"github.com/ava-labs/libevm/triedb"
 
+	"github.com/ava-labs/avalanchego/api/metrics"
 	"github.com/ava-labs/avalanchego/database/prefixdb"
 	"github.com/ava-labs/avalanchego/graft/evm/utils/rpc"
+	"github.com/ava-labs/avalanchego/network/p2p"
+	"github.com/ava-labs/avalanchego/network/p2p/gossip"
 	"github.com/ava-labs/avalanchego/snow"
 	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/utils/bloom"
 	"github.com/ava-labs/avalanchego/vms/evm/database"
 	"github.com/ava-labs/avalanchego/vms/saevm/cchain/state"
 	"github.com/ava-labs/avalanchego/vms/saevm/cchain/txpool"
@@ -36,9 +42,14 @@ import (
 type VM struct {
 	*sae.VM // created by [VM.Initialize]
 
-	ctx    *snow.Context
-	state  *state.State
-	txpool *txpool.Txpool
+	// gossip frequencies are configurable to speed up testing.
+	pullGossipPeriod time.Duration
+	pushGossipPeriod time.Duration
+
+	ctx          *snow.Context
+	state        *state.State
+	txpool       *txpool.Txpool
+	pushGossiper *gossip.PushGossiper[*gossipTx]
 
 	// onClose are executed in reverse order during [VM.Shutdown]. If a resource
 	// depends on another resource, it MUST be added AFTER the resource it
@@ -126,6 +137,61 @@ func (v *VM) Initialize(
 		v.txpool.Close()
 		return nil
 	})
+
+	reg, err := metrics.MakeAndRegister(snowCtx.Metrics, "cchain")
+	if err != nil {
+		return fmt.Errorf("making metrics: %w", err)
+	}
+	bloomMetrics, err := bloom.NewMetrics("gossip_bloom", reg)
+	if err != nil {
+		return fmt.Errorf("creating gossip bloom metrics: %w", err)
+	}
+	gossipSet, err := gossip.NewBloomSet(
+		newGossipTxPool(v.txpool),
+		gossip.BloomSetConfig{
+			Metrics: bloomMetrics,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("creating gossip bloom set: %w", err)
+	}
+	gossipHandler, pullGossiper, pushGossiper, err := gossip.NewSystem(
+		snowCtx.NodeID,
+		v.Network,
+		v.ValidatorPeers,
+		gossipSet,
+		gossipMarshaller{},
+		gossip.SystemConfig{
+			Log:           snowCtx.Log,
+			Registry:      reg,
+			Namespace:     "gossip",
+			HandlerID:     p2p.AtomicTxGossipHandlerID,
+			RequestPeriod: v.pullGossipPeriod,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("creating cross-chain tx gossip system: %w", err)
+	}
+	v.pushGossiper = pushGossiper
+
+	if err := v.VM.AddHandler(p2p.AtomicTxGossipHandlerID, gossipHandler); err != nil {
+		return fmt.Errorf("registering cross-chain tx gossip handler: %w", err)
+	}
+
+	gossipCtx, cancelGossip := context.WithCancel(context.Background())
+	var gossipWG sync.WaitGroup
+	gossipWG.Go(func() {
+		gossip.Every(gossipCtx, snowCtx.Log, pullGossiper, v.pullGossipPeriod)
+	})
+	gossipWG.Go(func() {
+		gossip.Every(gossipCtx, snowCtx.Log, pushGossiper, v.pushGossipPeriod)
+	})
+	v.onClose = append(v.onClose, func(context.Context) error {
+		cancelGossip()
+		gossipWG.Wait()
+		return nil
+	})
+
 	return nil
 }
 
@@ -142,7 +208,7 @@ func (v *VM) CreateHandlers(ctx context.Context) (map[string]http.Handler, error
 		return nil, fmt.Errorf("creating SAE handlers: %w", err)
 	}
 
-	service, err := newService(v.ctx, v.txpool, v.state)
+	service, err := newService(v.ctx, v.txpool, v.pushGossiper, v.state)
 	if err != nil {
 		return nil, fmt.Errorf("creating avax service: %w", err)
 	}