feat: Trusted auctioneer: use optimistic head event in txpool's maintanance loop

.github/workflows/astria-build-and-publish-image.yml

@@ -72,4 +72,4 @@ jobs:
           push: true
           tags: ${{ steps.metadata.outputs.tags }}
           labels: ${{ steps.metadata.outputs.labels }}
-          project: w2d6w0spqz
+          project: w2d6w0spqz

cmd/devp2p/internal/ethtest/suite_test.go

@@ -61,6 +61,7 @@ func TestEthSuite(t *testing.T) {
 	if err != nil {
 		t.Fatalf("could not create new test suite: %v", err)
 	}
+
 	for _, test := range suite.EthTests() {
 		t.Run(test.Name, func(t *testing.T) {
 			if test.Slow && testing.Short() {
@@ -149,5 +150,6 @@ func setupGeth(stack *node.Node, dir string) error {
 		return fmt.Errorf("failed to register catalyst service: %v", err)
 	}
 	_, err = backend.BlockChain().InsertChain(chain.blocks[1:])
+	backend.BlockChain().SetOptimistic(chain.blocks[len(chain.blocks)-1])
 	return err
 }

cmd/geth/config.go

@@ -21,6 +21,8 @@ import (
 	"errors"
 	"fmt"
 	"github.com/ethereum/go-ethereum/eth/catalyst"
+	"github.com/ethereum/go-ethereum/grpc/optimistic"
+	"github.com/ethereum/go-ethereum/grpc/shared"
 	"os"
 	"reflect"
 	"runtime"
@@ -206,11 +208,16 @@ func makeFullNode(ctx *cli.Context) *node.Node {
 
 	// Configure gRPC if requested.
 	if ctx.IsSet(utils.GRPCEnabledFlag.Name) {
-		serviceV1, err := execution.NewExecutionServiceServerV1(eth)
+		sharedService, err := shared.NewSharedServiceContainer(eth)
 		if err != nil {
-			utils.Fatalf("failed to create execution service: %v", err)
+			utils.Fatalf("failed to create shared service container: %v", err)
 		}
-		utils.RegisterGRPCExecutionService(stack, serviceV1, &cfg.Node)
+
+		serviceV1a2 := execution.NewExecutionServiceServerV1(sharedService)
+
+		optimisticServiceV1a1 := optimistic.NewOptimisticServiceV1Alpha(sharedService)
+
+		utils.RegisterGRPCServices(stack, serviceV1a2, optimisticServiceV1a1, optimisticServiceV1a1, &cfg.Node)
 	}
 
 	// Add the Ethereum Stats daemon if requested.

cmd/geth/main.go

@@ -123,6 +123,7 @@ var (
 		utils.MinerRecommitIntervalFlag,
 		utils.MinerPendingFeeRecipientFlag,
 		utils.MinerNewPayloadTimeoutFlag, // deprecated
+		utils.AuctioneerEnabledFlag,
 		utils.NATFlag,
 		utils.NoDiscoverFlag,
 		utils.DiscoveryV4Flag,

cmd/utils/flags.go

@@ -18,6 +18,7 @@
 package utils
 
 import (
+	optimisticGrpc "buf.build/gen/go/astria/execution-apis/grpc/go/astria/auction/v1alpha1/auctionv1alpha1grpc"
 	"context"
 	"crypto/ecdsa"
 	"encoding/hex"
@@ -769,6 +770,13 @@ var (
 		Category: flags.APICategory,
 	}
 
+	// auctioneer
+	AuctioneerEnabledFlag = &cli.BoolFlag{
+		Name:     "auctioneer",
+		Usage:    "Enable the auctioneer server",
+		Category: flags.MinerCategory,
+	}
+
 	// Network Settings
 	MaxPeersFlag = &cli.IntFlag{
 		Name:     "maxpeers",
@@ -1438,6 +1446,12 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	SetDataDir(ctx, cfg)
 	setSmartCard(ctx, cfg)
 
+	if ctx.Bool(AuctioneerEnabledFlag.Name) {
+		cfg.EnableAuctioneer = true
+	} else {
+		cfg.EnableAuctioneer = false
+	}
+
 	if ctx.IsSet(JWTSecretFlag.Name) {
 		cfg.JWTSecret = ctx.String(JWTSecretFlag.Name)
 	}
@@ -1987,10 +2001,10 @@ func RegisterGraphQLService(stack *node.Node, backend ethapi.Backend, filterSyst
 	}
 }
 
-// RegisterGRPCExecutionService adds the gRPC API to the node.
+// RegisterGRPCServices adds the gRPC API to the node.
 // It was done this way so that our grpc execution server can access the ethapi.Backend
-func RegisterGRPCExecutionService(stack *node.Node, execServ astriaGrpc.ExecutionServiceServer, cfg *node.Config) {
-	if err := node.NewGRPCServerHandler(stack, execServ, cfg); err != nil {
+func RegisterGRPCServices(stack *node.Node, execServ astriaGrpc.ExecutionServiceServer, optimisticExecutionServ optimisticGrpc.OptimisticExecutionServiceServer, auctionServiceServer optimisticGrpc.AuctionServiceServer, cfg *node.Config) {
+	if err := node.NewGRPCServerHandler(stack, execServ, optimisticExecutionServ, auctionServiceServer, cfg); err != nil {
 		Fatalf("Failed to register the gRPC service: %v", err)
 	}
 }

core/events.go

@@ -24,6 +24,12 @@ import (
 // NewTxsEvent is posted when a batch of transactions enter the transaction pool.
 type NewTxsEvent struct{ Txs []*types.Transaction }
 
+// NewMempoolClearedEvent is posted when the mempool is cleared after a head reset for trusted auctioneer
+type NewMempoolCleared struct {
+	// the new head to which the mempool state was reset to before clearing the mempool
+	NewHead *types.Header
+}
+
 // NewMinedBlockEvent is posted when a block has been imported.
 type NewMinedBlockEvent struct{ Block *types.Block }
 

core/txpool/blobpool/blobpool.go

@@ -1602,6 +1602,10 @@ func (p *BlobPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool
 	}
 }
 
+func (p *BlobPool) SubscribeMempoolClearance(ch chan<- core.NewMempoolCleared) event.Subscription {
+	return nil
+}
+
 // Nonce returns the next nonce of an account, with all transactions executable
 // by the pool already applied on top.
 func (p *BlobPool) Nonce(addr common.Address) uint64 {

core/txpool/legacypool/legacypool.go

@@ -205,13 +205,14 @@ func (config *Config) sanitize() Config {
 // current state) and future transactions. Transactions move between those
 // two states over time as they are received and processed.
 type LegacyPool struct {
-	config      Config
-	chainconfig *params.ChainConfig
-	chain       BlockChain
-	gasTip      atomic.Pointer[uint256.Int]
-	txFeed      event.Feed
-	signer      types.Signer
-	mu          sync.RWMutex
+	config           Config
+	chainconfig      *params.ChainConfig
+	chain            BlockChain
+	gasTip           atomic.Pointer[uint256.Int]
+	txFeed           event.Feed
+	mempoolClearFeed event.Feed
+	signer           types.Signer
+	mu               sync.RWMutex
 
 	astria *astriaOrdered
 
@@ -238,6 +239,8 @@ type LegacyPool struct {
 	initDoneCh      chan struct{}  // is closed once the pool is initialized (for tests)
 
 	changesSinceReorg int // A counter for how many drops we've performed in-between reorg.
+
+	auctioneerEnabled bool
 }
 
 type txpoolResetRequest struct {
@@ -246,26 +249,27 @@ type txpoolResetRequest struct {
 
 // New creates a new transaction pool to gather, sort and filter inbound
 // transactions from the network.
-func New(config Config, chain BlockChain) *LegacyPool {
+func New(config Config, chain BlockChain, auctioneerEnabled bool) *LegacyPool {
 	// Sanitize the input to ensure no vulnerable gas prices are set
 	config = (&config).sanitize()
 
 	// Create the transaction pool with its initial settings
 	pool := &LegacyPool{
-		config:          config,
-		chain:           chain,
-		chainconfig:     chain.Config(),
-		signer:          types.LatestSigner(chain.Config()),
-		pending:         make(map[common.Address]*list),
-		queue:           make(map[common.Address]*list),
-		beats:           make(map[common.Address]time.Time),
-		all:             newLookup(),
-		reqResetCh:      make(chan *txpoolResetRequest),
-		reqPromoteCh:    make(chan *accountSet),
-		queueTxEventCh:  make(chan *types.Transaction),
-		reorgDoneCh:     make(chan chan struct{}),
-		reorgShutdownCh: make(chan struct{}),
-		initDoneCh:      make(chan struct{}),
+		config:            config,
+		chain:             chain,
+		chainconfig:       chain.Config(),
+		signer:            types.LatestSigner(chain.Config()),
+		pending:           make(map[common.Address]*list),
+		queue:             make(map[common.Address]*list),
+		beats:             make(map[common.Address]time.Time),
+		all:               newLookup(),
+		reqResetCh:        make(chan *txpoolResetRequest),
+		reqPromoteCh:      make(chan *accountSet),
+		queueTxEventCh:    make(chan *types.Transaction),
+		reorgDoneCh:       make(chan chan struct{}),
+		reorgShutdownCh:   make(chan struct{}),
+		initDoneCh:        make(chan struct{}),
+		auctioneerEnabled: auctioneerEnabled,
 	}
 	pool.locals = newAccountSet(pool.signer)
 	for _, addr := range config.Locals {
@@ -521,6 +525,12 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs
 	return pool.txFeed.Subscribe(ch)
 }
 
+// SubscribeTransactions registers a subscription for the event which is triggered
+// when the mempool is cleared after a reset
+func (pool *LegacyPool) SubscribeMempoolClearance(ch chan<- core.NewMempoolCleared) event.Subscription {
+	return pool.mempoolClearFeed.Subscribe(ch)
+}
+
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
 // new transaction, and drops all transactions below this threshold.
 func (pool *LegacyPool) SetGasTip(tip *big.Int) {
@@ -1366,8 +1376,16 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 	}
 	pool.mu.Lock()
 	if reset != nil {
-		// Reset from the old head to the new, rescheduling any reorged transactions
-		pool.reset(reset.oldHead, reset.newHead)
+		// only reset the state root and the head of the txpool when we are running the auctioneer node.
+		// when we are not running the auctioneer node, we re-inject any re-orged transactions which is similar
+		// to the current functionality of geth
+		if pool.auctioneerEnabled {
+			// only reset from the old head to the new head
+			pool.resetHeadOnly(reset.oldHead, reset.newHead)
+		} else {
+			// Reset from the old head to the new, rescheduling any reorged transactions
+			pool.reset(reset.oldHead, reset.newHead)
+		}
 
 		// Nonces were reset, discard any events that became stale
 		for addr := range events {
@@ -1376,7 +1394,6 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 				delete(events, addr)
 			}
 		}
-		// Reset needs promote for all addresses
 		promoteAddrs = make([]common.Address, 0, len(pool.queue))
 		for addr := range pool.queue {
 			promoteAddrs = append(promoteAddrs, addr)
@@ -1389,7 +1406,13 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 	// remove any transaction that has been included in the block or was invalidated
 	// because of another transaction (e.g. higher gas price).
 	if reset != nil {
-		pool.demoteUnexecutables()
+		if pool.auctioneerEnabled {
+			// if we are running the pool as an auctioneer, then we should clear the mempool each time the head
+			// is reset
+			pool.clearPendingAndQueued(reset.newHead)
+		} else {
+			pool.demoteUnexecutables()
+		}
 		if reset.newHead != nil {
 			if pool.chainconfig.IsLondon(new(big.Int).Add(reset.newHead.Number, big.NewInt(1))) {
 				pendingBaseFee := eip1559.CalcBaseFee(pool.chainconfig, reset.newHead)
@@ -1414,6 +1437,10 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 	pool.changesSinceReorg = 0 // Reset change counter
 	pool.mu.Unlock()
 
+	if reset != nil {
+		pool.mempoolClearFeed.Send(core.NewMempoolCleared{NewHead: reset.newHead})
+	}
+
 	// Notify subsystems for newly added transactions
 	for _, tx := range promoted {
 		addr, _ := types.Sender(pool.signer, tx)
@@ -1511,6 +1538,7 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) {
 			}
 		}
 	}
+
 	// Initialize the internal state to the current head
 	if newHead == nil {
 		newHead = pool.chain.CurrentBlock() // Special case during testing
@@ -1524,12 +1552,30 @@ func (pool *LegacyPool) reset(oldHead, newHead *types.Header) {
 	pool.currentState = statedb
 	pool.pendingNonces = newNoncer(statedb)
 
+	// we don't care about these
 	// Inject any transactions discarded due to reorgs
 	log.Debug("Reinjecting stale transactions", "count", len(reinject))
 	core.SenderCacher.Recover(pool.signer, reinject)
 	pool.addTxsLocked(reinject, false)
 }
 
+// reset retrieves the current state of the blockchain and ensures the content
+// of the transaction pool is valid with regard to the chain state.
+func (pool *LegacyPool) resetHeadOnly(oldHead, newHead *types.Header) {
+	// Initialize the internal state to the current head
+	if newHead == nil {
+		newHead = pool.chain.CurrentBlock() // Special case during testing
+	}
+	statedb, err := pool.chain.StateAt(newHead.Root)
+	if err != nil {
+		log.Error("Failed to reset txpool state", "err", err)
+		return
+	}
+	pool.currentHead.Store(newHead)
+	pool.currentState = statedb
+	pool.pendingNonces = newNoncer(statedb)
+}
+
 // promoteExecutables moves transactions that have become processable from the
 // future queue to the set of pending transactions. During this process, all
 // invalidated transactions (low nonce, low balance) are deleted.
@@ -1732,6 +1778,55 @@ func (pool *LegacyPool) truncateQueue() {
 	}
 }
 
+// clearPendingAndQueued removes invalid and processed transactions from the pools
+// it assumes that the pool lock is being held
+func (pool *LegacyPool) clearPendingAndQueued(newHead *types.Header) {
+	// Iterate over all accounts and demote any non-executable transactions
+	addrsForWhichTxsRemoved := map[common.Address]bool{}
+
+	for addr, list := range pool.pending {
+		dropped, invalids := list.ClearList()
+
+		pendingGauge.Dec(int64(dropped.Len() + invalids.Len()))
+
+		for _, tx := range dropped {
+			pool.all.Remove(tx.Hash())
+		}
+		for _, tx := range invalids {
+			pool.all.Remove(tx.Hash())
+		}
+
+		if list.Empty() {
+			delete(pool.pending, addr)
+			delete(pool.beats, addr)
+
+			addrsForWhichTxsRemoved[addr] = true
+		}
+	}
+
+	for addr, list := range pool.queue {
+		dropped, invalids := list.ClearList()
+		queuedGauge.Dec(int64(dropped.Len() + invalids.Len()))
+
+		for _, tx := range dropped {
+			pool.all.Remove(tx.Hash())
+		}
+		for _, tx := range invalids {
+			pool.all.Remove(tx.Hash())
+		}
+
+		if list.Empty() {
+			delete(pool.queue, addr)
+
+			addrsForWhichTxsRemoved[addr] = true
+		}
+	}
+
+	for addr := range addrsForWhichTxsRemoved {
+		pool.reserve(addr, false)
+	}
+}
+
 // demoteUnexecutables removes invalid and processed transactions from the pools
 // executable/pending queue and any subsequent transactions that become unexecutable
 // are moved back into the future queue.
@@ -1742,6 +1837,7 @@ func (pool *LegacyPool) truncateQueue() {
 func (pool *LegacyPool) demoteUnexecutables() {
 	// Iterate over all accounts and demote any non-executable transactions
 	gasLimit := pool.currentHead.Load().GasLimit
+
 	for addr, list := range pool.pending {
 		nonce := pool.currentState.GetNonce(addr)