l2geth: Sync from Backend Queue

l2geth/cmd/geth/main.go

@@ -164,6 +164,10 @@ var (
 		utils.RollupFeeThresholdDownFlag,
 		utils.RollupFeeThresholdUpFlag,
 		utils.SequencerClientHttpFlag,
+		utils.TxPublisherEnableFlag,
+		utils.TxPublisherProjectIDFlag,
+		utils.TxPublisherTopicIDFlag,
+		utils.TxPublisherTimeoutFlag,
 	}
 
 	rpcFlags = []cli.Flag{

l2geth/cmd/geth/usage.go

@@ -78,6 +78,10 @@ var AppHelpFlagGroups = []flagGroup{
 			utils.RollupFeeThresholdDownFlag,
 			utils.RollupFeeThresholdUpFlag,
 			utils.SequencerClientHttpFlag,
+			utils.TxPublisherEnableFlag,
+			utils.TxPublisherProjectIDFlag,
+			utils.TxPublisherTopicIDFlag,
+			utils.TxPublisherTimeoutFlag,
 		},
 	},
 	{

l2geth/cmd/utils/flags.go

@@ -61,6 +61,7 @@ import (
 	"github.com/ethereum-optimism/optimism/l2geth/p2p/netutil"
 	"github.com/ethereum-optimism/optimism/l2geth/params"
 	"github.com/ethereum-optimism/optimism/l2geth/rollup"
+	"github.com/ethereum-optimism/optimism/l2geth/rollup/pub"
 	"github.com/ethereum-optimism/optimism/l2geth/rpc"
 	whisper "github.com/ethereum-optimism/optimism/l2geth/whisper/whisperv6"
 	pcsclite "github.com/gballet/go-libpcsclite"
@@ -831,7 +832,7 @@ var (
 	}
 	RollupBackendFlag = cli.StringFlag{
 		Name:   "rollup.backend",
-		Usage:  "Sync backend for verifiers (\"l1\" or \"l2\"), defaults to l1",
+		Usage:  "Sync backend for verifiers (\"l1\", \"l2\" or \"queue\"), defaults to l1",
 		Value:  "l1",
 		EnvVar: "ROLLUP_BACKEND",
 	}
@@ -866,6 +867,51 @@ var (
 		Usage:  "HTTP endpoint for the sequencer client",
 		EnvVar: "SEQUENCER_CLIENT_HTTP",
 	}
+	TxPublisherEnableFlag = cli.BoolFlag{
+		Name:   "txpublisher.enable",
+		Usage:  "Enable transaction logging to PubSub",
+		EnvVar: "TX_PUBLISHER_ENABLE",
+	}
+	TxPublisherProjectIDFlag = cli.StringFlag{
+		Name:   "txpublisher.projectid",
+		Usage:  "GCP project ID for the tx PubSub",
+		EnvVar: "TX_PUBLISHER_PROJECT_ID",
+	}
+	TxPublisherTopicIDFlag = cli.StringFlag{
+		Name:   "txpublisher.topicid",
+		Usage:  "Topic ID used for PubSub",
+		EnvVar: "TX_PUBLISHER_TOPIC_ID",
+	}
+	TxPublisherTimeoutFlag = cli.DurationFlag{
+		Name:   "txpublisher.timeout",
+		Usage:  "Transaction publishing timeout",
+		EnvVar: "TX_PUBLISHER_TIMEOUT",
+	}
+	TxQueueEnableFlag = cli.BoolFlag{
+		Name:   "txqueue.enable",
+		Usage:  "Enable transaction syncing from the Backend Queue",
+		EnvVar: "TX_QUEUE_ENABLE",
+	}
+	TxQueueProjectIDFlag = cli.StringFlag{
+		Name:   "txqueue.projectid",
+		Usage:  "Backend Queue project ID",
+		EnvVar: "TX_QUEUE_PROJECT_ID",
+	}
+	TxQueueSubscriptionIDFlag = cli.StringFlag{
+		Name:   "txqueue.subscriptionid",
+		Usage:  "Transaction Queue subscription ID",
+		EnvVar: "TX_QUEUE_SUBSCRIPTION_ID",
+	}
+	TxQueueMaxOutstandingMessagesFlag = cli.IntFlag{
+		Name:   "txqueue.maxoutstandingmessages",
+		Usage:  "Max number of messages buffered in the transaction queue subscriber",
+		EnvVar: "TX_QUEUE_MAX_OUTSTANDING_MESSAGES",
+	}
+	TxQueueMaxOutstandingBytesFlag = cli.IntFlag{
+		Name:   "txqueue.maxoutstandingbytes",
+		Usage:  "Max outstanding bytes bufferered in the transaction queue subscriber",
+		EnvVar: "TX_QUEUE_MAX_OUTSTANDING_BYTES",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1147,6 +1193,43 @@ func setRollup(ctx *cli.Context, cfg *rollup.Config) {
 	}
 }
 
+// UsingOVM
+// setTxPublisher configures the transaction logger
+func setTxPublisher(ctx *cli.Context, cfg *pub.Config) {
+	if ctx.GlobalIsSet(TxPublisherEnableFlag.Name) {
+		cfg.Enable = ctx.GlobalBool(TxPublisherEnableFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxPublisherProjectIDFlag.Name) {
+		cfg.ProjectID = ctx.GlobalString(TxPublisherProjectIDFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxPublisherTopicIDFlag.Name) {
+		cfg.TopicID = ctx.GlobalString(TxPublisherTopicIDFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxPublisherTimeoutFlag.Name) {
+		cfg.Timeout = ctx.GlobalDuration(TxPublisherTimeoutFlag.Name)
+	}
+}
+
+// UsingOVM
+// setTxQueueSubscriber configures the Queue Backend
+func setTxQueueSubscriber(ctx *cli.Context, cfg *rollup.QueueSubscriberConfig) {
+	if ctx.GlobalIsSet(TxPublisherEnableFlag.Name) {
+		cfg.Enable = ctx.GlobalBool(TxQueueEnableFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxPublisherProjectIDFlag.Name) {
+		cfg.ProjectID = ctx.GlobalString(TxQueueProjectIDFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxQueueSubscriptionIDFlag.Name) {
+		cfg.SubscriptionID = ctx.GlobalString(TxQueueSubscriptionIDFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxQueueMaxOutstandingMessagesFlag.Name) {
+		cfg.MaxOutstandingMessages = ctx.GlobalInt(TxQueueMaxOutstandingMessagesFlag.Name)
+	}
+	if ctx.GlobalIsSet(TxQueueMaxOutstandingBytesFlag.Name) {
+		cfg.MaxOutstandingBytes = ctx.GlobalInt(TxQueueMaxOutstandingBytesFlag.Name)
+	}
+}
+
 // setLes configures the les server and ultra light client settings from the command line flags.
 func setLes(ctx *cli.Context, cfg *eth.Config) {
 	if ctx.GlobalIsSet(LightLegacyServFlag.Name) {
@@ -1606,6 +1689,8 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
 	setLes(ctx, cfg)
 	setEth1(ctx, &cfg.Rollup)
 	setRollup(ctx, &cfg.Rollup)
+	setTxPublisher(ctx, &cfg.TxPublisher)
+	setTxQueueSubscriber(ctx, &cfg.TxQueueSubscriber)
 
 	if ctx.GlobalIsSet(SyncModeFlag.Name) {
 		cfg.SyncMode = *GlobalTextMarshaler(ctx, SyncModeFlag.Name).(*downloader.SyncMode)

l2geth/core/types/transaction_meta.go

@@ -32,6 +32,17 @@ func (q QueueOrigin) String() string {
 	}
 }
 
+func (q QueueOrigin) MarshalJSON() ([]byte, error) {
+	switch q {
+	case QueueOriginSequencer:
+		return []byte("\"sequencer\""), nil
+	case QueueOriginL1ToL2:
+		return []byte("\"l1\""), nil
+	default:
+		return []byte("\"\""), nil
+	}
+}
+
 func (q *QueueOrigin) UnmarshalJSON(b []byte) error {
 	switch string(b) {
 	case "\"sequencer\"":

l2geth/eth/backend.go

@@ -52,6 +52,7 @@ import (
 	"github.com/ethereum-optimism/optimism/l2geth/params"
 	"github.com/ethereum-optimism/optimism/l2geth/rlp"
 	"github.com/ethereum-optimism/optimism/l2geth/rollup"
+	"github.com/ethereum-optimism/optimism/l2geth/rollup/pub"
 	"github.com/ethereum-optimism/optimism/l2geth/rollup/rcfg"
 	"github.com/ethereum-optimism/optimism/l2geth/rpc"
 )
@@ -206,7 +207,19 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	}
 	eth.txPool = core.NewTxPool(config.TxPool, chainConfig, eth.blockchain)
 
-	eth.syncService, err = rollup.NewSyncService(context.Background(), config.Rollup, eth.txPool, eth.blockchain, eth.chainDb)
+	var txLogger pub.Publisher
+	txLogger, err = pub.NewGooglePublisher(context.Background(), config.TxPublisher)
+	if err != nil {
+		return nil, err
+	}
+
+	var txQueueSubscriber rollup.QueueSubscriber
+	txQueueSubscriber, err = rollup.NewQueueSubscriber(context.Background(), config.TxQueueSubscriber)
+	if err != nil {
+		return nil, err
+	}
+
+	eth.syncService, err = rollup.NewSyncService(context.Background(), config.Rollup, eth.txPool, eth.blockchain, eth.chainDb, txLogger, txQueueSubscriber)
 	if err != nil {
 		return nil, fmt.Errorf("Cannot initialize syncservice: %w", err)
 	}

l2geth/eth/config.go

@@ -32,6 +32,7 @@ import (
 	"github.com/ethereum-optimism/optimism/l2geth/miner"
 	"github.com/ethereum-optimism/optimism/l2geth/params"
 	"github.com/ethereum-optimism/optimism/l2geth/rollup"
+	"github.com/ethereum-optimism/optimism/l2geth/rollup/pub"
 )
 
 // DefaultConfig contains default settings for use on the Ethereum main net.
@@ -183,4 +184,8 @@ type Config struct {
 
 	// Optimism Rollup Config
 	Rollup rollup.Config
+
+	TxPublisher pub.Config
+
+	TxQueueSubscriber rollup.QueueSubscriberConfig
 }

l2geth/go.mod

@@ -3,6 +3,7 @@ module github.com/ethereum-optimism/optimism/l2geth
 go 1.15
 
 require (
+	cloud.google.com/go/pubsub v1.18.0
 	github.com/Azure/azure-storage-blob-go v0.7.0
 	github.com/VictoriaMetrics/fastcache v1.6.0
 	github.com/aristanetworks/goarista v0.0.0-20170210015632-ea17b1a17847
@@ -21,7 +22,7 @@ require (
 	github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff
 	github.com/go-resty/resty/v2 v2.4.0
 	github.com/go-stack/stack v1.8.0
-	github.com/golang/protobuf v1.4.3
+	github.com/golang/protobuf v1.5.2
 	github.com/golang/snappy v0.0.4
 	github.com/gorilla/websocket v1.4.2
 	github.com/graph-gophers/graphql-go v0.0.0-20201113091052-beb923fada29
@@ -51,9 +52,9 @@ require (
 	github.com/wsddn/go-ecdh v0.0.0-20161211032359-48726bab9208
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912
+	golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27
 	golang.org/x/text v0.3.6
-	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba
+	golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce
 	gopkg.in/olebedev/go-duktape.v3 v3.0.0-20200619000410-60c24ae608a6
 	gopkg.in/sourcemap.v1 v1.0.5 // indirect