Merge pull request #3 from astriaorg/bharath/implement-optimistic-fork-creation

feat: Trusted auctioneer: Maintain a separate optimistic block fork

Author: bharath-123

.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/blockchain.go

@@ -54,11 +54,12 @@ import (
 )
 
 var (
-	headBlockGauge          = metrics.NewRegisteredGauge("chain/head/block", nil)
-	headHeaderGauge         = metrics.NewRegisteredGauge("chain/head/header", nil)
-	headFastBlockGauge      = metrics.NewRegisteredGauge("chain/head/receipt", nil)
-	headFinalizedBlockGauge = metrics.NewRegisteredGauge("chain/head/finalized", nil)
-	headSafeBlockGauge      = metrics.NewRegisteredGauge("chain/head/safe", nil)
+	headBlockGauge           = metrics.NewRegisteredGauge("chain/head/block", nil)
+	headHeaderGauge          = metrics.NewRegisteredGauge("chain/head/header", nil)
+	headFastBlockGauge       = metrics.NewRegisteredGauge("chain/head/receipt", nil)
+	headFinalizedBlockGauge  = metrics.NewRegisteredGauge("chain/head/finalized", nil)
+	headSafeBlockGauge       = metrics.NewRegisteredGauge("chain/head/safe", nil)
+	headOptimisticBlockGauge = metrics.NewRegisteredGauge("chain/head/optimistic", nil)
 
 	chainInfoGauge = metrics.NewRegisteredGaugeInfo("chain/info", nil)
 
@@ -219,24 +220,26 @@ type BlockChain struct {
 	stateCache    state.Database                   // State database to reuse between imports (contains state cache)
 	txIndexer     *txIndexer                       // Transaction indexer, might be nil if not enabled
 
-	hc            *HeaderChain
-	rmLogsFeed    event.Feed
-	chainFeed     event.Feed
-	chainSideFeed event.Feed
-	chainHeadFeed event.Feed
-	logsFeed      event.Feed
-	blockProcFeed event.Feed
-	scope         event.SubscriptionScope
-	genesisBlock  *types.Block
+	hc                      *HeaderChain
+	rmLogsFeed              event.Feed
+	chainFeed               event.Feed
+	chainSideFeed           event.Feed
+	chainHeadFeed           event.Feed
+	chainOptimisticHeadFeed event.Feed
+	logsFeed                event.Feed
+	blockProcFeed           event.Feed
+	scope                   event.SubscriptionScope
+	genesisBlock            *types.Block
 
 	// This mutex synchronizes chain write operations.
 	// Readers don't need to take it, they can just read the database.
 	chainmu *syncx.ClosableMutex
 
-	currentBlock      atomic.Pointer[types.Header] // Current head of the chain
-	currentSnapBlock  atomic.Pointer[types.Header] // Current head of snap-sync
-	currentFinalBlock atomic.Pointer[types.Header] // Latest (consensus) finalized block
-	currentSafeBlock  atomic.Pointer[types.Header] // Latest (consensus) safe block
+	currentBlock           atomic.Pointer[types.Header] // Current head of the chain
+	currentSnapBlock       atomic.Pointer[types.Header] // Current head of snap-sync
+	currentFinalBlock      atomic.Pointer[types.Header] // Latest (consensus) finalized block
+	currentSafeBlock       atomic.Pointer[types.Header] // Latest (consensus) safe block
+	currentOptimisticBlock atomic.Pointer[types.Header] // Latest optimistic block
 
 	currentBaseCelestiaHeight atomic.Uint64 // Latest finalized block height on Celestia
 
@@ -325,6 +328,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 	bc.currentBlock.Store(bc.genesisBlock.Header())
 	bc.currentFinalBlock.Store(bc.genesisBlock.Header())
 	bc.currentSafeBlock.Store(bc.genesisBlock.Header())
+	bc.currentOptimisticBlock.Store(bc.genesisBlock.Header())
 	bc.currentBaseCelestiaHeight.Store(bc.Config().AstriaCelestiaInitialHeight)
 
 	// Update chain info data metrics
@@ -532,6 +536,7 @@ func (bc *BlockChain) loadLastState() error {
 			bc.currentFinalBlock.Store(block.Header())
 			headFinalizedBlockGauge.Update(int64(block.NumberU64()))
 			bc.currentSafeBlock.Store(block.Header())
+			bc.currentOptimisticBlock.Store(block.Header())
 			headSafeBlockGauge.Update(int64(block.NumberU64()))
 		}
 	}
@@ -639,6 +644,19 @@ func (bc *BlockChain) SetSafe(header *types.Header) {
 	}
 }
 
+// SetOptimistic sets the optimistic block.
+func (bc *BlockChain) SetOptimistic(block *types.Block) {
+	header := block.Header()
+	bc.currentOptimisticBlock.Store(header)
+	if header != nil {
+		headOptimisticBlockGauge.Update(int64(header.Number.Uint64()))
+	} else {
+		headOptimisticBlockGauge.Update(0)
+	}
+
+	bc.chainOptimisticHeadFeed.Send(ChainOptimisticHeadEvent{Block: block})
+}
+
 // rewindHashHead implements the logic of rewindHead in the context of hash scheme.
 func (bc *BlockChain) rewindHashHead(head *types.Header, root common.Hash) (*types.Header, uint64) {
 	var (

core/blockchain_reader.go

@@ -63,6 +63,12 @@ func (bc *BlockChain) CurrentSafeBlock() *types.Header {
 	return bc.currentSafeBlock.Load()
 }
 
+// CurrentOptimisticBlock retrieves the current optimistic block of the canonical
+// chain. The block is retrieved from the blockchain's internal cache.
+func (bc *BlockChain) CurrentOptimisticBlock() *types.Header {
+	return bc.currentOptimisticBlock.Load()
+}
+
 // CurrentBaseCelestiaHeight retrieves the current base celestia height of the
 // canonical chain. The height is retrieved from the blockchain's internal cache.
 func (bc *BlockChain) CurrentBaseCelestiaHeight() uint64 {
@@ -439,6 +445,11 @@ func (bc *BlockChain) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Su
 	return bc.scope.Track(bc.chainHeadFeed.Subscribe(ch))
 }
 
+// SubscribeChainOptimisticHeadEvent registers a subscription of ChainOptimisticHeadEvent.
+func (bc *BlockChain) SubscribeChainOptimisticHeadEvent(ch chan<- ChainOptimisticHeadEvent) event.Subscription {
+	return bc.scope.Track(bc.chainOptimisticHeadFeed.Subscribe(ch))
+}
+
 // SubscribeChainSideEvent registers a subscription of ChainSideEvent.
 func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Subscription {
 	return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))

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 }
 
@@ -41,3 +47,7 @@ type ChainSideEvent struct {
 }
 
 type ChainHeadEvent struct{ Block *types.Block }
+
+type ChainOptimisticHeadEvent struct {
+	Block *types.Block
+}

core/txpool/blobpool/blobpool.go

@@ -339,6 +339,7 @@ func (p *BlobPool) ClearAstriaOrdered()                             {}
 func (p *BlobPool) AddToAstriaExcludedFromBlock(*types.Transaction) {}
 func (p *BlobPool) AstriaExcludedFromBlock() *types.Transactions    { return &types.Transactions{} }
 func (p *BlobPool) AstriaOrdered() *types.Transactions              { return &types.Transactions{} }
+func (p *BlobPool) ValidateTx(tx *types.Transaction) error          { return nil }
 
 // Filter returns whether the given transaction can be consumed by the blob pool.
 func (p *BlobPool) Filter(tx *types.Transaction) bool {
@@ -1601,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 {