refactor: Trusted Auctioneer: Separate execution api service and optimistic execution api service

.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/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/txpool/legacypool/legacypool.go

@@ -239,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 {
@@ -247,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 {
@@ -1373,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 {
@@ -1395,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.clearPendingAndQueued(reset.newHead)
+		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)
@@ -1765,9 +1782,12 @@ func (pool *LegacyPool) truncateQueue() {
 // 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(len(dropped) + len(invalids)))
+
+		pendingGauge.Dec(int64(dropped.Len() + invalids.Len()))
 
 		for _, tx := range dropped {
 			pool.all.Remove(tx.Hash())
@@ -1779,12 +1799,14 @@ func (pool *LegacyPool) clearPendingAndQueued(newHead *types.Header) {
 		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(len(dropped) + len(invalids)))
+		queuedGauge.Dec(int64(dropped.Len() + invalids.Len()))
 
 		for _, tx := range dropped {
 			pool.all.Remove(tx.Hash())
@@ -1794,12 +1816,15 @@ func (pool *LegacyPool) clearPendingAndQueued(newHead *types.Header) {
 		}
 
 		if list.Empty() {
-			if _, ok := pool.queue[addr]; !ok {
-				pool.reserve(addr, false)
-			}
 			delete(pool.queue, addr)
+
+			addrsForWhichTxsRemoved[addr] = true
 		}
 	}
+
+	for addr := range addrsForWhichTxsRemoved {
+		pool.reserve(addr, false)
+	}
 }
 
 // demoteUnexecutables removes invalid and processed transactions from the pools

core/txpool/legacypool/legacypool2_test.go

@@ -85,7 +85,7 @@ func TestTransactionFutureAttack(t *testing.T) {
 	config := testTxPoolConfig
 	config.GlobalQueue = 100
 	config.GlobalSlots = 100
-	pool := New(config, blockchain)
+	pool := New(config, blockchain, true)
 	pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
 	defer pool.Close()
 	fillPool(t, pool)
@@ -119,7 +119,7 @@ func TestTransactionFuture1559(t *testing.T) {
 	// Create the pool to test the pricing enforcement with
 	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil)
 	blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
-	pool := New(testTxPoolConfig, blockchain)
+	pool := New(testTxPoolConfig, blockchain, true)
 	pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
 	defer pool.Close()
 
@@ -152,7 +152,7 @@ func TestTransactionZAttack(t *testing.T) {
 	// Create the pool to test the pricing enforcement with
 	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil)
 	blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
-	pool := New(testTxPoolConfig, blockchain)
+	pool := New(testTxPoolConfig, blockchain, true)
 	pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
 	defer pool.Close()
 	// Create a number of test accounts, fund them and make transactions
@@ -223,7 +223,7 @@ func BenchmarkFutureAttack(b *testing.B) {
 	config := testTxPoolConfig
 	config.GlobalQueue = 100
 	config.GlobalSlots = 100
-	pool := New(config, blockchain)
+	pool := New(config, blockchain, true)
 	pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
 	defer pool.Close()
 	fillPool(b, pool)