Merge pull request #4 from astriaorg/bharath/implement-optimistic-fork-event

feat:  Trusted auctioneer: Add an event for optimistic head creation

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

@@ -220,15 +220,16 @@ 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.
@@ -644,13 +645,16 @@ func (bc *BlockChain) SetSafe(header *types.Header) {
 }
 
 // SetOptimistic sets the optimistic block.
-func (bc *BlockChain) SetOptimistic(header *types.Header) {
+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.

core/blockchain_reader.go

@@ -445,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

@@ -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 {