feat: txgossip package

blocks/block.go

@@ -13,6 +13,7 @@ import (
 	"sync/atomic"
 
 	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/libevm/common"
 	"github.com/ava-labs/libevm/core/types"
 	"go.uber.org/zap"
 )
@@ -114,3 +115,27 @@ func (b *Block) CopyAncestorsFrom(c *Block) error {
 	a := c.ancestry.Load()
 	return b.setAncestors(a.parent, a.lastSettled)
 }
+
+// A Source returns a [Block] that matches both a hash and number, and a boolean
+// indicating if such a block was found.
+type Source func(hash common.Hash, number uint64) (*Block, bool)
+
+// EthBlock returns the [types.Block] with the given hash and number, or nil if
+// not found.
+func (s Source) EthBlock(h common.Hash, n uint64) *types.Block {
+	b, ok := s(h, n)
+	if !ok {
+		return nil
+	}
+	return b.EthBlock()
+}
+
+// Header returns the [types.Header] with the given hash and number, or nil if
+// not found.
+func (s Source) Header(h common.Hash, n uint64) *types.Header {
+	b, ok := s(h, n)
+	if !ok {
+		return nil
+	}
+	return b.Header()
+}

blocks/blockstest/chain.go

@@ -99,6 +99,8 @@ func (cb *ChainBuilder) AllExceptGenesis() []*blocks.Block {
 	return slices.Clone(cb.chain[1:])
 }
 
+var _ blocks.Source = (*ChainBuilder)(nil).GetBlock
+
 // GetBlock returns the block with specified hash and height, and a flag
 // indicating if it was found. If either argument does not match, it returns
 // `nil, false`.

cmputils/types.go

@@ -26,6 +26,14 @@ func BlocksByHash() cmp.Option {
 	})
 }
 
+// TransactionsByHash returns a [cmp.Comparer] for [types.Transaction] pointers,
+// equating them by hash alone.
+func TransactionsByHash() cmp.Option {
+	return ComparerWithNilCheck(func(t, u *types.Transaction) bool {
+		return t.Hash() == u.Hash()
+	})
+}
+
 // ReceiptsByTxHash returns a [cmp.Comparer] for [types.Receipt] pointers,
 // equating them by transaction hash alone.
 func ReceiptsByTxHash() cmp.Option {

go.mod

@@ -76,7 +76,7 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.16.0 // indirect
+	github.com/prometheus/client_golang v1.16.0
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect

saexec/context.go

@@ -13,23 +13,15 @@ import (
 	"github.com/ava-labs/strevm/blocks"
 )
 
-// A BlockSource returns a block that matches both a hash and number, or nil
-// if not found.
-type BlockSource func(hash common.Hash, number uint64) *blocks.Block
-
 var _ core.ChainContext = (*chainContext)(nil)
 
 type chainContext struct {
-	blocks BlockSource
+	blocks blocks.Source
 	log    logging.Logger
 }
 
 func (c *chainContext) GetHeader(h common.Hash, n uint64) *types.Header {
-	b := c.blocks(h, n)
-	if b == nil {
-		return nil
-	}
-	return b.Header()
+	return c.blocks.Header(h, n)
 }
 
 func (c *chainContext) Engine() consensus.Engine {

saexec/execution.go

@@ -32,6 +32,8 @@ func (e *Executor) Enqueue(ctx context.Context, block *blocks.Block) error {
 	for {
 		select {
 		case e.queue <- block:
+			e.lastEnqueued.Store(block)
+			e.enqueueEvents.Send(block.EthBlock())
 			return nil
 
 		case <-ctx.Done():

saexec/saexec.go

@@ -31,12 +31,13 @@ type Executor struct {
 	log        logging.Logger
 	hooks      hook.Points
 
-	queue        chan *blocks.Block
-	lastExecuted atomic.Pointer[blocks.Block]
+	queue                      chan *blocks.Block
+	lastEnqueued, lastExecuted atomic.Pointer[blocks.Block]
 
-	headEvents  event.FeedOf[core.ChainHeadEvent]
-	chainEvents event.FeedOf[core.ChainEvent]
-	logEvents   event.FeedOf[[]*types.Log]
+	enqueueEvents event.FeedOf[*types.Block]
+	headEvents    event.FeedOf[core.ChainHeadEvent]
+	chainEvents   event.FeedOf[core.ChainEvent]
+	logEvents     event.FeedOf[[]*types.Log]
 
 	chainContext core.ChainContext
 	chainConfig  *params.ChainConfig
@@ -55,7 +56,7 @@ type Executor struct {
 // executed block after shutdown and recovery.
 func New(
 	lastExecuted *blocks.Block,
-	blockSrc BlockSource,
+	blockSrc blocks.Source,
 	chainConfig *params.ChainConfig,
 	db ethdb.Database,
 	triedbConfig *triedb.Config,
@@ -84,6 +85,7 @@ func New(
 		stateCache:   cache,
 		snaps:        snaps,
 	}
+	e.lastEnqueued.Store(lastExecuted)
 	e.lastExecuted.Store(lastExecuted)
 
 	go e.processQueue()
@@ -127,3 +129,8 @@ func (e *Executor) StateCache() state.Database {
 func (e *Executor) LastExecuted() *blocks.Block {
 	return e.lastExecuted.Load()
 }
+
+// LastEnqueued returns the last-enqueued block in a threadsafe manner.
+func (e *Executor) LastEnqueued() *blocks.Block {
+	return e.lastEnqueued.Load()
+}

saexec/saexec_test.go

@@ -32,7 +32,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 
-	"github.com/ava-labs/strevm/blocks"
 	"github.com/ava-labs/strevm/blocks/blockstest"
 	"github.com/ava-labs/strevm/cmputils"
 	"github.com/ava-labs/strevm/gastime"
@@ -84,15 +83,8 @@ func newSUT(tb testing.TB, hooks hook.Points) (context.Context, SUT) {
 	chain.SetDefaultOptions(blockstest.WithBlockOptions(
 		blockstest.WithLogger(logger),
 	))
-	src := BlockSource(func(h common.Hash, n uint64) *blocks.Block {
-		b, ok := chain.GetBlock(h, n)
-		if !ok {
-			return nil
-		}
-		return b
-	})
 
-	e, err := New(genesis, src, config, db, tdbConfig, hooks, logger)
+	e, err := New(genesis, chain.GetBlock, config, db, tdbConfig, hooks, logger)
 	require.NoError(tb, err, "New()")
 	tb.Cleanup(e.Close)
 

saexec/subscription.go

@@ -24,6 +24,12 @@ func (e *Executor) sendPostExecutionEvents(b *types.Block, receipts types.Receip
 	e.logEvents.Send(logs)
 }
 
+// SubscribeBlockEnqueueEvent returns a new subscription for each block queued
+// [Executor.Enqueue].
+func (e *Executor) SubscribeBlockEnqueueEvent(ch chan<- *types.Block) event.Subscription {
+	return e.enqueueEvents.Subscribe(ch)
+}
+
 // SubscribeChainHeadEvent returns a new subscription for each
 // [core.ChainHeadEvent] emitted after execution of a [blocks.Block].
 func (e *Executor) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {

txgossip/blockchain.go

@@ -0,0 +1,108 @@
+// Copyright (C) 2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package txgossip
+
+import (
+	"github.com/ava-labs/libevm/common"
+	"github.com/ava-labs/libevm/core"
+	"github.com/ava-labs/libevm/core/state"
+	"github.com/ava-labs/libevm/core/txpool"
+	"github.com/ava-labs/libevm/core/txpool/legacypool"
+	"github.com/ava-labs/libevm/core/types"
+	"github.com/ava-labs/libevm/event"
+	"github.com/ava-labs/libevm/params"
+
+	"github.com/ava-labs/strevm/blocks"
+	"github.com/ava-labs/strevm/saexec"
+)
+
+// A BlockChain is the union of [txpool.BlockChain] and [legacypool.BlockChain].
+type BlockChain interface {
+	txpool.BlockChain
+	legacypool.BlockChain
+}
+
+// NewBlockChain wraps an [saexec.Executor] to be compatible with a
+// non-blob-transaction mempool.
+func NewBlockChain(exec *saexec.Executor, blocks blocks.Source) BlockChain {
+	return &blockchain{
+		exec:   exec,
+		blocks: blocks,
+	}
+}
+
+type blockchain struct {
+	exec   *saexec.Executor
+	blocks blocks.Source
+}
+
+func (bc *blockchain) Config() *params.ChainConfig {
+	return bc.exec.ChainConfig()
+}
+
+func (bc *blockchain) CurrentBlock() *types.Header {
+	return bc.exec.LastEnqueued().Header()
+}
+
+func (bc *blockchain) GetBlock(hash common.Hash, number uint64) *types.Block {
+	return bc.blocks.EthBlock(hash, number)
+}
+
+func (bc *blockchain) StateAt(root common.Hash) (*state.StateDB, error) {
+	return state.New(root, bc.exec.StateCache(), nil)
+}
+
+// SubscribeChainHeadEvent subscribes to block enqueueing, NOT to regular head
+// events as these only occur after execution. Enqueuing is equivalent to block
+// acceptance, which is when a transaction SHOULD be removed from the mempool.
+func (bc *blockchain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
+	bCh := make(chan *types.Block)
+	sub := bc.exec.SubscribeBlockEnqueueEvent(bCh)
+
+	p := &pipe{
+		in:   bCh,
+		out:  ch,
+		sub:  sub,
+		quit: make(chan struct{}),
+		done: make(chan struct{}),
+	}
+	go p.loop()
+	return p
+}
+
+// A pipe is an [event.Subscription] that converts from a [types.Block] channel
+// to a [core.ChainHeadEvent] one.
+type pipe struct {
+	in         <-chan *types.Block
+	out        chan<- core.ChainHeadEvent
+	sub        event.Subscription
+	quit, done chan struct{}
+}
+
+func (p *pipe) loop() {
+	defer close(p.done)
+	for {
+		select {
+		case b := <-p.in:
+			select {
+			case p.out <- core.ChainHeadEvent{Block: b}:
+
+			case <-p.quit:
+				return
+			}
+		case <-p.quit:
+			return
+		}
+	}
+}
+
+func (p *pipe) Err() <-chan error {
+	return p.sub.Err()
+}
+
+func (p *pipe) Unsubscribe() {
+	p.sub.Unsubscribe()
+	close(p.quit)
+	<-p.done
+}

txgossip/priority.go

@@ -0,0 +1,54 @@
+// Copyright (C) 2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package txgossip
+
+import (
+	"slices"
+
+	"github.com/ava-labs/libevm/common"
+	"github.com/ava-labs/libevm/core/txpool"
+)
+
+// A LazyTransaction couples a [txpool.LazyTransaction] with its sender.
+type LazyTransaction struct {
+	*txpool.LazyTransaction
+	Sender common.Address
+}
+
+// TransactionsByPriority calls [txpool.TxPool.Pending] with the given filter,
+// collapses the results into a slice, and sorts said slice by decreasing gas
+// tip then chronologically. Transactions from the same sender are merely sorted
+// by increasing nonce.
+func (s *Set) TransactionsByPriority(filter txpool.PendingFilter) []*LazyTransaction {
+	pending := s.Pool.Pending(filter)
+	var n int
+	for _, txs := range pending {
+		n += len(txs)
+	}
+
+	all := make([]*LazyTransaction, n)
+	var i int
+	for from, txs := range pending {
+		for _, tx := range txs {
+			all[i] = &LazyTransaction{
+				LazyTransaction: tx,
+				Sender:          from,
+			}
+			i++
+		}
+	}
+
+	slices.SortStableFunc(all, func(a, b *LazyTransaction) int {
+		if a.Sender == b.Sender {
+			// [txpool.TxPool.Pending] already returns each slice in nonce order
+			// and we're performing a stable sort.
+			return 0
+		}
+		if tip := a.GasTipCap.Cmp(b.GasTipCap); tip != 0 {
+			return -tip // Higher tips first
+		}
+		return a.Time.Compare(b.Time)
+	})
+	return all
+}